async: review and refactor interaction between main thread and worker threads

lib/orocos/async/async.rb

@@ -30,7 +30,7 @@
 #       puts state
 #   end
 #   reader.read do |value|
-#       puts value 
+#       puts value
 #   end
 #
 # If a method call needs the remote Orocos Task which is currently not reachable
@@ -44,13 +44,13 @@
 # short on worker threads.
 #
 #   # these events are generated by polling
-#   task.on_connect do 
+#   task.on_connect do
 #       puts "connected"
 #   end
-#   task.on_disconnect do 
+#   task.on_disconnect do
 #       puts "disconnected"
 #   end
-#   task.on_reconnect do 
+#   task.on_reconnect do
 #       puts "reconnected"
 #   end
 #
@@ -65,12 +65,22 @@
 #   port.on_new_data do |data|
 #       puts data
 #   end
-# 
+#
 # The polling frequency can be changed by setting the period attribute of each
 # asynchronous object.
 #
 module Orocos::Async
-    KNOWN_ERRORS = [Orocos::ComError,Orocos::NotFound,Typelib::NotFound,Orocos::TypekitTypeNotFound,Orocos::TypekitTypeNotExported,Orocos::StateTransitionFailed,Orocos::ConnectionFailed,OroGen::DefinitionTypekitNotFound]
+    KNOWN_ERRORS = [
+        Orocos::ComError,
+        Orocos::NotFound,
+        Typelib::NotFound,
+        Orocos::TypekitTypeNotFound,
+        Orocos::TypekitTypeNotExported,
+        Orocos::StateTransitionFailed,
+        Orocos::ConnectionFailed,
+        OroGen::DefinitionTypekitNotFound
+    ]
+
     class << self
         extend ::Forwardable
 

lib/orocos/async/attributes.rb

@@ -17,25 +17,26 @@ def initialize(async_task,attribute,options=Hash.new)
             disable_emitting do
                 reachable!(attribute)
             end
-            @poll_timer = @event_loop.async_every(method(:raw_read), {:period => period, :start => false,
-                                                  :known_errors => Orocos::Async::KNOWN_ERRORS}) do |data,error|
+
+            @poll_timer = @event_loop.async_every(
+                proc { @delegator_obj.raw_read },
+                { period: period, start: false,
+                  known_errors: Orocos::Async::KNOWN_ERRORS }
+            ) do |data, error|
                 if error
                     @poll_timer.cancel
                     self.period = @poll_timer.period
-                    @event_loop.once do
-                        event :error,error
-                    end
-                else
-                    if data
-                        if @raw_last_sample != data
-                            @raw_last_sample = data
-                            event :raw_change,data
-                            event :change,Typelib.to_ruby(data)
-                        end
+                    event :error, error
+                elsif data
+                    if @raw_last_sample != data
+                        @raw_last_sample = data
+                        event :raw_change, data
+                        event :change, Typelib.to_ruby(data)
                     end
                 end
             end
             @poll_timer.doc = attribute.full_name
+
             @task.on_unreachable do
                 unreachable!
             end
@@ -92,14 +93,10 @@ def wait(timeout = 5.0)
         def really_add_listener(listener)
             super
             if listener.event == :raw_change
-                if !@poll_timer.running?
-                    @poll_timer.start(period)
-                end
+                @poll_timer.start(period) unless @poll_timer.running?
                 listener.call(@raw_last_sample) if @raw_last_sample && listener.use_last_value?
             elsif listener.event == :change
-                if !@poll_timer.running?
-                    @poll_timer.start(period)
-                end
+                @poll_timer.start(period) unless @poll_timer.running?
                 listener.call(Typelib.to_ruby(@raw_last_sample)) if @raw_last_sample && listener.use_last_value?
             end
         end