Help: Running several task concurrently and one final task about taskinator

[virtualstaticvoid]

Hi @leoplct

Firstly, the ['a', 'b', 'c'].each do |report| part of the definition won't work as expected since each :work_step emitted won't have the report parameter.

  # ...

  ['a', 'b', 'c'].each do |report|
    task :work_step
  end

  # ...

Instead, you will need to use the for_each method, explained in the Data Driven Process Definitions section, to yield each report to be run by the work_step task.

  # ...

  for_each :list_of_reports do
    task :work_step
  end
  # ...

Where list_of_reports is an iterator method:

  # ...

  def list_of_reports
    yield ...
  end

  # ...

Secondly, in the implementation of the work_step and end methods, you don't need to re-enqueue the SyncReportJob and FinalActiveJob jobs (using perform_later), since these methods will already be executing as background jobs.

I.e.

  #...

  def work_step(report)
    p "REPORT: #{report}"
    SyncReportJob.perform(report)  # use perform instead, to execute immediately
  end

  #...

Thirdly, as you wanted, this process definition will result in each work_step tasks being executed concurrently and once they have all successfully completed, the final_step will be executed.

Note: I've rename end to be final_step so as not to be confused with with Ruby's end.

  # ...

  define_process do

    concurrent do
      for_each :report do
        task :work_step
      end
    end

    task :final_step

  end

  # ...

And the DAG will be:

    / work_step (a) \
   /                 \
o - - work_step (b) - - final_step - O
   \                 /
    \ work_step (c) /

Finally, I notice that the end step takes id as an argument, but this is never specified anywhere, so I imagine it will probably be passed in when the process gets defined, so it should be specified as a parameter:

  # ...

  define_process :id do

    # ...

  end

  # ...

Rewriting your example to make the changes, as follows:

module ReportsProcess
  extend Taskinator::Definition

  define_process :id do

    concurrent do
      for_each :list_of_reports do
        task :work_step
      end
    end

    task :final_step

  end

  def list_of_reports(id)
    ['a', 'b', 'c'].each do |r|
      yield r
    end
  end

  def work_step(report)
    p "REPORT: #{report}"
    SyncReportJob.perform(report)
  end

  def final_step(id)
    FinalActiveJob.perform(id)
  end

end

And then running this process, providing the id parameter:

id = 1234 # provide ID
process = ReportsProcess.create_process(id)
process.enqueue!

NOTE:

If you have already implemented SyncReportJob and FinalActiveJob as jobs you can use the job directive instead of task and taskinator will enqueue and process your jobs. See Reusing ActiveJob jobs for more details.

The code, refactored to use job instead:

module ReportsProcess
  extend Taskinator::Definition

  define_process :id do

    concurrent do
      for_each :report do
        job SyncReportJob
      end
    end

    job FinalActiveJob

  end

  def list_of_reports(id)
    ['a', 'b', 'c'].each do |r|
      yield r
    end
  end

end

Hope this helps!

[virtualstaticvoid]

Closing due to inactivity