Sha256: fae73755a161e7c9ddc4ddf93c863690ce931033d1e6ab7963fcc0b5b1c1c94e
Contents?: true
Size: 1.02 KB
Versions: 25
Compression:
Stored size: 1.02 KB
Contents
# frozen_string_literal: true
module Dynflow
module Executors
class Parallel
class Worker < Actor
def initialize(pool, transaction_adapter, telemetry_options = {})
@pool = Type! pool, Concurrent::Actor::Reference
@transaction_adapter = Type! transaction_adapter, TransactionAdapters::Abstract
@telemetry_options = telemetry_options
end
def on_message(work_item)
already_responded = false
Executors.run_user_code do
work_item.execute
end
rescue Errors::PersistenceError => e
@pool.tell([:handle_persistence_error, reference, e, work_item])
already_responded = true
ensure
Dynflow::Telemetry.with_instance { |t| t.increment_counter(:dynflow_worker_events, 1, @telemetry_options) }
if !already_responded && Concurrent.global_io_executor.running?
@pool.tell([:worker_done, reference, work_item])
end
end
end
end
end
end
Version data entries
25 entries across 25 versions & 1 rubygems