Sha256: 6d258f3cb9127e59c42e90d6c58c8f0ceac67082e2dd7d94b3a1432b0c2b6b75
Contents?: true
Size: 988 Bytes
Versions: 10
Compression:
Stored size: 988 Bytes
Contents
module Dynflow
module Executors
class Parallel < Abstract
class Worker < MicroActor
def initialize(pool, transaction_adapter)
super(pool.logger, pool, transaction_adapter)
end
private
def delayed_initialize(pool, transaction_adapter)
@pool = pool
@transaction_adapter = Type! transaction_adapter, TransactionAdapters::Abstract
end
def on_message(message)
match message,
(on Work::Step.(step: ~any) |
Work::Event.(step: ~any, event: Parallel::Event.(event: ~any)) do |step, event|
step.execute event
end),
(on Work::Finalize.(~any, any) do |sequential_manager|
sequential_manager.finalize
end)
@pool << WorkerDone[work: message, worker: self]
ensure
@transaction_adapter.cleanup
end
end
end
end
end
Version data entries
10 entries across 10 versions & 1 rubygems