Module: Legion::Extensions::Helpers::Task

Defined in:
lib/legion/extensions/helpers/task.rb

Instance Method Summary collapse

Instance Method Details

#generate_task_id(function_id:, status: 'task.queued', **opts) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
 
# File 'lib/legion/extensions/helpers/task.rb', line 27

def generate_task_id(function_id:, status: 'task.queued', **opts)
  insert = { status: status, function_id: function_id }
  insert[:payload] = Legion::JSON.dump(opts[:payload]) if opts.key? :payload
  insert[:function_args] = Legion::JSON.dump(opts[:args]) if opts.key? :args
  %i[master_id parent_id relationship_id task_id].each do |column|
    insert[column] = opts[column] if opts.key? column
  end

  { success: true, task_id: Legion::Data::Model::Task.insert(insert), **insert }
end

#generate_task_log(task_id:, runner_class: to_s, function:, **payload) ⇒ Object 



9
10
11
 
# File 'lib/legion/extensions/helpers/task.rb', line 9

def generate_task_log(task_id:, runner_class: to_s, function:, **payload)
  Legion::Transport::Messages::TaskLog.new(task_id: task_id, runner_class: runner_class, function: function, entry: payload).publish
end

#task_update(task_id, status, **opts) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/legion/extensions/helpers/task.rb', line 13

def task_update(task_id, status, **opts)
  return if task_id.nil? || status.nil?

  update_hash = { task_id: task_id, status: status }
  %i[results payload function_args payload results].each do |column|
    update_hash[column] = opts[column] if opts.key? column
  end
  Legion::Transport::Messages::TaskUpdate.new(**update_hash).publish
rescue StandardError => e
  log.fatal e.message
  log.fatal e.backtrace
  raise e
end