steps

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)

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.