Atom updates.

- No longer `Dereferenceable`
- Now `Observable`
- Added a `#reset` method

Author: jdantonio

CHANGELOG.md

@@ -5,6 +5,10 @@
   - Each object uses its own `SingleThreadExecutor` instead of the global thread pool.
   - No longers supports executor injection
   - Much better documentation
+* `Atom` updates
+  - No longer `Dereferenceable`
+  - Now `Observable`
+  - Added a `#reset` method
 
 ## Current Release v1.0.0.pre1 (19 Aug 2015)
 

lib/concurrent/atom.rb

@@ -1,5 +1,6 @@
-require 'concurrent/concern/dereferenceable'
 require 'concurrent/atomic/atomic_reference'
+require 'concurrent/collection/copy_on_notify_observer_set'
+require 'concurrent/concern/observable'
 require 'concurrent/synchronization/object'
 
 module Concurrent
@@ -18,11 +19,9 @@ module Concurrent
   # new value to the result of running the given block if and only if that
   # value validates.
   #
-  # @!macro copy_options
-  #
   # @see http://clojure.org/atoms Clojure Atoms
   class Atom < Synchronization::Object
-    include Concern::Dereferenceable
+    include Concern::Observable
 
     # Create a new atom with the given initial value.
     #
@@ -38,20 +37,18 @@ class Atom < Synchronization::Object
     # @raise [ArgumentError] if the validator is not a `Proc` (when given)
     def initialize(value, opts = {})
       super()
-
-      @validator = opts.fetch(:validator, ->(v){ true })
-      raise ArgumentError.new('validator must be a proc') unless @validator.is_a? Proc
-
-      @value = Concurrent::AtomicReference.new(value)
-      ns_set_deref_options(opts)
-      ensure_ivar_visibility!
+      synchronize do
+        @validator = opts.fetch(:validator, ->(v){ true })
+        @value = Concurrent::AtomicReference.new(value)
+        self.observers = Collection::CopyOnNotifyObserverSet.new
+      end
     end
 
     # The current value of the atom.
     #
     # @return [Object] The current value.
     def value
-      apply_deref_options(@value.value)
+      @value.value
     end
     alias_method :deref, :value
 
@@ -87,45 +84,66 @@ def value
     def swap(*args)
       raise ArgumentError.new('no block given') unless block_given?
 
-      begin
-        loop do
-          old_value = @value.value
+      loop do
+        old_value = @value.value
+        begin
           new_value = yield(old_value, *args)
-          return old_value unless @validator.call(new_value)
-          return new_value if compare_and_set!(old_value, new_value)
+          break old_value unless valid?(new_value)
+          break new_value if compare_and_set(old_value, new_value)
+        rescue
+          break old_value
         end
-      rescue
-        return @value.value
       end
     end
 
-    # @!macro [attach] atom_compare_and_set
-    #   Atomically sets the value of atom to the new value if and only if the
-    #   current value of the atom is identical to the old value and the new
-    #   value successfully validates against the (optional) validator given
-    #   at construction.
+    # Atomically sets the value of atom to the new value if and only if the
+    # current value of the atom is identical to the old value and the new
+    # value successfully validates against the (optional) validator given
+    # at construction.
     #
-    #   @param [Object] old_value The expected current value.
-    #   @param [Object] new_value The intended new value.
+    # @param [Object] old_value The expected current value.
+    # @param [Object] new_value The intended new value.
     #
-    #   @return [Boolean] True if the value is changed else false.
+    # @return [Boolean] True if the value is changed else false.
     def compare_and_set(old_value, new_value)
-      compare_and_set!(old_value, new_value)
-    rescue
-      false
+      if valid?(new_value) && @value.compare_and_set(old_value, new_value)
+        observers.notify_observers(Time.now, new_value, nil)
+        true
+      else
+        false
+      end
     end
 
-    private
-
-    # @!macro atom_compare_and_set
-    # @raise [Exception] if the validator proc raises an exception
-    # @!visibility private
-    def compare_and_set!(old_value, new_value)
-      if @validator.call(new_value) # may raise exception
-        @value.compare_and_set(old_value, new_value)
+    # Atomically sets the value of atom to the new value without regard for the
+    # current value so long as the new value successfully validates against the
+    # (optional) validator given at construction.
+    #
+    # @param [Object] new_value The intended new value.
+    #
+    # @return [Object] The final value of the atom after all operations and
+    #   validations are complete.
+    def reset(new_value)
+      old_value = @value.value
+      if valid?(new_value)
+        @value.set(new_value)
+        observers.notify_observers(Time.now, new_value, nil)
+        new_value
       else
-        false
+        old_value
       end
     end
+
+    private
+
+    # Is the new value valid?
+    #
+    # @param [Object] new_value The intended new value.
+    # @return [Boolean] false if the validator function returns false or raises
+    #   an exception else true
+    def valid?(new_value)
+      @validator.call(new_value)
+    rescue
+      false
+    end
   end
 end

lib/concurrent/concern/observable.rb

@@ -21,8 +21,8 @@ module Concern
     #
     # In a multi threaded environment things are more complex. The `subject` must
     # synchronize the access to its data structure and to do so currently we're
-    # using two specialized ObserverSet: CopyOnWriteObserverSet and
-    # CopyOnNotifyObserverSet.
+    # using two specialized ObserverSet: {Concurrent::Concern::CopyOnWriteObserverSet}
+    # and {Concurrent::Concern::CopyOnNotifyObserverSet}.
     #
     # When implementing and `observer` there's a very important rule to remember:
     # **there are no guarantees about the thread that will execute the callback**

spec/concurrent/atom_spec.rb

@@ -1,27 +1,15 @@
-require_relative 'concern/dereferenceable_shared'
+require_relative 'concern/observable_shared'
 
 module Concurrent
 
   describe Atom do
 
-    it_should_behave_like :dereferenceable do
-      def dereferenceable_subject(value, opts = {})
-        Atom.new(value, opts)
-      end
-    end
-
     context 'construction' do
 
       it 'sets the initial value to the given value' do
         atom = Atom.new(42)
         expect(atom.value).to eq 42
       end
-
-      it 'raises an exception if the validator is not a proc' do
-        expect {
-          Atom.new(42, validator: 42)
-        }.to raise_error(ArgumentError)
-      end
     end
 
     context '#compare_and_set' do
@@ -75,7 +63,7 @@ def dereferenceable_subject(value, opts = {})
       end
     end
 
-    context 'swap' do
+    context '#swap' do
 
       it 'raises an exception when no block is given' do
         atom = Atom.new(42)
@@ -141,25 +129,70 @@ def dereferenceable_subject(value, opts = {})
         expect(atom.swap{ 100 }).to eq 42
       end
 
-      #it 'calls the block more than once if the value changes underneath' do
-      #latch = Concurrent::CountDownLatch.new
-      #counter = Concurrent::AtomicBoolean.new(0)
-      #atom = Atom.new(0)
+      it 'calls the block more than once if the value changes underneath' do
+        latch1 = Concurrent::CountDownLatch.new
+        latch2 = Concurrent::CountDownLatch.new
+        counter = Concurrent::AtomicFixnum.new(0)
+        atom = Atom.new(0)
+
+        t = Thread.new do
+          atom.swap do |value|
+            latch1.count_down
+            latch2.wait(1)
+            counter.increment
+            42
+          end
+        end
+
+        latch1.wait(1)
+        atom.swap{ 100 }
+        latch2.count_down
+        t.join(1)
+
+        expect(counter.value).to be > 1
+      end
+    end
+
+    context '#reset' do
+
+      it 'sets the new value' do
+        atom = Atom.new(42)
+        atom.reset(:foo)
+        expect(atom.value).to eq :foo
+      end
+
+      it 'returns the new value on success' do
+        atom = Atom.new(42)
+        expect(atom.reset(:foo)).to eq :foo
+      end
+
+      it 'returns the new value on success' do
+        atom = Atom.new(42)
+        expect(atom.reset(:foo)).to eq :foo
+      end
+
+      it 'returns the old value if the validator returns false' do
+        validator = ->(value){ false }
+        atom = Atom.new(42, validator: validator)
+        expect(atom.reset(:foo)).to eq 42
+      end
+
+      it 'returns the old value if the validator raises an exception' do
+        validator = ->(value){ raise StandardError }
+        atom = Atom.new(42, validator: validator)
+        expect(atom.reset(:foo)).to eq 42
+      end
+    end
 
-      #t = Thread.new do
-      #atom.swap do |value|
-      #counter.increment
-      #latch.wait
-      #42
-      #end
-      #end
+    context :observable do
 
-      #atom.swap{ 100 }
-      #latch.count_down
-      #t.join(1)
+      subject { Atom.new(0) }
+
+      def trigger_observable(observable)
+        observable.reset(42)
+      end
 
-      #expect(counter.value).to eq 2
-      #end
+      it_behaves_like :observable
     end
   end
 end

spec/concurrent/timer_task_spec.rb

@@ -4,11 +4,6 @@
 module Concurrent
 
   describe TimerTask do
-    #before(:each) do
-      ## suppress deprecation warnings.
-      #allow_any_instance_of(Concurrent::TimerTask).to receive(:warn)
-      #allow(Concurrent::TimerTask).to receive(:warn)
-    #end
 
     context :dereferenceable do