Module: Legion::Extensions::Transformer::Runners::Transform

Extended by:
Helpers::Task
Includes:
Helpers::Lex
Defined in:
lib/legion/extensions/transformer/runners/transform.rb

Instance Method Summary collapse

Instance Method Details

#send_task(**opts) ⇒ Object 



53
54
55
56
57
58
59
60
 
# File 'lib/legion/extensions/transformer/runners/transform.rb', line 53

def send_task(**opts)
  payload = {}
  i[task_id relationship_id trigger_function_id runner_class function_id function chain_id debug args].each do |thing| # rubocop:disable Layout/LineLength
    payload[thing] = opts[thing] if opts.key? thing
  end

  Legion::Extensions::Transformer::Transport::Messages::Message.new(**payload).publish
end

#transform(transformation:, **payload) ⇒ Object 



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/legion/extensions/transformer/runners/transform.rb', line 6

def transform(transformation:, **payload)
  if transformation.include?('<%') || transformation.include?('%>')
    template = Tilt['erb'].new { transformation }
    variables = { **payload }
    variables[:crypt] = Legion::Crypt if transformation.include? 'crypt'
    variables[:settings] = Legion::Settings if transformation.include? 'settings'
    variables[:cache] = Legion::Cache if transformation.include? 'cache'
    if payload.key?(:task_id) && transformation.include?('task')
      variables[:task] = Legion::Data::Model::Task[payload[:task_id]]
    end

    payload[:args] = from_json(template.render(self, variables))
  else
    payload[:args] = from_json(transformation)
  end

  case payload[:args]
  when Hash
    unless payload[:task_id].nil?
      task_update(payload[:task_id], 'transformer.succeeded', function_args: payload[:args])
    end
    send_task(**payload)
    task_update(payload[:task_id], 'task.queued', use_database: false) unless payload[:task_id].nil?
  when Array
    payload[:args].each do |thing|
      new_payload = payload
      task = Legion::Runner::Status.generate_task_id(function_args: thing,
                                                     status:        'task.queued',
                                                     args:          thing,
                                                     **new_payload)
      new_payload[:task_id] = task[:task_id]
      new_payload[:args] = thing
      send_task(**new_payload)
    end
    unless payload[:task_id].nil?
      task_update(payload[:task_id],
                  'task.multiplied',
                  function_args: payload[:args])
    end
  end

  if payload[:debug] && payload.key?(:task_id)
    generate_task_log(task_id: payload[:task_id], function: 'transform', values: payload)
  end
  { success: true, **payload }
end