wip! to be completed

Author: virtualstaticvoid

lib/taskinator/create_process_worker.rb

@@ -31,6 +31,10 @@ def perform
         process_args << { :uuid => uuid }
       end
 
+      # generate the process for the given definition and arguments
+      # and enqueue the processes tasks
+      # -> sequential processes - enqueues the first task
+      # -> concurrent processes - enqueues all the tasks
       @definition._create_process_(false, *process_args).enqueue!
 
     end

lib/taskinator/executor.rb

@@ -26,9 +26,15 @@ def options
     end
 
     # helpers
+
     def logger
       Taskinator.logger
     end
 
+    def error
+      # task.process.error
+      raise NoMethodError
+    end
+
   end
 end

lib/taskinator/process.rb

@@ -51,6 +51,10 @@ def parent=(value)
       @key = nil # NB: invalidate memoized key
     end
 
+    def sub_process?
+      defined?(@parent)
+    end
+
     def tasks
       @tasks ||= Tasks.new
     end
@@ -155,9 +159,6 @@ def complete!
       after_completed_tasks.each(&:enqueue!)
     end
 
-    # TODO: add retry method - to pick up from a failed task
-    #  e.g. like retrying a failed job in Resque Web
-
     def tasks_completed?
       # TODO: optimize this
       tasks.all?(&:completed?)
@@ -185,8 +186,25 @@ def fail!(error)
       after_failed_tasks.each(&:enqueue!)
     end
 
+    #--------------------------------------------------
+
+    # TODO: add retry method - to pick up from a failed task
+    #  e.g. like retrying a failed job in Resque Web
+
+    def task_started(task)
+      return if processing? || sub_process?
+
+      transition(:processing) do
+        # enqueue before started tasks independently
+        before_started_tasks.each(&:enqueue!)
+      end
+    end
+
+    def task_cancelled(task)
+      cancel!
+    end
+
     def task_failed(task, error)
-      # for now, fail this process
       fail!(error)
     end
 
@@ -271,7 +289,6 @@ def enqueue
         end
       end
 
-      # this method only called in-process (usually from the console)
       def start
         if tasks.empty?
           complete! # weren't any tasks to start with

lib/taskinator/task.rb

@@ -83,6 +83,9 @@ def start!
 
       transition(:processing) do
         instrument('taskinator.task.processing', processing_payload) do
+          # notify the process that this task has started
+          process.task_started(self) if notify_process?
+
           start
         end
       end
@@ -99,21 +102,27 @@ def paused?
     end
 
     def complete!
+      self.incr_completed if incr_count?
+
       transition(:completed) do
-        self.incr_completed if incr_count?
         instrument('taskinator.task.completed', completed_payload) do
           complete if respond_to?(:complete)
+
           # notify the process that this task has completed
           process.task_completed(self) if notify_process?
         end
       end
     end
 
     def cancel!
+      self.incr_cancelled if incr_count?
+
       transition(:cancelled) do
-        self.incr_cancelled if incr_count?
         instrument('taskinator.task.cancelled', cancelled_payload) do
           cancel if respond_to?(:cancel)
+
+          # notify the process that this task has cancelled
+          process.task_cancelled(self) if notify_process?
         end
       end
     end
@@ -123,10 +132,12 @@ def cancelled?
     end
 
     def fail!(error)
+      self.incr_failed if incr_count?
+
       transition(:failed) do
-        self.incr_failed if incr_count?
         instrument('taskinator.task.failed', failed_payload(error)) do
           fail(error) if respond_to?(:fail)
+
           # notify the process that this task has failed
           process.task_failed(self, error) if notify_process?
         end

spec/support/test_definitions.rb

@@ -6,6 +6,9 @@ def self.perform(*args)
     end
   end
 
+  class TestTaskFailed < StandardError
+  end
+
   module Support
 
     def iterator(task_count, *args)
@@ -22,6 +25,22 @@ def iterator(task_count, *args)
       end
     end
 
+    def task_fail(*args)
+      raise TestTaskFailed
+    end
+
+    def task_before_started(*args)
+      Taskinator.logger.info(">>> Executing before started task #{__method__} [#{uuid}]...")
+    end
+
+    def task_after_completed(*args)
+      Taskinator.logger.info(">>> Executing after completed task #{__method__} [#{uuid}]...")
+    end
+
+    def task_after_failed(*args)
+      Taskinator.logger.info(">>> Executing after failed task #{__method__} [#{uuid}]...")
+    end
+
   end
 
   module Definition
@@ -46,12 +65,19 @@ module TaskBeforeStarted
     extend Taskinator::Definition
     include Support
 
-    define_process :task_count do
-      before_started :task1, :queue => :foo
+    define_process do
+      before_started :task_before_started
 
-      for_each :iterator do
-        task :task2, :queue => :foo
-      end
+      task :task1
+    end
+  end
+
+  module TaskBeforeStartedSubProcess
+    extend Taskinator::Definition
+    include Support
+
+    define_process do
+      sub_process TaskBeforeStarted
     end
 
   end
@@ -60,12 +86,20 @@ module TaskAfterCompleted
     extend Taskinator::Definition
     include Support
 
-    define_process :task_count do
-      for_each :iterator do
-        task :task1, :queue => :foo
-      end
+    define_process do
+      task :task1
+
+      after_completed :task_after_completed
+    end
+
+  end
+
+  module TaskAfterCompletedSubProcess
+    extend Taskinator::Definition
+    include Support
 
-      after_completed :task2, :queue => :foo
+    define_process do
+      sub_process TaskAfterCompleted
     end
 
   end
@@ -74,12 +108,20 @@ module TaskAfterFailed
     extend Taskinator::Definition
     include Support
 
-    define_process :task_count do
-      for_each :iterator do
-        task :task1, :queue => :foo
-      end
+    define_process do
+      task :task_fail
+
+      after_failed :task_after_failed
+    end
 
-      after_failed :task2, :queue => :foo
+  end
+
+  module TaskAfterFailedSubProcess
+    extend Taskinator::Definition
+    include Support
+
+    define_process do
+      sub_process TaskAfterFailed
     end
 
   end

spec/taskinator/persistence_spec.rb

@@ -501,6 +501,9 @@ def initialize
         TestDefinitions::EmptySequentialProcessTest,
         TestDefinitions::EmptyConcurrentProcessTest,
         TestDefinitions::NestedTask,
+        TestDefinitions::TaskBeforeStarted,
+        TestDefinitions::TaskAfterCompleted,
+        TestDefinitions::TaskAfterFailed,
       ].each do |definition|
 
         describe "#{definition.name} expire immediately" do

spec/taskinator/test_definitions_spec.rb

@@ -92,6 +92,52 @@
       Taskinator.queue_adapter = :test_queue_worker
     end
 
+    context "before_started" do
+      let(:definition) { TestDefinitions::TaskBeforeStarted }
+      subject { definition.create_process }
+
+      it "invokes before_started task" do
+        expect(subject.before_started_tasks.count).to eq(1)
+        expect_any_instance_of(definition).to receive(:task_before_started)
+
+        expect {
+          subject.enqueue!
+        }.to change { Taskinator.queue.tasks.length }.by(2)
+      end
+    end
+
+    context "after_completed" do
+      let(:definition) { TestDefinitions::TaskAfterCompleted }
+      subject { definition.create_process }
+
+      it "invokes after_completed task" do
+        expect(subject.after_completed_tasks.count).to eq(1)
+        expect_any_instance_of(definition).to receive(:task_after_completed)
+
+        expect {
+          subject.enqueue!
+        }.to change { Taskinator.queue.tasks.length }.by(2)
+      end
+    end
+
+    context "after_failed" do
+      let(:definition) { TestDefinitions::TaskAfterFailed }
+      subject { definition.create_process }
+
+      it "invokes after_failed task" do
+        expect(subject.after_failed_tasks.count).to eq(1)
+        expect_any_instance_of(definition).to receive(:task_after_failed)
+
+        expect {
+          begin
+            subject.enqueue!
+          rescue TestDefinitions::TestTaskFailed
+            # ignore error
+          end
+        }.to change { Taskinator.queue.tasks.length }.by(2)
+      end
+    end
+
     context "empty subprocesses" do
 
       context "sequential" do
@@ -133,6 +179,54 @@
       end
 
     end
+
+    context "subprocesses" do
+
+      context "before_started" do
+        let(:definition) { TestDefinitions::TaskBeforeStartedSubProcess }
+        subject { definition.create_process }
+
+        it "invokes before_started task" do
+          expect_any_instance_of(definition).to receive(:task_before_started)
+
+          expect {
+            subject.enqueue!
+          }.to change { Taskinator.queue.tasks.length }.by(1)
+        end
+      end
+
+      context "after_completed" do
+        let(:definition) { TestDefinitions::TaskAfterCompletedSubProcess }
+        subject { definition.create_process }
+
+        it "invokes after_completed task" do
+          expect_any_instance_of(definition).to receive(:task_after_completed)
+
+          expect {
+            subject.enqueue!
+          }.to change { Taskinator.queue.tasks.length }.by(2)
+        end
+      end
+
+      context "after_failed" do
+        let(:definition) { TestDefinitions::TaskAfterFailedSubProcess }
+        subject { definition.create_process }
+
+        it "invokes after_failed task" do
+          expect_any_instance_of(definition).to receive(:task_after_failed)
+
+          expect {
+            begin
+              subject.enqueue!
+            rescue TestDefinitions::TestTaskFailed
+              # ignore error
+            end
+          }.to change { Taskinator.queue.tasks.length }.by(2)
+        end
+      end
+
+    end
+
   end
 
   describe "statuses" do