Documentation
¶
Overview ¶
Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations dispatched through the providers.RuntimeBroker abstraction.
Index ¶
- Variables
- func AckHandler(ctx context.Context, ...) (*sdk.TypedStepResult[*eventbusv1.AckResponse], error)
- func All() []sdk.TypedStepProvider
- func ConsumeHandler(ctx context.Context, ...) (*sdk.TypedStepResult[*eventbusv1.ConsumeResponse], error)
- func PublishHandler(ctx context.Context, ...) (*sdk.TypedStepResult[*eventbusv1.PublishResponse], error)
Constants ¶
This section is empty.
Variables ¶
var AckFactory = sdk.NewTypedStepFactory( "step.eventbus.ack", &emptypb.Empty{}, &eventbusv1.AckRequest{}, AckHandler, )
AckFactory is the TypedStepProvider for step.eventbus.ack.
var ConsumeFactory = sdk.NewTypedStepFactory( "step.eventbus.consume", &emptypb.Empty{}, &eventbusv1.ConsumeRequest{}, ConsumeHandler, )
ConsumeFactory is the TypedStepProvider for step.eventbus.consume.
var PublishFactory = sdk.NewTypedStepFactory( "step.eventbus.publish", &emptypb.Empty{}, &eventbusv1.PublishRequest{}, PublishHandler, )
PublishFactory is the TypedStepProvider for step.eventbus.publish.
Functions ¶
func AckHandler ¶
func AckHandler( ctx context.Context, req sdk.TypedStepRequest[*emptypb.Empty, *eventbusv1.AckRequest], ) (*sdk.TypedStepResult[*eventbusv1.AckResponse], error)
AckHandler implements step.eventbus.ack. It dispatches the ack through the broker named by AckRequest.broker_ref via RuntimeBroker.Ack on the registered runtime/connection pair.
Legacy fallback: when broker_ref is empty AND exactly one broker module is registered, that broker is used automatically (see LookupRuntimeWithFallback).
func All ¶
func All() []sdk.TypedStepProvider
All returns all step TypedStepProvider factories exported by this package. Register each with the plugin's sdk.Server via WithTypedStepProvider.
func ConsumeHandler ¶
func ConsumeHandler( ctx context.Context, req sdk.TypedStepRequest[*emptypb.Empty, *eventbusv1.ConsumeRequest], ) (*sdk.TypedStepResult[*eventbusv1.ConsumeResponse], error)
ConsumeHandler implements step.eventbus.consume. It resolves the durable consumer's stream name from the in-process registry (populated by the eventbus.consumer module on Init), then dispatches the bounded-batch fetch through the broker named by ConsumeRequest.broker_ref via RuntimeBroker.Consume.
Returned messages include ack_token, which the caller passes to step.eventbus.ack to acknowledge each message individually.
Legacy fallback: when broker_ref is empty AND exactly one broker module is registered, that broker is used automatically (see LookupRuntimeWithFallback).
func PublishHandler ¶
func PublishHandler( ctx context.Context, req sdk.TypedStepRequest[*emptypb.Empty, *eventbusv1.PublishRequest], ) (*sdk.TypedStepResult[*eventbusv1.PublishResponse], error)
PublishHandler implements step.eventbus.publish. It dispatches the publish through the broker named by PublishRequest.broker_ref via LookupRuntime, then calls RuntimeBroker.Publish on the registered runtime/connection pair.
Legacy fallback: when broker_ref is empty AND exactly one broker module is registered, that broker is used automatically. The fallback exists so configs predating the broker_ref proto field continue to work in single-bus deployments; multi-bus workflows must set broker_ref explicitly.
The step config is empty (no per-step config required). All parameters are supplied via the typed input message.
Types ¶
This section is empty.
Source Files