Parallel interactor performance in organizers

README.md

@@ -24,8 +24,9 @@ Ruby interactors with [ActiveModel::Validations] based on the [interactor][colle
     * [Validating the Context](#validating-the-context)
   * [Using Interactors](#using-interactors)
     * [Kinds of Interactors](#kinds-of-interactors)
-    * [Interactors](#interactors)
-    * [Organizers](#organizers)
+      * [Interactors](#interactors)
+      * [Organizers](#organizers)
+      * [Parallel Organizers](#parallel-organizers)
     * [Rollback](#rollback)
     * [Callbacks](#callbacks)
       * [Validation Callbacks](#validation-callbacks)
@@ -346,7 +347,7 @@ Finally, the context (along with any changes made to it) is returned.
 
 There are two kinds of interactors built into the Interactor library: basic interactors and organizers.
 
-#### Interactors
+##### Interactors
 
 A basic interactor is a class that includes Interactor and defines `perform`.\
 
@@ -366,7 +367,7 @@ end
 
 Basic interactors are the building blocks. They are your application's single-purpose units of work.
 
-#### Organizers
+##### Organizers
 
 An organizer is an important variation on the basic interactor. Its single purpose is to run other interactors.
 
@@ -421,6 +422,41 @@ end
 The organizer passes its context to the interactors that it organizes, one at a time and in order. Each interactor may
 change that context before it's passed along to the next interactor.
 
+##### Parallel Organizers
+
+Organizers can be told to run their interactors in parallel with the `#perform_in_parallel` class method.  This
+will run each interactor in parallel with one and other only passing the original context to each organizer.
+This means each interactor must be able to perform without dependencies on prior interactor runs.
+
+```ruby
+class CreateNewUser < ActiveInteractor::Base
+  def perform
+    context.user = User.create(
+      first_name: context.first_name,
+      last_name: context.last_name
+    )
+  end
+end
+
+class LogNewUserCreation < ActiveInteractor::Base
+  def perform
+    context.log = Log.create(
+      event: 'new user created',
+      first_name: context.first_name,
+      last_name: context.last_name
+    )
+  end
+end
+
+class CreateUser < ActiveInteractor::Organizer
+  perform_in_parallel
+  organize :create_new_user, :log_new_user_creation
+end
+
+CreateUser.perform(first_name: 'Aaron', last_name: 'Allen')
+#=> <#CreateUser::Context first_name='Aaron' last_name='Allen' user=>#<User ...> log=<#Log ...>>
+```
+
 #### Rollback
 
 If any one of the organized interactors fails its context, the organizer stops. If the `ChargeCard` interactor fails,

lib/active_interactor/context/attributes.rb

@@ -36,14 +36,6 @@ def __attributes
         end
       end
 
-      # @api private
-      # @param context [Hash|Context::Base] attributes to assign to the context
-      # @return [Context::Base] a new instance of {Context::Base}
-      def initialize(context = {})
-        copy_flags!(context)
-        super
-      end
-
       # Attributes defined on the instance
       # @example Get attributes defined on an instance
       #   class MyInteractor::Context < ActiveInteractor::Context::Base
@@ -65,15 +57,6 @@ def attributes
           hash[attribute] = self[attribute] if self[attribute]
         end
       end
-
-      private
-
-      def copy_flags!(context)
-        %w[_called _failed _rolled_back].each do |flag|
-          value = context.instance_variable_get("@#{flag}")
-          instance_variable_set("@#{flag}", value)
-        end
-      end
     end
   end
 end

lib/active_interactor/context/base.rb

@@ -15,6 +15,15 @@ class Base < OpenStruct
       include ActiveModel::Validations
       include Attributes
 
+      # @param context [Hash|Context::Base] attributes to assign to the context
+      # @return [Context::Base] a new instance of {Context::Base}
+      def initialize(context = {})
+        merge_errors!(context.errors) if context.respond_to?(:errors)
+        copy_flags!(context)
+        copy_called!(context)
+        super
+      end
+
       # @!method valid?(context = nil)
       # @see
       #   https://github.com/rails/rails/blob/master/activemodel/lib/active_model/validations.rb#L305
@@ -72,6 +81,37 @@ def failure?
       end
       alias fail? failure?
 
+      # Merge an instance of context or a hash into an existing context
+      # @since 1.0.0
+      # @example
+      #   class MyInteractor1 < ActiveInteractor::Base
+      #     def perform
+      #       context.first_name = 'Aaron'
+      #     end
+      #   end
+      #
+      #  class MyInteractor2 < ActiveInteractor::Base
+      #    def perform
+      #      context.last_name = 'Allen'
+      #    end
+      #  end
+      #
+      #  result = MyInteractor1.perform
+      #  #=> <#MyInteractor1::Context first_name='Aaron'>
+      #
+      #  result.merge!(MyInteractor2.perform)
+      #  #=> <#MyInteractor1::Context first_name='Aaron' last_name='Allen'>
+      # @param context [Base|Hash] attributes to merge into the context
+      # @return [Base] an instance of {Base}
+      def merge!(context)
+        merge_errors!(context.errors) if context.respond_to?(:errors)
+        copy_flags!(context)
+        context.each_pair do |key, value|
+          self[key] = value
+        end
+        self
+      end
+
       # Roll back an interactor context. Any interactors to which this
       # context has been passed and which have been successfully called are asked
       # to roll themselves back by invoking their
@@ -129,6 +169,18 @@ def _called
         @_called ||= []
       end
 
+      def copy_called!(context)
+        value = context.instance_variable_get('@_called') || []
+        instance_variable_set('@_called', value)
+      end
+
+      def copy_flags!(context)
+        %w[_failed _rolled_back].each do |flag|
+          value = context.instance_variable_get("@#{flag}")
+          instance_variable_set("@#{flag}", value)
+        end
+      end
+
       def merge_errors!(errors)
         if errors.is_a? String
           self.errors.add(:context, errors)

lib/active_interactor/interactor/worker.rb

@@ -17,25 +17,29 @@ def initialize(interactor)
       end
 
       # Calls {#execute_perform!} and rescues {Error::ContextFailure}
+      # @param options [Hash] execution options for the interactor perform step
+      # @option options [Boolean] skip_rollback whether or not to skip rollback
+      #  on the interactor
       # @return [Context::Base] an instance of {Context::Base}
-      def execute_perform
-        execute_perform!
+      def execute_perform(options = {})
+        execute_perform!(options)
       rescue Error::ContextFailure => e
         ActiveInteractor.logger.error("ActiveInteractor: #{e}")
         context
       end
 
       # Calls {Interactor#perform} with callbacks and context validation
+      # @param options [Hash] execution options for the interactor perform step
+      # @option options [Boolean] skip_rollback whether or not to skip rollback
+      #  on the interactor
       # @raise [Error::ContextFailure] if the context fails
       # @return [Context::Base] an instance of {Context::Base}
-      def execute_perform!
+      def execute_perform!(options = {})
         run_callbacks :perform do
           execute_context!
           @context = interactor.finalize_context!
-        rescue StandardError
-          @context = interactor.finalize_context!
-          execute_rollback
-          raise
+        rescue StandardError => e
+          handle_error(e, options)
         end
       end
 
@@ -58,6 +62,12 @@ def execute_context!
         interactor.context_fail! unless validate_context(:called)
       end
 
+      def handle_error(exception, options)
+        @context = interactor.finalize_context!
+        execute_rollback unless options[:skip_rollback]
+        raise exception
+      end
+
       def validate_context(validation_context = nil)
         run_callbacks :validation do
           interactor.context_valid?(validation_context)

lib/active_interactor/organizer.rb

@@ -11,6 +11,11 @@ module ActiveInteractor
   # @!attribute [r] organized
   #  @!scope class
   #  @return [Array<Base>] the organized interactors
+  # @!attribute [r] parallel
+  #  @since 1.0.0
+  #  @!scope class
+  #  @return [Boolean] whether or not to run the interactors
+  #    in parallel
   # @example a basic organizer
   #  class MyInteractor1 < ActiveInteractor::Base
   #    def perform
@@ -32,6 +37,7 @@ module ActiveInteractor
   #  #=> <MyOrganizer::Context interactor1=true interactor2=true>
   class Organizer < Base
     class_attribute :organized, instance_writer: false, default: []
+    class_attribute :parallel, instance_writer: false, default: false
     define_callbacks :each_perform
 
     # Define a callback to call after each organized interactor's
@@ -183,25 +189,51 @@ def self.organize(*interactors)
       end.compact
     end
 
+    # Run organized interactors in parallel
+    # @since 1.0.0
+    def self.perform_in_parallel
+      self.parallel = true
+    end
+
     # Invoke the organized interactors. An organizer is
     #  expected not to define its own {Base#perform} method
     #  in favor of this default implementation.
     def perform
-      self.class.organized.each do |interactor|
-        self.context = execute_interactor_perform_with_callbacks!(interactor)
+      if self.class.parallel
+        perform_in_parallel
+      else
+        perform_in_order
       end
-    rescue Error::ContextFailure => e
-      self.context = e.context
-    ensure
-      self.context = self.class.context_class.new(context)
     end
 
     private
 
-    def execute_interactor_perform_with_callbacks!(interactor)
+    def execute_interactor_with_callbacks(interactor, fail_on_error = false, execute_options = {})
       run_callbacks :each_perform do
-        interactor.new(context).execute_perform!
+        instance = interactor.new(context)
+        method = fail_on_error ? :execute_perform! : :execute_perform
+        instance.send(method, execute_options)
+      end
+    end
+
+    def merge_contexts(contexts)
+      contexts.each { |context| @context.merge!(context) }
+      context_fail! if contexts.any?(&:failure?)
+    end
+
+    def perform_in_order
+      self.class.organized.each do |interactor|
+        context.merge!(execute_interactor_with_callbacks(interactor, true))
+      end
+    rescue Error::ContextFailure => e
+      context.merge!(e.context)
+    end
+
+    def perform_in_parallel
+      results = self.class.organized.map do |interactor|
+        Thread.new { execute_interactor_with_callbacks(interactor, false, skip_rollback: true) }
       end
+      merge_contexts(results.map(&:value))
     end
   end
 end