Table of Contents

Class WorkerBase<TInput, TOutput>

Namespace
Virtufin.Worker.DevKit
Assembly
Virtufin.Worker.DevKit.dll

Generic base class for Virtufin workers. The worker's typed input and output representations (TInput / TOutput) are bridged to the CloudEvent's Data payload via an ICloudEventCodec<TInput, TOutput> supplied to the constructor.

Subclasses override ProcessAsync(CloudEvent, TInput); the base class decodes the input CloudEvent's Data via the codec, calls the override, encodes the result back into a response CloudEvent (Id, Type, Source, correlationid, timestamp are applied automatically), and either returns it or null if the override returned null.

Format-agnostic: the base class itself makes no assumption about payload format. Format-specific subclasses (CommandWorker<TInput, TOutput, TCommand>, CommandWorker<T>) handle the format-specific bits on top of this base. 

public abstract class WorkerBase<TInput, TOutput> : IWorker

Type Parameters

TInput

Worker-typed input representation. Decoded from Data by the codec.

TOutput

Worker-typed output representation. Encoded by the codec into the response CloudEvent's Data.

Inheritance
WorkerBase<TInput, TOutput>
Implements
IWorker
Derived
CommandWorker<TInput, TOutput, TCommand>
Inherited Members

Constructors

WorkerBase(string, ICloudEventCodec<TInput, TOutput>)

A worker of type workerType, named by the naming library (WorkerTypeNames): source urn:com.virtufin:worker:<type>; responses com.virtufin.<type>.response on wrk.<type>.response; errors com.virtufin.<type>.error on wrk.<type>.error. An incoming replytopic changes the topic of both, never the ce-type.

protected WorkerBase(string workerType, ICloudEventCodec<TInput, TOutput> codec)

Parameters

workerType string
codec ICloudEventCodec<TInput, TOutput>

WorkerBase(Uri, string, ICloudEventCodec<TInput, TOutput>)

A worker with an explicit source and response ce-type; the response is published to the topic named by its ce-type (or replytopic). Prefer WorkerBase(string, ICloudEventCodec<TInput, TOutput>), which takes every name from the naming library.

protected WorkerBase(Uri source, string responseEventType, ICloudEventCodec<TInput, TOutput> codec)

Parameters

source Uri
responseEventType string
codec ICloudEventCodec<TInput, TOutput>

Fields

CorrelationIdAttribute

CloudEvent extension attribute key carrying the correlation id.

public const string CorrelationIdAttribute = "correlationid"

Field Value

string

FieldError

Response action used by Error(CloudEvent, string).

public const string FieldError = "error"

Field Value

string

PublishTopicAttribute

CloudEvent extension attribute a response can carry to publish to a topic other than its own ce-type. By default WorkManager publishes a response to the topic named by its own Type -- fine for most workers, where the two are the same value. Set this (via the topic parameter on BuildResponse(CloudEvent, TOutput, string?, string?, IReadOnlyDictionary<string, string>?)) when a response's ce-type must stay a stable identifier independent of a per-scenario topic -- e.g. the pubsub-topics spec's trading events (topic sc.<scenarioId>.trading.order.submitted, ce-type com.virtufin.trading.order.submitted). WorkManager strips this attribute from the envelope before publish; subscribers never see it.

public const string PublishTopicAttribute = "publishtopic"

Field Value

string

ReplyTopicAttribute

CloudEvent extension attribute key used to override the response topic.

public const string ReplyTopicAttribute = "replytopic"

Field Value

string

Properties

Names

The worker's names, when it was constructed from a worker type; otherwise null.

public WorkerTypeNames? Names { get; }

Property Value

WorkerTypeNames

Methods

BuildEnvelope(CloudEvent, TOutput)

Hook controlling how a non-null ProcessAsync(CloudEvent, TInput) result becomes the response envelope. Default delegates to BuildResponse(CloudEvent, TOutput, string?, string?, IReadOnlyDictionary<string, string>?) with no overrides (ce-type/topic driven by the worker's responseEventType/replytopic as before). Override when a worker needs a response topic that differs from its own ce-type, or extension attributes beyond correlationid/timestamp -- see StrategyWorkerBase (virtufin-strategy-devkit) for an example driven by the pubsub-topics spec's trading events.

protected virtual CloudEvent BuildEnvelope(CloudEvent input, TOutput result)

Parameters

input CloudEvent
result TOutput

Returns

CloudEvent

BuildResponse(CloudEvent, object, string, string?, string?, IReadOnlyDictionary<string, string>?)

Build a CloudEvent envelope from raw (data, contentType). Used by Error(CloudEvent, string) and any subclass that produces its own payload (e.g. bypassing the codec for a debug-only response). See the typed overload for parameter semantics.

protected CloudEvent BuildResponse(CloudEvent input, object data, string contentType, string? type = null, string? topic = null, IReadOnlyDictionary<string, string>? extensions = null)

Parameters

input CloudEvent
data object
contentType string
type string
topic string
extensions IReadOnlyDictionary<string, string>

Returns

CloudEvent

BuildResponse(CloudEvent, TOutput, string?, string?, IReadOnlyDictionary<string, string>?)

Build a CloudEvent envelope from a typed response. Id (always a new id -- never the input's), Source, the correlationid extension, and Time are populated automatically. Type defaults to the worker's responseEventType (or the incoming replytopic override) unless type is supplied.

protected CloudEvent BuildResponse(CloudEvent input, TOutput payload, string? type = null, string? topic = null, IReadOnlyDictionary<string, string>? extensions = null)

Parameters

input CloudEvent

The triggering CloudEvent.

payload TOutput

The typed response payload.

type string

Explicit ce-type override. Defaults to the incoming replytopic extension, else the worker's responseEventType.

topic string

Explicit publish-topic override (see PublishTopicAttribute) for when the response must route to a topic other than its own ce-type. Unset by default -- WorkManager falls back to Type as the topic, as before.

extensions IReadOnlyDictionary<string, string>

Additional CloudEvent extension attributes to stamp.

Returns

CloudEvent

Error(CloudEvent, string)

Build an error CloudEvent envelope. Encodes message via the codec's EncodeError(string) and applies the standard envelope construction.

protected CloudEvent Error(CloudEvent input, string message)

Parameters

input CloudEvent
message string

Returns

CloudEvent

ProcessAsync(CloudEvent)

IWorker entry point. ProcessAsync(CloudEvent) is sealed; subclasses implement the typed ProcessAsync(CloudEvent, TInput) instead.

public Task<CloudEvent?> ProcessAsync(CloudEvent input)

Parameters

input CloudEvent

Returns

Task<CloudEvent>

ProcessAsync(CloudEvent, TInput)

Subclasses override this. The base class already handled codec decode/encode and envelope construction; the override is the pure worker logic on typed inputs and outputs.

protected abstract Task<TOutput?> ProcessAsync(CloudEvent input, TInput payload)

Parameters

input CloudEvent
payload TInput

Returns

Task<TOutput>