Actor migrated to Edge::Future and integration added

Author: pitr-ch

lib/concurrent/actor.rb

@@ -1,9 +1,8 @@
 require 'concurrent/configuration'
-require 'concurrent/delay'
 require 'concurrent/executor/serialized_execution'
-require 'concurrent/ivar'
 require 'concurrent/logging'
 require 'concurrent/synchronization'
+require 'concurrent/edge/future'
 
 module Concurrent
   # TODO https://github.com/celluloid/celluloid/wiki/Supervision-Groups ?
@@ -40,9 +39,9 @@ def self.current
       Thread.current[:__current_actor__]
     end
 
-    @root = Delay.new do
-      Core.new(parent: nil, name: '/', class: Root, initialized: ivar = IVar.new).reference.tap do
-        ivar.no_error!
+    @root = Concurrent.delay do
+      Core.new(parent: nil, name: '/', class: Root, initialized: future = Concurrent.future).reference.tap do
+        future.wait!
       end
     end
 
@@ -77,7 +76,7 @@ def self.spawn(*args, &block)
 
     # as {.spawn} but it'll raise when Actor not initialized properly
     def self.spawn!(*args, &block)
-      spawn(spawn_optionify(*args).merge(initialized: ivar = IVar.new), &block).tap { ivar.no_error! }
+      spawn(spawn_optionify(*args).merge(initialized: future = Concurrent.future), &block).tap { future.wait! }
     end
 
     # @overload spawn_optionify(context_class, name, *args)

lib/concurrent/actor/behaviour/abstract.rb

@@ -38,7 +38,7 @@ def broadcast(event)
 
         def reject_envelope(envelope)
           envelope.reject! ActorTerminated.new(reference)
-          dead_letter_routing << envelope unless envelope.ivar
+          dead_letter_routing << envelope unless envelope.future
           log Logging::DEBUG, "rejected #{envelope.message} from #{envelope.sender_path}"
         end
       end

lib/concurrent/actor/behaviour/sets_results.rb

@@ -1,7 +1,7 @@
 module Concurrent
   module Actor
     module Behaviour
-      # Collects returning value and sets the IVar in the {Envelope} or error on failure.
+      # Collects returning value and sets the CompletableFuture in the {Envelope} or error on failure.
       class SetResults < Abstract
         attr_reader :error_strategy
 
@@ -12,8 +12,8 @@ def initialize(core, subsequent, error_strategy)
 
         def on_envelope(envelope)
           result = pass envelope
-          if result != MESSAGE_PROCESSED && !envelope.ivar.nil?
-            envelope.ivar.set result
+          if result != MESSAGE_PROCESSED && !envelope.future.nil?
+            envelope.future.success result
           end
           nil
         rescue => error
@@ -28,7 +28,7 @@ def on_envelope(envelope)
           else
             raise
           end
-          envelope.ivar.fail error unless envelope.ivar.nil?
+          envelope.future.fail error unless envelope.future.nil?
         end
       end
     end

lib/concurrent/actor/behaviour/termination.rb

@@ -7,18 +7,18 @@ module Behaviour
       class Termination < Abstract
 
         # @!attribute [r] terminated
-        #   @return [Event] event which will become set when actor is terminated.
+        #   @return [Edge::Event] event which will become set when actor is terminated.
         attr_reader :terminated
 
         def initialize(core, subsequent)
           super core, subsequent
-          @terminated = Event.new
+          @terminated = Concurrent.event
         end
 
         # @note Actor rejects envelopes when terminated.
         # @return [true, false] if actor is terminated
         def terminated?
-          @terminated.set?
+          @terminated.completed?
         end
 
         def on_envelope(envelope)
@@ -43,7 +43,7 @@ def on_envelope(envelope)
         # Terminates all its children, does not wait until they are terminated.
         def terminate!
           return true if terminated?
-          terminated.set
+          terminated.complete
           broadcast(:terminated) # TODO do not end up in Dead Letter Router
           parent << :remove_child if parent
           true

lib/concurrent/actor/context.rb

@@ -20,7 +20,7 @@ class AbstractContext
 
       # @abstract override to define Actor's behaviour
       # @param [Object] message
-      # @return [Object] a result which will be used to set the IVar supplied to Reference#ask
+      # @return [Object] a result which will be used to set the Future supplied to Reference#ask
       # @note self should not be returned (or sent to other actors), {#reference} should be used
       #   instead
       def on_message(message)
@@ -46,7 +46,7 @@ def pass
       end
 
       # Defines an actor responsible for dead letters. Any rejected message send
-      # with {Reference#tell} is sent there, a message with ivar is considered
+      # with {Reference#tell} is sent there, a message with future is considered
       # already monitored for failures. Default behaviour is to use
       # {AbstractContext#dead_letter_routing} of the parent, so if no
       # {AbstractContext#dead_letter_routing} method is overridden in

lib/concurrent/actor/core.rb

@@ -40,7 +40,7 @@ class Core < Synchronization::Object
       # @option opts [Array<Array(Behavior::Abstract, Array<Object>)>]
       #   behaviour_definition, array of pairs where each pair is behaviour
       #   class and its args, see {Behaviour.basic_behaviour_definition}
-      # @option opts [IVar, nil] initialized, if present it'll be set or failed
+      # @option opts [CompletableFuture, nil] initialized, if present it'll be set or failed
       #   after {Context} initialization
       # @option opts [Proc, nil] logger a proc accepting (level, progname,
       #   message = nil, &block) params, can be used to hook actor instance to
@@ -77,7 +77,7 @@ def initialize(opts = {}, &block)
 
           @args       = opts.fetch(:args, [])
           @block      = block
-          initialized = Type! opts[:initialized], IVar, NilClass
+          initialized = Type! opts[:initialized], Edge::CompletableFuture, NilClass
 
           messages = []
           messages << :link if opts[:link]
@@ -91,7 +91,7 @@ def initialize(opts = {}, &block)
                 handle_envelope Envelope.new(message, nil, parent, reference)
               end
 
-              initialized.set reference if initialized
+              initialized.success reference if initialized
             rescue => ex
               log ERROR, ex
               @first_behaviour.terminate!

lib/concurrent/actor/envelope.rb

@@ -5,18 +5,18 @@ class Envelope
 
       # @!attribute [r] message
       #   @return [Object] a message
-      # @!attribute [r] ivar
-      #   @return [IVar] an ivar which becomes resolved after message is processed
+      # @!attribute [r] future
+      #   @return [Edge::Future] a future which becomes resolved after message is processed
       # @!attribute [r] sender
       #   @return [Reference, Thread] an actor or thread sending the message
       # @!attribute [r] address
       #   @return [Reference] where this message will be delivered
 
-      attr_reader :message, :ivar, :sender, :address
+      attr_reader :message, :future, :sender, :address
 
-      def initialize(message, ivar, sender, address)
+      def initialize(message, future, sender, address)
         @message = message
-        @ivar    = Type! ivar, IVar, NilClass
+        @future  = Type! future, Edge::CompletableFuture, NilClass
         @sender  = Type! sender, Reference, Thread
         @address = Type! address, Reference
       end
@@ -34,7 +34,7 @@ def address_path
       end
 
       def reject!(error)
-        ivar.fail error unless ivar.nil?
+        future.fail error unless future.nil?
       end
     end
   end

lib/concurrent/actor/internal_delegations.rb

@@ -25,7 +25,7 @@ def dead_letter_routing
       end
 
       def redirect(reference, envelope = self.envelope)
-        reference.message(envelope.message, envelope.ivar)
+        reference.message(envelope.message, envelope.future)
         Behaviour::MESSAGE_PROCESSED
       end
 

lib/concurrent/actor/reference.rb

@@ -32,10 +32,13 @@ def tell(message)
       #
       # sends message to the actor and asks for the result of its processing, returns immediately
       # @param [Object] message
-      # @param [Ivar] ivar to be fulfilled be message's processing result
-      # @return [IVar] supplied ivar
-      def ask(message, ivar = IVar.new)
-        message message, ivar
+      # @param [Edge::Future] future to be fulfilled be message's processing result
+      # @return [Edge::Future] supplied future
+      def ask(message, future = Concurrent.future)
+        message message, future
+        # # @return [Future] a future
+        # def ask(message)
+        #   message message, ConcurrentNext.promise
       end
 
       # @note it's a good practice to use tell whenever possible. Ask should be used only for
@@ -45,17 +48,26 @@ def ask(message, ivar = IVar.new)
       #
       # sends message to the actor and asks for the result of its processing, blocks
       # @param [Object] message
-      # @param [Ivar] ivar to be fulfilled be message's processing result
+      # @param [Edge::Future] future to be fulfilled be message's processing result
       # @return [Object] message's processing result
-      # @raise [Exception] ivar.reason if ivar is #rejected?
-      def ask!(message, ivar = IVar.new)
-        ask(message, ivar).value!
+      # @raise [Exception] future.reason if future is #rejected?
+      def ask!(message, future = Concurrent.future)
+        ask(message, future).value!
+        # # @param [Object] message
+        # # @return [Object] message's processing result
+        # # @raise [Exception] future.reason if future is #failed?
+        # def ask!(message)
+        #   ask(message).value!
       end
 
-      # behaves as {#tell} when no ivar and as {#ask} when ivar
-      def message(message, ivar = nil)
-        core.on_envelope Envelope.new(message, ivar, Actor.current || Thread.current, self)
-        return ivar || self
+      # behaves as {#tell} when no future and as {#ask} when future
+      def message(message, future = nil)
+        core.on_envelope Envelope.new(message, future, Actor.current || Thread.current, self)
+        return future || self
+        # # behaves as {#tell} when no promise and as {#ask} when promise
+        # def message(message, promise = nil)
+        #   core.on_envelope Envelope.new(message, promise, Actor.current || Thread.current, self)
+        #   return promise ? promise.future : self
       end
 
       # @see AbstractContext#dead_letter_routing

lib/concurrent/edge/future.rb

@@ -394,6 +394,11 @@ def then(executor = nil, &callback)
         pr_then(default_executor, executor, &callback)
       end
 
+      # Creates new future where its value is result of asking actor with value of this Future.
+      def then_ask(actor)
+        self.then { |v| actor.ask(v) }.flat
+      end
+
       # @yield [reason] executed only on parent failure
       def rescue(executor = nil, &callback)
         pr_rescue(default_executor, executor, &callback)