eventbus

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 4, 2026 License: MIT Imports: 14 Imported by: 0

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

Constants

This section is empty.

Variables

This section is empty.

Functions

func DefaultBusConn 

func DefaultBusConn() (*nats.Conn, error)

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 GetBusURI 

func GetBusURI(instanceName string) (string, bool)

GetBusURI returns the broker URI for instanceName.

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 

func GetNATSConn(instanceName string) (*nats.Conn, bool)

GetNATSConn returns the cached *nats.Conn for instanceName, or false if absent.

func GetOrDialNATSConn 

func GetOrDialNATSConn(instanceName string) (*nats.Conn, error)

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 

func RegisterNATSConn(instanceName string, conn *nats.Conn)

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

View all 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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.