Documentation
¶
Overview ¶
Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations over NATS JetStream.
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 publishes an empty message to the JetStream reply subject (ack_token) supplied by step.eventbus.consume, which causes the broker to mark the message as acknowledged.
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 binds to an existing JetStream durable consumer (looked up by the durable name in ConsumeRequest.consumer) and fetches up to batch_size messages, waiting at most max_wait for the batch to fill.
Returned messages include ack_token = msg.Reply, which the caller passes to step.eventbus.ack to acknowledge each message individually.
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 publishes a single message to the NATS JetStream subject specified in PublishRequest.subject and returns the broker-assigned sequence number and acknowledgement timestamp.
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