New observers work better with Dereferenceable.

* CopyOnNotifyObserverSet#notify_observers now accepts a block
* CopyOnNotifyObserverSet#notify_and_delete_observers now accepts a block
* CopyOnWriteObserverSet#notify_observers now accepts a block
* CopyOnWriteObserverSet#notify_and_delete_observers now accepts a block
* Agent observer notification uses blocks to better support Dereferenceable
* IVar observer notification uses blocks to better support Dereferenceable
* ScheduledTask observer notification uses blocks to better support Dereferenceable

Author: jdantonio

lib/concurrent/agent.rb

@@ -170,7 +170,8 @@ def work(&handler) # :nodoc:
             should_notify = true
           end
         end
-        @observers.notify_observers(Time.now, self.value) if should_notify
+        time = Time.now
+        @observers.notify_observers{ [time, self.value] } if should_notify
       rescue Exception => ex
         try_rescue(ex)
       end

lib/concurrent/copy_on_notify_observer_set.rb

@@ -41,9 +41,9 @@ def count_observers
     # Notifies all registered observers with optional args
     # @param [Object] args arguments to be passed to each observer
     # @return [CopyOnWriteObserverSet] self
-    def notify_observers(*args)
+    def notify_observers(*args, &block)
       observers = @mutex.synchronize { @observers.dup }
-      notify_to(observers, *args)
+      notify_to(observers, *args, &block)
 
       self
     end
@@ -52,28 +52,32 @@ def notify_observers(*args)
     #
     # @param [Object] args arguments to be passed to each observer
     # @return [CopyOnWriteObserverSet] self
-    def notify_and_delete_observers(*args)
+    def notify_and_delete_observers(*args, &block)
       observers = duplicate_and_clear_observers
-      notify_to(observers, *args)
+      notify_to(observers, *args, &block)
 
       self
     end
 
     private
 
-      def duplicate_and_clear_observers
-        @mutex.synchronize do
-          observers = @observers.dup
-          @observers.clear
-          observers
-        end
+    def duplicate_and_clear_observers
+      @mutex.synchronize do
+        observers = @observers.dup
+        @observers.clear
+        observers
       end
+    end
 
-      def notify_to(observers, *args)
-        observers.each do |observer, function|
-          observer.send function, *args
+    def notify_to(observers, *args)
+      raise ArgumentError.new('cannot give arguments and a block') if block_given? && ! args.empty?
+      observers.each do |observer, function|
+        if block_given?
+          observer.send(function, *[yield].flatten)
+        else
+          observer.send(function, *args)
         end
       end
-
+    end
   end
-end
+end

lib/concurrent/copy_on_write_observer_set.rb

@@ -50,44 +50,48 @@ def count_observers
     # Notifies all registered observers with optional args
     # @param [Object] args arguments to be passed to each observer
     # @return [CopyOnWriteObserverSet] self
-    def notify_observers(*args)
-      notify_to(observers, *args)
+    def notify_observers(*args, &block)
+      notify_to(observers, *args, &block)
       self
     end
 
     # Notifies all registered observers with optional args and deletes them.
     #
     # @param [Object] args arguments to be passed to each observer
     # @return [CopyOnWriteObserverSet] self
-    def notify_and_delete_observers(*args)
+    def notify_and_delete_observers(*args, &block)
       old = clear_observers_and_return_old
-      notify_to(old, *args)
+      notify_to(old, *args, &block)
       self
     end
 
     private
 
-      def notify_to(observers, *args)
-        observers.each do |observer, function|
-          observer.send function, *args
+    def notify_to(observers, *args)
+      raise ArgumentError.new('cannot give arguments and a block') if block_given? && ! args.empty?
+      observers.each do |observer, function|
+        if block_given?
+          observer.send(function, *[yield].flatten)
+        else
+          observer.send(function, *args)
         end
       end
+    end
 
-      def observers
-        @mutex.synchronize { @observers }
-      end
+    def observers
+      @mutex.synchronize { @observers }
+    end
 
-      def observers=(new_set)
-        @mutex.synchronize { @observers = new_set}
-      end
+    def observers=(new_set)
+      @mutex.synchronize { @observers = new_set}
+    end
 
-      def clear_observers_and_return_old
-        @mutex.synchronize do
-          old_observers = @observers
-          @observers = {}
-          old_observers
-        end
+    def clear_observers_and_return_old
+      @mutex.synchronize do
+        old_observers = @observers
+        @observers = {}
+        old_observers
       end
-
+    end
   end
-end
+end

lib/concurrent/ivar.rb

@@ -71,8 +71,8 @@ def complete(success, value, reason)
         event.set
       end
 
-      @observers.notify_and_delete_observers(Time.now, value, reason)
+      time = Time.now
+      @observers.notify_and_delete_observers{ [time, self.value, reason] }
     end
-
   end
 end

lib/concurrent/scheduled_task.rb

@@ -75,7 +75,8 @@ def work
           event.set
         end
 
-        @observers.notify_and_delete_observers(Time.now, self.value, reason)
+        time = Time.now
+        @observers.notify_and_delete_observers{ [time, self.value, reason] }
       end
 
     end

spec/concurrent/observer_set_shared.rb

@@ -84,6 +84,41 @@
         observer_set.notify_observers(4, 'a')
       end
     end
+
+    context 'with a block' do
+
+      before(:each) do
+        observer.stub(:update).with(any_args)
+        another_observer.stub(:update).with(any_args)
+      end
+
+      it 'calls the block once for every observer' do
+
+        counter = double('block call counter')
+        expect(counter).to receive(:called).with(no_args).exactly(2).times
+
+        observer_set.add_observer(observer)
+        observer_set.add_observer(another_observer)
+
+        observer_set.notify_observers{ counter.called }
+      end
+
+      it 'passes the block return value to the update method' do
+
+        expect(observer).to receive(:update).with(1, 2, 3, 4)
+        observer_set.add_observer(observer)
+        observer_set.notify_observers{ [1, 2, 3, 4] }
+      end
+
+      it 'raises an exception if given both arguments and a block' do
+
+        observer_set.add_observer(observer)
+
+        expect {
+          observer_set.notify_observers(1, 2, 3, 4){ nil }
+        }.to raise_error(ArgumentError)
+      end
+    end
   end
 
   context '#count_observers' do