Skip to content

Bridge

ExecutorWorkerBase<TState> is the bridge between an IExecutor<TState, RichTradeAction, TradeEvent> and the WorkManager's worker contract (IWorker.ProcessAsync(CloudEvent)).

Topic binding

Bind to one topic per scenario the executor serves:

sc.<scenarioid>.trading.order.submitted

Not a wildcard -- the deployed Dapr pubsub component is Redis Streams, which has no wildcard subscription support. A worker instance MAY list multiple such topics (CreateWorkerRequest.topics) if one executor serves multiple scenarios (it carries no scenario-specific state, so this is safe for e.g. SimulatedExecutor); a live executor should only ever bind sc.LIVE.trading.order.submitted.

Dispatch

flowchart TD
    E[CloudEvent] --> T{ce-type?}
    T -->|"com.virtufin.trading.order.submitted"| DEC["OrderSubmissionJson.Decode"]
    DEC --> EXE["Executor.ExecuteAsync(state, action)"]
    EXE --> ENC["TradeEventJson.Encode"]
    ENC --> ENV["BuildEnvelope"]
    ENV --> RESP[response CloudEvent]
    T -->|other| N[ignored]
  • Anything but an "Order submitted" event is ignored (no response) -- a defensive check; in practice a worker only ever receives what it subscribed to, but nothing guarantees the topic never carries anything else.
  • OrderSubmissionJson.Decode parses the pubsub-topics spec's "Order submitted" payload (order_id, symbol, side, type, qty, price?, venue, submitted_at) into a RichTradeAction. Strategy attribution (strategy id, latency budget) isn't part of the wire payload and isn't meaningful to an executor -- both are placeholder values on the decoded action.
  • Executor.ExecuteAsync is the actual fill/routing logic -- everything upstream and downstream of this call is generic bridge plumbing, common to every executor.
  • TradeEventJson.Encode/Route map the resulting TradeEvent case to its Trade Lifecycle Events payload and topic/ce-type/ ce-subject. Only the cases reachable from one action-in/event-out call are mapped (OrderPlaced -> "Order accepted", FillReceived -> "Order filled", OrderCancelled, OrderExpired) -- the rest throw NotSupportedException rather than silently mis-route.

Response routing

Unlike a typical WorkerBase worker (which publishes on CloudEvent.Type, overridable per message via replytopic), the trading envelope keeps ce-type fixed at the event's own stable identifier (e.g. com.virtufin.trading.order.filled) regardless of any inbound override -- it's meant to be a stable identifier per the pubsub-topics spec, independent of the per-scenario topic. Routing instead goes through Virtufin.Worker.DevKit.WorkerBase.PublishTopicAttribute (publishtopic), set by BuildEnvelope to sc.<scenarioid>.trading.<event>; the WorkManager reads that extension to pick the publish topic and strips it before publish, so subscribers never see it (see virtufin-workmanager's WorkManager.PublishOutputAsync, 0.6.1+).

scenarioid is read back from the triggering "Order submitted" event's own extension (stamped there by StrategyWorkerBase) -- an executor never needs its own scenario config, it just echoes whatever scenario the order it's processing came from.

State

The executor's own TState lives in the worker instance (via ExecutorWorkerBase.State) -- the engine keeps one worker object alive for the instance's lifetime, so state threads across deliveries, same as StrategyWorkerBase.State/Portfolio on the producer side.

Limitations

One action in, one event out, matching IExecutor's own shape and WorkerBase.IWorker.ProcessAsync's one-response-per-trigger limit. An executor that needs to publish two events from one submitted order (e.g. an explicit "accepted" event and an immediate "filled" event) isn't supported today.