add before_started, after_completed and after_failed functionality

Author: virtualstaticvoid

README.md

@@ -113,7 +113,7 @@ end
 process = MyProcess.create_process Date.today, :option_1 => true
 ```
 
-_NOTE:_ The current implementation performs a naive check on the count of arguments.
+_NOTE:_ The current implementation performs a naïve check on the count of arguments.
 
 Next, specify the tasks with their corresponding implementation methods, that make up the
 process, using the `task` method and providing the `method` to execute for the task.
@@ -291,6 +291,44 @@ module MyProcess
 end
 ```
 
+#### Before Process Started and After Process Completion or Failure
+
+You may want to run further tasks asynchrously before or after a process has completed
+or failed. These tasks provide a way to execute logic independently of the process.
+
+Specify these tasks using the `before_started`, `after_completed` or `after_failed` methods.
+
+For example, using `after_completed` to set off another business process or `after_failed` to 
+send an email to an operator.
+
+```ruby
+module MyProcess
+  extend Taskinator::Definition
+
+  # defines a process
+  define_process do
+
+    # tasks, sub-process, etc.
+
+    # define task to execute on completion
+    after_completed :further_process
+
+    # define task to execute on failure
+    after_failed :email_operations
+
+  end
+
+  def further_process
+    # ...
+  end
+
+  def email_operations
+    # ...
+  end
+
+end
+```
+
 #### Complex Process Definitions
 
 Any combination or nesting of `task`, `sequential`, `concurrent` and `for_each` steps are

lib/taskinator/builder.rb

@@ -83,9 +83,32 @@ def job(job, options={})
       nil
     end
 
-    # TODO: add mailer
-    # TODO: add complete!
-    # TODO: add fail!
+    # defines a task which executes the given @method before the process has started
+    def before_started(method, options={})
+      raise ArgumentError, 'method' if method.nil?
+      raise NoMethodError, method unless @executor.respond_to?(method)
+
+      define_before_started_task(@process, method, @args, options)
+      nil
+    end
+
+    # defines a task which executes the given @method after the process has completed
+    def after_completed(method, options={})
+      raise ArgumentError, 'method' if method.nil?
+      raise NoMethodError, method unless @executor.respond_to?(method)
+
+      define_after_completed_task(@process, method, @args, options)
+      nil
+    end
+
+    # defines a task which executes the given @method after the process has failed
+    def after_failed(method, options={})
+      raise ArgumentError, 'method' if method.nil?
+      raise NoMethodError, method unless @executor.respond_to?(method)
+
+      define_after_failed_task(@process, method, @args, options)
+      nil
+    end
 
     # defines a sub process task, for the given @definition
     # the definition specified must have input compatible arguments
@@ -104,13 +127,31 @@ def sub_process(definition, options={})
   private
 
     def define_step_task(process, method, args, options={})
-      define_task(process) {
+      add_task(process.tasks) {
         Task.define_step_task(process, method, args, combine_options(options))
       }
     end
 
+    def define_before_started_task(process, method, args, options={})
+      add_task(process.before_started_tasks) {
+        Task.define_hook_task(process, method, args, combine_options(options))
+      }
+    end
+
+    def define_after_completed_task(process, method, args, options={})
+      add_task(process.after_completed_tasks) {
+        Task.define_hook_task(process, method, args, combine_options(options))
+      }
+    end
+
+    def define_after_failed_task(process, method, args, options={})
+      add_task(process.after_failed_tasks) {
+        Task.define_hook_task(process, method, args, combine_options(options))
+      }
+    end
+
     def define_job_task(process, job, args, options={})
-      define_task(process) {
+      add_task(process.tasks) {
         Task.define_job_task(process, job, args, combine_options(options))
       }
     end
@@ -119,8 +160,8 @@ def define_sub_process_task(process, sub_process, options={})
       Task.define_sub_process_task(process, sub_process, combine_options(options))
     end
 
-    def define_task(process)
-      process.tasks << task = yield
+    def add_task(list)
+      list << task = yield
       task
     end
 

lib/taskinator/persistence.rb

@@ -274,16 +274,19 @@ def visit_process(attribute)
       end
 
       def visit_tasks(tasks)
-        tasks.each do |task|
-          RedisSerializationVisitor.new(@conn, task, @base_visitor).visit
-          @conn.rpush "#{@key}:tasks", task.uuid
-          unless task.is_a?(Task::SubProcess)
-            incr_task_count unless self == @base_visitor
-            @base_visitor.incr_task_count
-          end
-        end
-        @conn.set("#{@key}.count", tasks.count)
-        @conn.set("#{@key}.pending", tasks.count)
+        _visit_tasks(tasks)
+      end
+
+      def visit_before_started_tasks(tasks)
+        _visit_tasks(tasks, ':before_started')
+      end
+
+      def visit_after_completed_tasks(tasks)
+        _visit_tasks(tasks, ':after_completed')
+      end
+
+      def visit_after_failed_tasks(tasks)
+        _visit_tasks(tasks, ':after_failed')
       end
 
       def visit_attribute(attribute)
@@ -333,6 +336,21 @@ def task_count
       def incr_task_count
         @task_count += 1
       end
+
+      private
+
+      def _visit_tasks(tasks, list='')
+        tasks.each do |task|
+          RedisSerializationVisitor.new(@conn, task, @base_visitor).visit
+          @conn.rpush "#{@key}#{list}:tasks", task.uuid
+          unless task.is_a?(Task::SubProcess)
+            incr_task_count unless self == @base_visitor
+            @base_visitor.incr_task_count
+          end
+        end
+        @conn.set("#{@key}#{list}.count", tasks.count)
+        @conn.set("#{@key}#{list}.pending", tasks.count)
+      end
     end
 
     class XmlSerializationVisitor < Taskinator::Visitor::Base
@@ -385,17 +403,19 @@ def visit_process(attribute)
       end
 
       def visit_tasks(tasks)
-        builder.tag!('tasks', :count => tasks.count) do |xml|
-          tasks.each do |task|
-            xml.tag!('task', :key => task.key) do |xml2|
-              XmlSerializationVisitor.new(xml2, task, @base_visitor).visit
-              unless task.is_a?(Task::SubProcess)
-                incr_task_count unless self == @base_visitor
-                @base_visitor.incr_task_count
-              end
-            end
-          end
-        end
+        _visit_tasks(tasks)
+      end
+
+      def visit_before_started_tasks(tasks)
+        _visit_tasks(tasks, 'before_started')
+      end
+
+      def visit_after_completed_tasks(tasks)
+        _visit_tasks(tasks, 'after_completed')
+      end
+
+      def visit_after_failed_tasks(tasks)
+        _visit_tasks(tasks, 'after_failed')
       end
 
       def visit_attribute(attribute)
@@ -445,6 +465,22 @@ def task_count
       def incr_task_count
         @task_count += 1
       end
+
+      private
+
+      def _visit_tasks(tasks, list='tasks')
+        builder.tag!(list, :count => tasks.count) do |xml|
+          tasks.each do |task|
+            xml.tag!('task', :key => task.key) do |xml2|
+              XmlSerializationVisitor.new(xml2, task, @base_visitor).visit
+              unless task.is_a?(Task::SubProcess)
+                incr_task_count unless self == @base_visitor
+                @base_visitor.incr_task_count
+              end
+            end
+          end
+        end
+      end
     end
 
     class UnknownTypeError < StandardError
@@ -540,11 +576,19 @@ def visit_process(attribute)
       end
 
       def visit_tasks(tasks)
-        # tasks are a linked list, so just get the first one
-        Taskinator.redis do |conn|
-          uuid = conn.lindex("#{@key}:tasks", 0)
-          tasks.attach(lazy_instance_for(Task, uuid), conn.get("#{@key}.count").to_i) if uuid
-        end
+        _visit_tasks(tasks)
+      end
+
+      def visit_before_started_tasks(tasks)
+        _visit_tasks(tasks, ':before_started')
+      end
+
+      def visit_after_completed_tasks(tasks)
+        _visit_tasks(tasks, ':after_completed')
+      end
+
+      def visit_after_failed_tasks(tasks)
+        _visit_tasks(tasks, ':after_failed')
       end
 
       def visit_process_reference(attribute)
@@ -606,6 +650,14 @@ def visit_args(attribute)
 
       private
 
+      def _visit_tasks(tasks, list='')
+        # tasks are a linked list, so just get the first one
+        Taskinator.redis do |conn|
+          uuid = conn.lindex("#{@key}#{list}:tasks", 0)
+          tasks.attach(lazy_instance_for(Task, uuid), conn.get("#{@key}#{list}.count").to_i) if uuid
+        end
+      end
+
       #
       # creates a proxy for the instance which
       # will only fetch the instance when used
@@ -648,14 +700,31 @@ def visit_process(attribute)
       end
 
       def visit_tasks(tasks)
-        @conn.expire "#{@key}:tasks", expire_in
-        @conn.expire "#{@key}.count", expire_in
-        @conn.expire "#{@key}.pending", expire_in
+        _visit_tasks(tasks)
+      end
+
+      def visit_before_started_tasks(tasks)
+        _visit_tasks(tasks, ':before_started')
+      end
+
+      def visit_after_completed_tasks(tasks)
+        _visit_tasks(tasks, ':after_completed')
+      end
+
+      def visit_after_failed_tasks(tasks)
+        _visit_tasks(tasks, ':after_failed')
+      end
+
+      private
+
+      def _visit_tasks(tasks, list='')
+        @conn.expire "#{@key}#{list}:tasks", expire_in
+        @conn.expire "#{@key}#{list}.count", expire_in
+        @conn.expire "#{@key}#{list}.pending", expire_in
         tasks.each do |task|
           RedisCleanupVisitor.new(@conn, task, expire_in).visit
         end
       end
-
     end
 
     # lazily loads the object specified by the type and uuid

lib/taskinator/process.rb

@@ -55,6 +55,18 @@ def tasks
       @tasks ||= Tasks.new
     end
 
+    def before_started_tasks
+      @before_started_tasks ||= Tasks.new
+    end
+
+    def after_completed_tasks
+      @after_completed_tasks ||= Tasks.new
+    end
+
+    def after_failed_tasks
+      @after_failed_tasks ||= Tasks.new
+    end
+
     def no_tasks_defined?
       tasks.empty?
     end
@@ -64,6 +76,9 @@ def accept(visitor)
       visitor.visit_task_reference(:parent)
       visitor.visit_type(:definition)
       visitor.visit_tasks(tasks)
+      visitor.visit_before_started_tasks(before_started_tasks)
+      visitor.visit_after_completed_tasks(after_completed_tasks)
+      visitor.visit_after_failed_tasks(after_failed_tasks)
       visitor.visit_args(:options)
       visitor.visit_attribute(:scope)
       visitor.visit_attribute(:queue)
@@ -92,6 +107,9 @@ def enqueue!
     def start!
       return if paused? || cancelled?
 
+      # enqueue before started tasks independently
+      before_started_tasks.each(&:enqueue!)
+
       transition(:processing) do
         instrument('taskinator.process.processing', processing_payload) do
           start
@@ -132,6 +150,9 @@ def complete!
           end
         end
       end
+
+      # enqueue completion tasks independently
+      after_completed_tasks.each(&:enqueue!)
     end
 
     # TODO: add retry method - to pick up from a failed task
@@ -159,6 +180,9 @@ def fail!(error)
           parent.fail!(error) unless parent.nil?
         end
       end
+
+      # enqueue completion tasks independently
+      after_failed_tasks.each(&:enqueue!)
     end
 
     def task_failed(task, error)