README
¶
workflow-plugin-eventbus
Status: pre-pilot scaffold — Provider implementations are in progress (BMW E2E fulfillment pilot, PR 5).
A workflow external plugin that provisions durable event-bus clusters as IaC and exposes typed pipeline steps for publish/consume operations.
Providers
| Provider | Deploy targets |
|---|---|
nats |
DO App Platform, AWS ECS/EKS, Kubernetes StatefulSet |
kafka |
DO Managed Kafka, AWS MSK, Kubernetes (Strimzi) |
kinesis |
AWS (Kinesis Data Streams) |
Usage
Declare a cluster
modules:
- name: my-events
type: infra.eventbus
config:
provider: nats
deploy_target: digitalocean.app_platform
version: "2.10"
replicas: 2
jetstream:
enabled: true
max_storage_bytes: 53687091200 # 50 GB
Declare streams and consumers
- name: my-stream
type: infra.eventbus.stream
config:
name: MY_EVENTS
subjects: ["events.>"]
retention_policy: RETENTION_POLICY_LIMITS
max_bytes: 10737418240 # 10 GB
- name: my-consumer
type: infra.eventbus.consumer
config:
stream_name: MY_EVENTS
name: my-handler
filter_subject: "events.>"
ack_policy: ACK_POLICY_EXPLICIT
max_deliver: 5
Publish from a pipeline step
steps:
- name: publish
type: step.eventbus.publish
config:
subject: events.created
payload: '{{ toJson .input }}'
Subscribe trigger
my-handler:
trigger:
type: trigger.eventbus.subscribe
config:
stream_name: MY_EVENTS
name: my-handler
filter_subject: "events.>"
ack_policy: ACK_POLICY_EXPLICIT
steps:
- name: ack
type: step.eventbus.ack
config:
ack_token: '{{ .nats.message.ack_token }}'
Development
# Regenerate proto bindings after editing proto/eventbus.proto
make proto-gen
# Build
make build
# Test
make test
Planned providers
nats— NATS JetStream (in progress)kafka— stub (in progress)kinesis— stub (in progress)
Documentation
¶
Overview ¶
Package eventbus implements the workflow-plugin-eventbus plugin. It provides infra.eventbus, infra.eventbus.stream, and infra.eventbus.consumer module types plus step and trigger types for durable event-bus integration.
Index ¶
- func DefaultBusConn() (*nats.Conn, error)
- func GetBusURI(instanceName string) (string, bool)
- func GetCluster(instanceName string) (*eventbusv1.ClusterConfig, bool)
- func GetConsumer(instanceName string) (*eventbusv1.ConsumerConfig, bool)
- func GetConsumerByName(durableName string) (*eventbusv1.ConsumerConfig, bool)
- func GetNATSConn(instanceName string) (*nats.Conn, bool)
- func GetOrDialNATSConn(instanceName string) (*nats.Conn, error)
- func GetStream(instanceName string) (*eventbusv1.StreamConfig, bool)
- func NewClusterModule(instanceName string, cfg *eventbusv1.ClusterConfig) (sdk.ModuleInstance, error)
- func NewConsumerModule(instanceName string, cfg *eventbusv1.ConsumerConfig) (sdk.ModuleInstance, error)
- func NewStreamModule(instanceName string, cfg *eventbusv1.StreamConfig) (sdk.ModuleInstance, error)
- func NewSubscribeTrigger(instanceName string, cfg *eventbusv1.ConsumerConfig, cb sdk.TriggerCallback) (sdk.ModuleInstance, error)
- func RegisterBusURI(instanceName, uri string)
- func RegisterCluster(instanceName string, cfg *eventbusv1.ClusterConfig)
- func RegisterConsumer(instanceName string, cfg *eventbusv1.ConsumerConfig)
- func RegisterNATSConn(instanceName string, conn *nats.Conn)
- func RegisterStream(instanceName string, cfg *eventbusv1.StreamConfig)
- func UnregisterBusURI(instanceName string)
- func UnregisterCluster(instanceName string)
- func UnregisterConsumer(instanceName string)
- func UnregisterNATSConn(instanceName string)
- func UnregisterStream(instanceName string)
- type ClusterModuleFactory
- type ConsumerModuleFactory
- type StreamModuleFactory
- type SubscribeTriggerModuleFactory
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefaultBusConn ¶
DefaultBusConn returns a live NATS connection for the lexicographically first registered infra.eventbus module. Sorting ensures deterministic selection across invocations and concurrent goroutines, even when multiple buses are registered. For multi-bus workflows, use GetOrDialNATSConn(instanceName) directly.
func GetCluster ¶
func GetCluster(instanceName string) (*eventbusv1.ClusterConfig, bool)
GetCluster looks up a ClusterConfig by instance name.
func GetConsumer ¶
func GetConsumer(instanceName string) (*eventbusv1.ConsumerConfig, bool)
GetConsumer looks up a ConsumerConfig by instance name.
func GetConsumerByName ¶
func GetConsumerByName(durableName string) (*eventbusv1.ConsumerConfig, bool)
GetConsumerByName looks up a ConsumerConfig by its durable consumer name (cfg.name), iterating all registered instances. This is used by step.eventbus.consume to resolve the consumer config from the durable name supplied in ConsumeRequest.consumer.
func GetNATSConn ¶
GetNATSConn returns the cached *nats.Conn for instanceName, or false if absent.
func GetOrDialNATSConn ¶
GetOrDialNATSConn returns the cached NATS connection for instanceName, dialing a new one (via natsDialFn) if no live connection is cached. Returns an error if no URI is registered for instanceName or the dial fails.
Lock ordering: connCacheMu and urlMu (held inside GetBusURI) are never held simultaneously. The URI lookup happens between the fast-path unlock and the slow-path re-lock so that no nested acquisition is possible.
func GetStream ¶
func GetStream(instanceName string) (*eventbusv1.StreamConfig, bool)
GetStream looks up a StreamConfig by instance name.
func NewClusterModule ¶
func NewClusterModule(instanceName string, cfg *eventbusv1.ClusterConfig) (sdk.ModuleInstance, error)
NewClusterModule creates a clusterModule from a typed ClusterConfig proto.
Returns an error if:
- config.provider is empty or unknown
- config.deploy_target is empty or unsupported for the given provider
func NewConsumerModule ¶
func NewConsumerModule(instanceName string, cfg *eventbusv1.ConsumerConfig) (sdk.ModuleInstance, error)
NewConsumerModule creates a consumerModule from a typed ConsumerConfig proto.
Returns an error if:
- config.name is empty
- config.stream_name is empty
func NewStreamModule ¶
func NewStreamModule(instanceName string, cfg *eventbusv1.StreamConfig) (sdk.ModuleInstance, error)
NewStreamModule creates a streamModule from a typed StreamConfig proto.
Returns an error if:
- config.name is empty
- config.subjects contains no entries
func NewSubscribeTrigger ¶
func NewSubscribeTrigger(instanceName string, cfg *eventbusv1.ConsumerConfig, cb sdk.TriggerCallback) (sdk.ModuleInstance, error)
NewSubscribeTrigger creates a subscribeTrigger from a typed ConsumerConfig proto.
Returns an error if:
- config.name is empty
- config.stream_name is empty
func RegisterBusURI ¶
func RegisterBusURI(instanceName, uri string)
RegisterBusURI stores a broker URI under instanceName.
func RegisterCluster ¶
func RegisterCluster(instanceName string, cfg *eventbusv1.ClusterConfig)
RegisterCluster stores a ClusterConfig in the global registry under instanceName.
func RegisterConsumer ¶
func RegisterConsumer(instanceName string, cfg *eventbusv1.ConsumerConfig)
RegisterConsumer stores a ConsumerConfig in the global registry under instanceName.
func RegisterNATSConn ¶
RegisterNATSConn stores a live connection under instanceName. Exported so that integration tests and the trigger can pre-populate the cache.
func RegisterStream ¶
func RegisterStream(instanceName string, cfg *eventbusv1.StreamConfig)
RegisterStream stores a StreamConfig in the global registry under instanceName.
func UnregisterBusURI ¶
func UnregisterBusURI(instanceName string)
UnregisterBusURI removes the URI entry for instanceName.
func UnregisterCluster ¶
func UnregisterCluster(instanceName string)
UnregisterCluster removes a ClusterConfig from the registry.
func UnregisterConsumer ¶
func UnregisterConsumer(instanceName string)
UnregisterConsumer removes a ConsumerConfig from the registry.
func UnregisterNATSConn ¶
func UnregisterNATSConn(instanceName string)
UnregisterNATSConn removes the cached connection entry for instanceName without closing the connection. Use this in tests that manage the connection's lifetime separately (e.g., via nc.Close() + embedded-server shutdown).
func UnregisterStream ¶
func UnregisterStream(instanceName string)
UnregisterStream removes a StreamConfig from the registry.
Types ¶
type ClusterModuleFactory ¶
type ClusterModuleFactory struct{}
ClusterModuleFactory implements sdk.TypedModuleProvider for the infra.eventbus module type. The plugin wires this factory into CreateTypedModule.
func (*ClusterModuleFactory) CreateTypedModule ¶
func (f *ClusterModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)
CreateTypedModule unpacks the typed proto config and delegates to NewClusterModule.
func (*ClusterModuleFactory) TypedModuleTypes ¶
func (f *ClusterModuleFactory) TypedModuleTypes() []string
TypedModuleTypes returns the single module type served by this factory.
type ConsumerModuleFactory ¶
type ConsumerModuleFactory struct{}
ConsumerModuleFactory implements sdk.TypedModuleProvider for the infra.eventbus.consumer module type.
func (*ConsumerModuleFactory) CreateTypedModule ¶
func (f *ConsumerModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)
CreateTypedModule unpacks the typed proto config and delegates to NewConsumerModule.
func (*ConsumerModuleFactory) TypedModuleTypes ¶
func (f *ConsumerModuleFactory) TypedModuleTypes() []string
TypedModuleTypes returns the single module type served by this factory.
type StreamModuleFactory ¶
type StreamModuleFactory struct{}
StreamModuleFactory implements sdk.TypedModuleProvider for the infra.eventbus.stream module type.
func (*StreamModuleFactory) CreateTypedModule ¶
func (f *StreamModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)
CreateTypedModule unpacks the typed proto config and delegates to NewStreamModule.
func (*StreamModuleFactory) TypedModuleTypes ¶
func (f *StreamModuleFactory) TypedModuleTypes() []string
TypedModuleTypes returns the single module type served by this factory.
type SubscribeTriggerModuleFactory ¶
type SubscribeTriggerModuleFactory struct{}
SubscribeTriggerModuleFactory implements sdk.TypedModuleProvider for the trigger.eventbus.subscribe module type. The external plugin adapter calls CreateTypedModule with the trigger type name to instantiate triggers over gRPC.
func (*SubscribeTriggerModuleFactory) CreateTypedModule ¶
func (f *SubscribeTriggerModuleFactory) CreateTypedModule(typeName, name string, config *anypb.Any) (sdk.ModuleInstance, error)
CreateTypedModule unpacks the typed proto config and delegates to NewSubscribeTrigger. cb is always nil in the external gRPC subprocess path (the callback client is never wired in production SDK code); triggers that receive cb=nil behave as no-ops on Start, which is correct for IaC-only and plan/apply workflows.
func (*SubscribeTriggerModuleFactory) TypedModuleTypes ¶
func (f *SubscribeTriggerModuleFactory) TypedModuleTypes() []string
TypedModuleTypes returns the single trigger module type served by this factory.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
cmd
|
|
|
workflow-plugin-eventbus
command
Command workflow-plugin-eventbus is a workflow engine external plugin that provisions durable event-bus clusters (NATS / Kafka / Kinesis) as IaC and exposes typed pipeline steps for publish / consume operations.
|
Command workflow-plugin-eventbus is a workflow engine external plugin that provisions durable event-bus clusters (NATS / Kafka / Kinesis) as IaC and exposes typed pipeline steps for publish / consume operations. |
|
Package iac defines typed IaC primitives emitted by eventbus Providers.
|
Package iac defines typed IaC primitives emitted by eventbus Providers. |
|
Package providers defines the Provider interface and DeployTarget compatibility matrix for workflow-plugin-eventbus.
|
Package providers defines the Provider interface and DeployTarget compatibility matrix for workflow-plugin-eventbus. |
|
kafka
Package kafka provides a stub implementation of the kafka event-bus Provider.
|
Package kafka provides a stub implementation of the kafka event-bus Provider. |
|
kinesis
Package kinesis provides a stub implementation of the kinesis event-bus Provider.
|
Package kinesis provides a stub implementation of the kinesis event-bus Provider. |
|
nats
Package nats provides the NATS event-bus Provider implementation.
|
Package nats provides the NATS event-bus Provider implementation. |
|
Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations over NATS JetStream.
|
Package steps provides typed step handlers for the workflow-plugin-eventbus plugin: publish, consume, and ack operations over NATS JetStream. |