Promise extends IVar

Author: jdantonio

lib/concurrent/promise.rb

@@ -1,5 +1,6 @@
 require 'thread'
 
+require 'concurrent/ivar'
 require 'concurrent/obligation'
 require 'concurrent/executor/executor_options'
 
@@ -181,8 +182,7 @@ module Concurrent
   # - `on_success { |result| ... }` is the same as `then {|result| ... }`
   # - `rescue { |reason| ... }` is the same as `then(Proc.new { |reason| ... } )`
   # - `rescue` is aliased by `catch` and `on_error`
-  class Promise
-    include Obligation
+  class Promise < IVar
     include ExecutorOptions
 
     # Initialize a new Promise with the provided options.
@@ -203,6 +203,7 @@ class Promise
     # @see http://promises-aplus.github.io/promises-spec/
     def initialize(opts = {}, &block)
       opts.delete_if { |k, v| v.nil? }
+      super(IVar::NO_VALUE, opts)
 
       @executor = get_executor_from(opts) || Concurrent.global_io_executor
       @args = get_arguments_from(opts)
@@ -214,17 +215,13 @@ def initialize(opts = {}, &block)
       @promise_body = block || Proc.new { |result| result }
       @state = :unscheduled
       @children = []
-
-      init_obligation
-      set_deref_options(opts)
     end
 
     # @return [Promise]
     def self.fulfill(value, opts = {})
       Promise.new(opts).tap { |p| p.send(:synchronized_set_state!, true, value, nil) }
     end
 
-
     # @return [Promise]
     def self.reject(reason, opts = {})
       Promise.new(opts).tap { |p| p.send(:synchronized_set_state!, false, nil, reason) }
@@ -388,6 +385,18 @@ def self.any?(*promises)
       aggregate(:any?, *promises)
     end
 
+    def set(value)
+      raise PromiseExecutionError.new('supported only on root promises') unless root?
+      super
+    end
+
+    def fail(reason = StandardError.new)
+      raise PromiseExecutionError.new('supported only on root promises') unless root?
+      super
+    end
+
+    protected :complete
+
     protected
 
     # Aggregate a collection of zero or more promises under a composite promise,
@@ -444,17 +453,21 @@ def notify_child(child)
       if_state(:rejected) { child.on_reject(@reason) }
     end
 
+    # @!visibility private
+    def complete(success, value, reason)
+      children_to_notify = mutex.synchronize do
+        set_state!(success, value, reason)
+        @children.dup
+      end
+
+      children_to_notify.each { |child| notify_child(child) }
+    end
+
     # @!visibility private
     def realize(task)
       @executor.post do
         success, value, reason = SafeTaskExecutor.new(task).execute(*@args)
-
-        children_to_notify = mutex.synchronize do
-          set_state!(success, value, reason)
-          @children.dup
-        end
-
-        children_to_notify.each { |child| notify_child(child) }
+        complete(success, value, reason)
       end
     end
 

spec/concurrent/promise_spec.rb

@@ -374,41 +374,41 @@ def get_ivar_from_args(opts)
 
           composite = Promise.all?(promise1, promise2, promise3).
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq 1
-        end
+          end
 
         it 'executes the #then condition when no promises are given' do
           counter = Concurrent::AtomicFixnum.new(0)
           latch = Concurrent::CountDownLatch.new(1)
 
           composite = Promise.all?.
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq 1
-        end
+          end
 
         it 'executes the #rescue handler if even one component fails' do
           counter = Concurrent::AtomicFixnum.new(0)
           latch = Concurrent::CountDownLatch.new(1)
 
           composite = Promise.all?(promise1, promise2, rejected_subject, promise3).
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq -1
-        end
+          end
       end
 
       describe '.any?' do
@@ -429,46 +429,84 @@ def get_ivar_from_args(opts)
 
           composite = Promise.any?(promise1, promise2, rejected_subject, promise3).
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq 1
-        end
+          end
 
         it 'executes the #then condition when no promises are given' do
           counter = Concurrent::AtomicFixnum.new(0)
           latch = Concurrent::CountDownLatch.new(1)
 
           composite = Promise.any?.
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq 1
-        end
+          end
 
         it 'executes the #rescue handler if all componenst fail' do
           counter = Concurrent::AtomicFixnum.new(0)
           latch = Concurrent::CountDownLatch.new(1)
 
           composite = Promise.any?(rejected_subject, rejected_subject, rejected_subject, rejected_subject).
             then { counter.up; latch.count_down }.
-            rescue { counter.down; latch.count_down }.
-            execute
+        rescue { counter.down; latch.count_down }.
+          execute
 
           latch.wait(1)
 
           expect(counter.value).to eq -1
-        end
+          end
       end
     end
 
     context 'fulfillment' do
 
+      context '#set' do
+
+        it '#can only be called on the root promise' do
+          root = Promise.new{ :foo }
+          child = root.then{ :bar }
+
+          expect { child.set('foo') }.to raise_error PromiseExecutionError
+          expect { root.set('foo') }.not_to raise_error
+        end
+
+        it 'triggers children' do
+          expected = nil
+          root = Promise.new(executor: :immediate){ nil }
+          root.then{ |result| expected = result }
+          root.set(20)
+          expect(expected).to eq 20
+        end
+      end
+
+      context '#fail' do
+
+        it 'can only be called on the root promise' do
+          root = Promise.new{ :foo }
+          child = root.then{ :bar }
+
+          expect { child.fail }.to raise_error PromiseExecutionError
+          expect { root.fail }.not_to raise_error
+        end
+
+        it 'rejects children' do
+          expected = nil
+          root = Promise.new(executor: :immediate)
+          root.then(Proc.new{ |reason| expected = reason })
+          root.fail(ArgumentError.new('simulated error'))
+          expect(expected).to be_a ArgumentError
+        end
+      end
+
       it 'passes the result of each block to all its children' do
         expected = nil
         Promise.new(executor: executor){ 20 }.then{ |result| expected = result }.execute