Merge pull request #63 from jdantonio/block_observer

Observable Shared Specs and Block Observer

Author: jdantonio

README.md

@@ -142,6 +142,7 @@ ibm.value #=> 187.57
 * [Michele Della Torre](https://github.com/mighe)
 * [Chris Seaton](https://github.com/chrisseaton)
 * [Lucas Allan](https://github.com/lucasallan)
+* [Ravil Bayramgalin](https://github.com/brainopia)
 * [Giuseppe Capizzi](https://github.com/gcapizzi)
 * [Brian Shirai](https://github.com/brixen)
 * [Chip Miller](https://github.com/chip-miller)

lib/concurrent/atomic/copy_on_notify_observer_set.rb

@@ -15,12 +15,23 @@ def initialize
     # @param [Object] observer the observer to add
     # @param [Symbol] func the function to call on the observer during notification. Default is :update
     # @return [Symbol] the added function
-    def add_observer(observer, func=:update)
+    def add_observer(observer=nil, func=:update, &block)
+      if observer.nil? && block.nil?
+        raise ArgumentError, 'should pass observer as a first argument or block'
+      elsif observer && block
+        raise ArgumentError.new('cannot provide both an observer and a block')
+      end
+
+      if block
+        observer = block
+        func = :call
+      end
+
       @mutex.lock
       @observers[observer] = func
       @mutex.unlock
 
-      func
+      observer
     end
 
     # @param [Object] observer the observer to remove

lib/concurrent/atomic/copy_on_write_observer_set.rb

@@ -14,14 +14,25 @@ def initialize
     # @param [Object] observer the observer to add
     # @param [Symbol] func the function to call on the observer during notification. Default is :update
     # @return [Symbol] the added function
-    def add_observer(observer, func=:update)
+    def add_observer(observer=nil, func=:update, &block)
+      if observer.nil? && block.nil?
+        raise ArgumentError, 'should pass observer as a first argument or block'
+      elsif observer && block
+        raise ArgumentError.new('cannot provide both an observer and a block')
+      end
+
+      if block
+        observer = block
+        func = :call
+      end
+
       @mutex.lock
       new_observers = @observers.dup
       new_observers[observer] = func
       @observers = new_observers
       @mutex.unlock
 
-      func
+      observer
     end
 
     # @param [Object] observer the observer to remove

lib/concurrent/ivar.rb

@@ -42,9 +42,15 @@ def initialize(value = NO_VALUE, opts = {})
     #
     # @param [Object] observer the object that will be notified of changes
     # @param [Symbol] func symbol naming the method to call when this `Observable` has changes`
-    def add_observer(observer, func = :update)
+    def add_observer(observer = nil, func = :update, &block)
+      raise ArgumentError.new('cannot provide both an observer and a block') if observer && block
       direct_notification = false
 
+      if block
+        observer = block
+        func = :call
+      end
+
       mutex.synchronize do
         if event.set?
           direct_notification = true
@@ -54,7 +60,7 @@ def add_observer(observer, func = :update)
       end
 
       observer.send(func, Time.now, self.value, reason) if direct_notification
-      func
+      observer
     end
 
     def set(value)

lib/concurrent/observable.rb

@@ -5,8 +5,8 @@ module Concurrent
 
   module Observable
 
-    def add_observer(*args)
-      observers.add_observer(*args)
+    def add_observer(*args, &block)
+      observers.add_observer(*args, &block)
     end
 
     def delete_observer(*args)
@@ -15,6 +15,7 @@ def delete_observer(*args)
 
     def delete_observers
       observers.delete_observers
+      self
     end
 
     def count_observers

lib/concurrent/scheduled_task.rb

@@ -51,9 +51,9 @@ def cancel
     end
     alias_method :stop, :cancel
 
-    def add_observer(*args)
+    def add_observer(*args, &block)
       if_state(:unscheduled, :pending, :in_progress) do
-        observers.add_observer(*args)
+        observers.add_observer(*args, &block)
       end
     end
 

spec/concurrent/agent_spec.rb

@@ -1,5 +1,6 @@
 require 'spec_helper'
 require_relative 'dereferenceable_shared'
+require_relative 'observable_shared'
 
 module Concurrent
 
@@ -38,6 +39,17 @@ def execute_dereferenceable(subject)
       end
 
       it_should_behave_like :dereferenceable
+
+      # observable
+      
+      subject{ Agent.new(0) }
+      
+      def trigger_observable(observable)
+        observable.post{ nil }
+        sleep(0.1)
+      end
+
+      it_should_behave_like :observable
     end
 
     context '#initialize' do

spec/concurrent/atomic/observer_set_shared.rb

@@ -8,15 +8,16 @@
 
   describe '#add_observer' do
 
-    context 'with argument' do
-      it 'should return the passed function' do
-        observer_set.add_observer(observer, :a_method).should eq(:a_method)
+    context 'with arguments' do
+      it 'should return the observer' do
+        observer_set.add_observer(observer, :a_method).should == observer
       end
     end
 
-    context 'without arguments' do
-      it 'should return the default function' do
-        observer_set.add_observer(observer).should eq(:update)
+    context 'with a block' do
+      it 'should return the observer based on a block' do
+        observer = observer_set.add_observer { :block }
+        observer.call.should == :block
       end
     end
   end
@@ -61,6 +62,14 @@
         observer_set.notify_observers('a string arg')
       end
 
+      it 'should notify an observer from a block' do
+        notification = double
+        expect(notification).to receive(:catch)
+
+        observer_set.add_observer {|arg| arg.catch }
+        observer_set.notify_observers notification
+      end
+
       it 'can be called many times' do
         expect(observer).to receive(:update).with(:an_arg).twice
         expect(observer).to receive(:update).with(no_args).once

spec/concurrent/channel/probe_spec.rb

@@ -1,4 +1,5 @@
 require 'spec_helper'
+require_relative '../observable_shared'
 
 module Concurrent
 
@@ -7,6 +8,19 @@ module Concurrent
     let(:channel) { Object.new }
     let(:probe) { Channel::Probe.new }
 
+    describe 'behavior' do
+
+      # observable
+      
+      subject{ Channel::Probe.new }
+      
+      def trigger_observable(observable)
+        observable.set('value')
+      end
+
+      it_should_behave_like :observable
+    end
+
     describe '#set_unless_assigned' do
       context 'empty probe' do
         it 'assigns the value' do

spec/concurrent/future_spec.rb

@@ -1,6 +1,7 @@
 require 'spec_helper'
 require_relative 'dereferenceable_shared'
 require_relative 'obligation_shared'
+require_relative 'observable_shared'
 
 module Concurrent
 
@@ -54,6 +55,17 @@ def execute_dereferenceable(subject)
       end
 
       it_should_behave_like :dereferenceable
+
+      # observable
+      
+      subject{ Future.new{ nil } }
+      
+      def trigger_observable(observable)
+        observable.execute
+        sleep(0.1)
+      end
+
+      it_should_behave_like :observable
     end
 
     context 'subclassing' do