From dc2ab38b1666a70e322ec96317ad51201c160824 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:55:23 +0200 Subject: [PATCH 01/12] Add AWS SQS provider --- sqs_provider.go | 173 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) create mode 100644 sqs_provider.go diff --git a/sqs_provider.go b/sqs_provider.go new file mode 100644 index 0000000..9c91014 --- /dev/null +++ b/sqs_provider.go @@ -0,0 +1,173 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "time" + + "github.com/aws/aws-sdk-go-v2/service/sqs" +) + +type SQSAPI interface { + SendMessage(context.Context, *sqs.SendMessageInput, ...func(*sqs.Options)) (*sqs.SendMessageOutput, error) + ReceiveMessage(context.Context, *sqs.ReceiveMessageInput, ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error) + DeleteMessage(context.Context, *sqs.DeleteMessageInput, ...func(*sqs.Options)) (*sqs.DeleteMessageOutput, error) +} + +type SQSProviderConfig struct { + Client SQSAPI + QueueURL string + + WaitTimeSeconds int32 + VisibilityTimeout int32 + MaxMessages int32 + + FIFO bool + MessageGroupID string + MessageDeduplicationID func(Event) string + + Codec EventCodec +} + +type SQSProvider struct { + client SQSAPI + cfg SQSProviderConfig + codec EventCodec + closed atomic.Bool +} + +func WithSQS(config SQSProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewSQSProvider(config) }) +} + +func NewSQSProvider(config SQSProviderConfig) (*SQSProvider, error) { + if config.Client == nil { + return nil, errors.New("goeventbus: SQS client must not be nil") + } + if config.QueueURL == "" { + return nil, errors.New("goeventbus: SQS queue URL must not be empty") + } + if config.WaitTimeSeconds <= 0 { + config.WaitTimeSeconds = 20 + } + if config.WaitTimeSeconds > 20 { + config.WaitTimeSeconds = 20 + } + if config.MaxMessages <= 0 { + config.MaxMessages = 10 + } + if config.MaxMessages > 10 { + config.MaxMessages = 10 + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + if config.FIFO && config.MessageGroupID == "" { + config.MessageGroupID = "goeventbus" + } + return &SQSProvider{client: config.Client, cfg: config, codec: config.Codec}, nil +} + +func (p *SQSProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if _, err := remoteProjection(e); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + body := string(payload) + input := &sqs.SendMessageInput{ + QueueUrl: stringPtr(p.cfg.QueueURL), + MessageBody: &body, + } + if p.cfg.FIFO { + input.MessageGroupId = stringPtr(p.cfg.MessageGroupID) + if p.cfg.MessageDeduplicationID != nil { + if id := p.cfg.MessageDeduplicationID(e); id != "" { + input.MessageDeduplicationId = &id + } + } else if e.ID != "" { + input.MessageDeduplicationId = stringPtr(e.ID) + } + } + if _, err := p.client.SendMessage(ctx, input); err != nil { + return fmt.Errorf("goeventbus: publish SQS event: %w", err) + } + return nil +} + +func (p *SQSProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + + for { + if p.closed.Load() { + return ErrProviderClosed + } + if err := ctx.Err(); err != nil { + return err + } + out, err := p.client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: stringPtr(p.cfg.QueueURL), + MaxNumberOfMessages: p.cfg.MaxMessages, + WaitTimeSeconds: p.cfg.WaitTimeSeconds, + VisibilityTimeout: p.cfg.VisibilityTimeout, + }) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("goeventbus: receive SQS events: %w", err) + } + for _, msg := range out.Messages { + if msg.Body == nil || msg.ReceiptHandle == nil { + continue + } + event, err := p.codec.Decode([]byte(*msg.Body)) + if err != nil { + return fmt.Errorf("goeventbus: decode SQS event: %w", err) + } + if err := consumer(ctx, event); err != nil { + return err + } + if _, err := p.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: stringPtr(p.cfg.QueueURL), + ReceiptHandle: msg.ReceiptHandle, + }); err != nil { + return fmt.Errorf("goeventbus: delete SQS event after successful handling: %w", err) + } + } + if len(out.Messages) == 0 && p.cfg.WaitTimeSeconds == 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(100 * time.Millisecond): + } + } + } +} + +func (p *SQSProvider) Close() error { + p.closed.Store(true) + return nil +} + +func stringPtr(v string) *string { return &v } + +var _ Provider = (*SQSProvider)(nil) From f6d52b6705e4dc685e2142c33222cb8ef5aa749d Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:55:37 +0200 Subject: [PATCH 02/12] Add AWS SNS provider --- sns_provider.go | 112 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 112 insertions(+) create mode 100644 sns_provider.go diff --git a/sns_provider.go b/sns_provider.go new file mode 100644 index 0000000..eaaf906 --- /dev/null +++ b/sns_provider.go @@ -0,0 +1,112 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + + "github.com/aws/aws-sdk-go-v2/service/sns" +) + +type SNSAPI interface { + Publish(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) +} + +type SNSProviderConfig struct { + Client SNSAPI + TopicARN string + + FIFO bool + MessageGroupID string + MessageDeduplicationID func(Event) string + + // Subscriber is optional. SNS itself is publish-only; configure an SQS-backed + // Provider (or another supported subscriber transport) to enable Consume. + Subscriber Provider + Codec EventCodec +} + +type SNSProvider struct { + client SNSAPI + cfg SNSProviderConfig + codec EventCodec + closed atomic.Bool +} + +func WithSNS(config SNSProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewSNSProvider(config) }) +} + +func NewSNSProvider(config SNSProviderConfig) (*SNSProvider, error) { + if config.Client == nil { + return nil, errors.New("goeventbus: SNS client must not be nil") + } + if config.TopicARN == "" { + return nil, errors.New("goeventbus: SNS topic ARN must not be empty") + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + if config.FIFO && config.MessageGroupID == "" { + config.MessageGroupID = "goeventbus" + } + return &SNSProvider{client: config.Client, cfg: config, codec: config.Codec}, nil +} + +func (p *SNSProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if _, err := remoteProjection(e); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + message := string(payload) + input := &sns.PublishInput{TopicArn: stringPtr(p.cfg.TopicARN), Message: &message} + if p.cfg.FIFO { + input.MessageGroupId = stringPtr(p.cfg.MessageGroupID) + if p.cfg.MessageDeduplicationID != nil { + if id := p.cfg.MessageDeduplicationID(e); id != "" { + input.MessageDeduplicationId = &id + } + } else if e.ID != "" { + input.MessageDeduplicationId = stringPtr(e.ID) + } + } + if _, err := p.client.Publish(ctx, input); err != nil { + return fmt.Errorf("goeventbus: publish SNS event: %w", err) + } + return nil +} + +func (p *SNSProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if p.cfg.Subscriber == nil { + return errors.New("goeventbus: SNS consume requires a subscriber transport such as SQS") + } + return p.cfg.Subscriber.Consume(ctx, consumer) +} + +func (p *SNSProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + if p.cfg.Subscriber != nil { + return p.cfg.Subscriber.Close() + } + return nil +} + +var _ Provider = (*SNSProvider)(nil) From f315a57b3bdf344790cb67e35294042f7fa04f22 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:55:58 +0200 Subject: [PATCH 03/12] Add Google Cloud Pub/Sub provider --- gcp_pubsub_provider.go | 157 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 gcp_pubsub_provider.go diff --git a/gcp_pubsub_provider.go b/gcp_pubsub_provider.go new file mode 100644 index 0000000..a57f0f0 --- /dev/null +++ b/gcp_pubsub_provider.go @@ -0,0 +1,157 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + + "cloud.google.com/go/pubsub" +) + +type GCPPubSubProviderConfig struct { + Client *pubsub.Client + Topic string + Subscription string + + // EnableOrdering enables Pub/Sub message ordering. OrderingKey defaults to + // the event projection when this is true. + EnableOrdering bool + OrderingKey func(Event) string + + Codec EventCodec +} + +type GCPPubSubProvider struct { + topic *pubsub.Topic + subscription *pubsub.Subscription + codec EventCodec + orderingKey func(Event) string + closed atomic.Bool +} + +func WithGCPPubSub(config GCPPubSubProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewGCPPubSubProvider(config) }) +} + +func NewGCPPubSubProvider(config GCPPubSubProviderConfig) (*GCPPubSubProvider, error) { + if config.Client == nil { + return nil, errors.New("goeventbus: GCP Pub/Sub client must not be nil") + } + if config.Topic == "" { + return nil, errors.New("goeventbus: GCP Pub/Sub topic must not be empty") + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + topic := config.Client.Topic(config.Topic) + topic.EnableMessageOrdering = config.EnableOrdering + + var subscription *pubsub.Subscription + if config.Subscription != "" { + subscription = config.Client.Subscription(config.Subscription) + } + + return &GCPPubSubProvider{ + topic: topic, + subscription: subscription, + codec: config.Codec, + orderingKey: config.OrderingKey, + }, nil +} + +func (p *GCPPubSubProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + + msg := &pubsub.Message{ + Data: payload, + Attributes: map[string]string{ + "goeventbus-projection": projection, + "goeventbus-event-id": e.ID, + }, + } + if p.topic.EnableMessageOrdering { + if p.orderingKey != nil { + msg.OrderingKey = p.orderingKey(e) + } else { + msg.OrderingKey = projection + } + } + + if _, err := p.topic.Publish(ctx, msg).Get(ctx); err != nil { + return fmt.Errorf("goeventbus: publish GCP Pub/Sub event: %w", err) + } + return nil +} + +func (p *GCPPubSubProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if p.subscription == nil { + return errors.New("goeventbus: GCP Pub/Sub subscription is required to consume events") + } + if ctx == nil { + ctx = context.Background() + } + + err := p.subscription.Receive(ctx, func(messageCtx context.Context, msg *pubsub.Message) { + if p.closed.Load() { + msg.Nack() + return + } + event, err := p.codec.Decode(msg.Data) + if err != nil { + msg.Nack() + return + } + if err := consumer(messageCtx, event); err != nil { + msg.Nack() + return + } + msg.Ack() + }) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: consume GCP Pub/Sub events: %w", err) + } + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return nil +} + +func (p *GCPPubSubProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.topic.Stop() + return nil +} + +var _ Provider = (*GCPPubSubProvider)(nil) From 0f75136df0f574352e616226062bdb12f15d1ac8 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:56:23 +0200 Subject: [PATCH 04/12] Add Azure Service Bus provider --- azure_servicebus_provider.go | 167 +++++++++++++++++++++++++++++++++++ 1 file changed, 167 insertions(+) create mode 100644 azure_servicebus_provider.go diff --git a/azure_servicebus_provider.go b/azure_servicebus_provider.go new file mode 100644 index 0000000..4a105b4 --- /dev/null +++ b/azure_servicebus_provider.go @@ -0,0 +1,167 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + + "github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus" +) + +type AzureServiceBusProviderConfig struct { + Client *azservicebus.Client + + // Configure Queue for queue semantics, or Topic for topic publishing. + // Topic consumption additionally requires Subscription. + Queue string + Topic string + Subscription string + + Codec EventCodec +} + +type AzureServiceBusProvider struct { + client *azservicebus.Client + sender *azservicebus.Sender + receiver *azservicebus.Receiver + codec EventCodec + + closeMu sync.Mutex + closed atomic.Bool +} + +func WithAzureServiceBus(config AzureServiceBusProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewAzureServiceBusProvider(config) }) +} + +func NewAzureServiceBusProvider(config AzureServiceBusProviderConfig) (*AzureServiceBusProvider, error) { + if config.Client == nil { + return nil, errors.New("goeventbus: Azure Service Bus client must not be nil") + } + if config.Queue == "" && config.Topic == "" { + return nil, errors.New("goeventbus: Azure Service Bus queue or topic is required") + } + if config.Queue != "" && config.Topic != "" { + return nil, errors.New("goeventbus: Azure Service Bus queue and topic are mutually exclusive") + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + entity := config.Queue + if entity == "" { + entity = config.Topic + } + sender, err := config.Client.NewSender(entity, nil) + if err != nil { + return nil, fmt.Errorf("goeventbus: create Azure Service Bus sender: %w", err) + } + + var receiver *azservicebus.Receiver + if config.Queue != "" { + receiver, err = config.Client.NewReceiverForQueue(config.Queue, &azservicebus.ReceiverOptions{ + ReceiveMode: azservicebus.ReceiveModePeekLock, + }) + } else if config.Subscription != "" { + receiver, err = config.Client.NewReceiverForSubscription(config.Topic, config.Subscription, &azservicebus.ReceiverOptions{ + ReceiveMode: azservicebus.ReceiveModePeekLock, + }) + } + if err != nil { + _ = sender.Close(context.Background()) + return nil, fmt.Errorf("goeventbus: create Azure Service Bus receiver: %w", err) + } + + return &AzureServiceBusProvider{ + client: config.Client, + sender: sender, + receiver: receiver, + codec: config.Codec, + }, nil +} + +func (p *AzureServiceBusProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if _, err := remoteProjection(e); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + if err := p.sender.SendMessage(ctx, &azservicebus.Message{Body: payload}, nil); err != nil { + return fmt.Errorf("goeventbus: publish Azure Service Bus event: %w", err) + } + return nil +} + +func (p *AzureServiceBusProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if p.receiver == nil { + return errors.New("goeventbus: Azure Service Bus receiver is not configured") + } + if ctx == nil { + ctx = context.Background() + } + + for { + if p.closed.Load() { + return ErrProviderClosed + } + messages, err := p.receiver.ReceiveMessages(ctx, 1, nil) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: receive Azure Service Bus event: %w", err) + } + for _, msg := range messages { + event, err := p.codec.Decode(msg.Body) + if err != nil { + _ = p.receiver.AbandonMessage(ctx, msg, nil) + return fmt.Errorf("goeventbus: decode Azure Service Bus event: %w", err) + } + if err := consumer(ctx, event); err != nil { + _ = p.receiver.AbandonMessage(ctx, msg, nil) + return err + } + if err := p.receiver.CompleteMessage(ctx, msg, nil); err != nil { + return fmt.Errorf("goeventbus: complete Azure Service Bus event: %w", err) + } + } + } +} + +func (p *AzureServiceBusProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.closeMu.Lock() + defer p.closeMu.Unlock() + ctx := context.Background() + var errs []error + if p.receiver != nil { + errs = append(errs, p.receiver.Close(ctx)) + } + if p.sender != nil { + errs = append(errs, p.sender.Close(ctx)) + } + return errors.Join(errs...) +} + +var _ Provider = (*AzureServiceBusProvider)(nil) From 8b107f9d32ef98c12018744d6b9a71c8a72a5ea6 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:56:53 +0200 Subject: [PATCH 05/12] Add Apache Pulsar provider --- pulsar_provider.go | 191 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 pulsar_provider.go diff --git a/pulsar_provider.go b/pulsar_provider.go new file mode 100644 index 0000000..0b09ae1 --- /dev/null +++ b/pulsar_provider.go @@ -0,0 +1,191 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + + "github.com/apache/pulsar-client-go/pulsar" +) + +type PulsarProviderConfig struct { + Client pulsar.Client + Topic string + Subscription string + + // SubscriptionType defaults to Shared. Set explicitly for Exclusive, + // Failover, or KeyShared behavior. + SubscriptionType *pulsar.SubscriptionType + Codec EventCodec +} + +type PulsarProvider struct { + client pulsar.Client + cfg PulsarProviderConfig + codec EventCodec + + mu sync.Mutex + producer pulsar.Producer + consumer pulsar.Consumer + closed atomic.Bool +} + +func WithPulsar(config PulsarProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewPulsarProvider(config) }) +} + +func NewPulsarProvider(config PulsarProviderConfig) (*PulsarProvider, error) { + if config.Client == nil { + return nil, errors.New("goeventbus: Pulsar client must not be nil") + } + if config.Topic == "" { + return nil, errors.New("goeventbus: Pulsar topic must not be empty") + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + return &PulsarProvider{client: config.Client, cfg: config, codec: config.Codec}, nil +} + +func (p *PulsarProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + producer, err := p.producerForUse() + if err != nil { + return err + } + _, err = producer.Send(ctx, &pulsar.ProducerMessage{ + Payload: payload, + Key: projection, + Properties: map[string]string{ + "goeventbus-event-id": e.ID, + "goeventbus-projection": projection, + }, + }) + if err != nil { + return fmt.Errorf("goeventbus: publish Pulsar event: %w", err) + } + return nil +} + +func (p *PulsarProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if p.cfg.Subscription == "" { + return errors.New("goeventbus: Pulsar subscription is required to consume events") + } + if ctx == nil { + ctx = context.Background() + } + + pc, err := p.consumerForUse() + if err != nil { + return err + } + for { + if p.closed.Load() { + return ErrProviderClosed + } + msg, err := pc.Receive(ctx) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: receive Pulsar event: %w", err) + } + event, err := p.codec.Decode(msg.Payload()) + if err != nil { + pc.Nack(msg) + return fmt.Errorf("goeventbus: decode Pulsar event: %w", err) + } + if err := consumer(ctx, event); err != nil { + pc.Nack(msg) + return err + } + if err := pc.Ack(msg); err != nil { + return fmt.Errorf("goeventbus: acknowledge Pulsar event: %w", err) + } + } +} + +func (p *PulsarProvider) producerForUse() (pulsar.Producer, error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.closed.Load() { + return nil, ErrProviderClosed + } + if p.producer != nil { + return p.producer, nil + } + producer, err := p.client.CreateProducer(pulsar.ProducerOptions{Topic: p.cfg.Topic}) + if err != nil { + return nil, fmt.Errorf("goeventbus: create Pulsar producer: %w", err) + } + p.producer = producer + return producer, nil +} + +func (p *PulsarProvider) consumerForUse() (pulsar.Consumer, error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.closed.Load() { + return nil, ErrProviderClosed + } + if p.consumer != nil { + return p.consumer, nil + } + subType := pulsar.Shared + if p.cfg.SubscriptionType != nil { + subType = *p.cfg.SubscriptionType + } + consumer, err := p.client.Subscribe(pulsar.ConsumerOptions{ + Topic: p.cfg.Topic, + SubscriptionName: p.cfg.Subscription, + Type: subType, + }) + if err != nil { + return nil, fmt.Errorf("goeventbus: create Pulsar consumer: %w", err) + } + p.consumer = consumer + return consumer, nil +} + +func (p *PulsarProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.mu.Lock() + defer p.mu.Unlock() + if p.consumer != nil { + p.consumer.Close() + p.consumer = nil + } + if p.producer != nil { + p.producer.Close() + p.producer = nil + } + return nil +} + +var _ Provider = (*PulsarProvider)(nil) From 90ad7d495b11b81a9a3ed345edc02bcf30ce5af2 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:57:13 +0200 Subject: [PATCH 06/12] Add MQTT provider --- mqtt_provider.go | 203 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 203 insertions(+) create mode 100644 mqtt_provider.go diff --git a/mqtt_provider.go b/mqtt_provider.go new file mode 100644 index 0000000..cb15047 --- /dev/null +++ b/mqtt_provider.go @@ -0,0 +1,203 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +type MQTTProviderConfig struct { + // Client can be injected. If nil, Options is used to construct and own one. + Client mqtt.Client + Options *mqtt.ClientOptions + + Topic string + TopicPrefix string + QoS byte + Retained bool + + OperationTimeout time.Duration + Codec EventCodec +} + +type MQTTProvider struct { + client mqtt.Client + owned bool + cfg MQTTProviderConfig + codec EventCodec + + connectMu sync.Mutex + closed atomic.Bool +} + +func WithMQTT(config MQTTProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewMQTTProvider(config) }) +} + +func NewMQTTProvider(config MQTTProviderConfig) (*MQTTProvider, error) { + if config.Client == nil && config.Options == nil { + return nil, errors.New("goeventbus: MQTT client or client options are required") + } + if config.QoS > 2 { + return nil, errors.New("goeventbus: MQTT QoS must be 0, 1, or 2") + } + if config.Topic == "" && config.TopicPrefix == "" { + return nil, errors.New("goeventbus: MQTT topic or topic prefix is required") + } + if config.OperationTimeout <= 0 { + config.OperationTimeout = 30 * time.Second + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + client := config.Client + owned := false + if client == nil { + client = mqtt.NewClient(config.Options) + owned = true + } + return &MQTTProvider{client: client, owned: owned, cfg: config, codec: config.Codec}, nil +} + +func (p *MQTTProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + if err := p.ensureConnected(ctx); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + topic := p.cfg.Topic + if p.cfg.TopicPrefix != "" { + topic = p.cfg.TopicPrefix + projection + } + token := p.client.Publish(topic, p.cfg.QoS, p.cfg.Retained, payload) + if err := waitMQTTToken(ctx, token, p.cfg.OperationTimeout); err != nil { + return fmt.Errorf("goeventbus: publish MQTT event: %w", err) + } + return nil +} + +func (p *MQTTProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if err := p.ensureConnected(ctx); err != nil { + return err + } + + topic := p.cfg.Topic + if p.cfg.TopicPrefix != "" { + topic = p.cfg.TopicPrefix + "#" + } + if strings.TrimSpace(topic) == "" { + return errors.New("goeventbus: MQTT subscription topic is empty") + } + + errCh := make(chan error, 1) + token := p.client.Subscribe(topic, p.cfg.QoS, func(_ mqtt.Client, msg mqtt.Message) { + if p.closed.Load() { + return + } + event, err := p.codec.Decode(msg.Payload()) + if err == nil { + err = consumer(ctx, event) + } + if err != nil { + select { + case errCh <- err: + default: + } + } + }) + if err := waitMQTTToken(ctx, token, p.cfg.OperationTimeout); err != nil { + return fmt.Errorf("goeventbus: subscribe MQTT topic %q: %w", topic, err) + } + defer func() { + if p.client.IsConnectionOpen() { + _ = waitMQTTToken(context.Background(), p.client.Unsubscribe(topic), p.cfg.OperationTimeout) + } + }() + + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-errCh: + return err + } +} + +func (p *MQTTProvider) ensureConnected(ctx context.Context) error { + if p.client.IsConnectionOpen() { + return nil + } + p.connectMu.Lock() + defer p.connectMu.Unlock() + if p.client.IsConnectionOpen() { + return nil + } + if p.closed.Load() { + return ErrProviderClosed + } + if err := waitMQTTToken(ctx, p.client.Connect(), p.cfg.OperationTimeout); err != nil { + return fmt.Errorf("goeventbus: connect MQTT client: %w", err) + } + return nil +} + +func (p *MQTTProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + if p.owned && p.client.IsConnectionOpen() { + p.client.Disconnect(250) + } + return nil +} + +func waitMQTTToken(ctx context.Context, token mqtt.Token, maxWait time.Duration) error { + if token == nil { + return errors.New("MQTT operation returned a nil token") + } + deadline := time.NewTimer(maxWait) + defer deadline.Stop() + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + for { + if token.WaitTimeout(0) { + return token.Error() + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-deadline.C: + return errors.New("MQTT operation timed out") + case <-ticker.C: + } + } +} + +var _ Provider = (*MQTTProvider)(nil) From bc9927b665179efbbf4badbd79013e4ab226fa05 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:58:09 +0200 Subject: [PATCH 07/12] Add PostgreSQL provider --- postgres_provider.go | 251 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 postgres_provider.go diff --git a/postgres_provider.go b/postgres_provider.go new file mode 100644 index 0000000..2285573 --- /dev/null +++ b/postgres_provider.go @@ -0,0 +1,251 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "regexp" + "sync" + "sync/atomic" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var postgresIdentifierRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) + +type PostgresProviderConfig struct { + // Pool can be injected and remains application-owned. If Pool is nil, + // ConnectionString is used to create a pool owned by the provider. + Pool *pgxpool.Pool + ConnectionString string + Table string + + PollInterval time.Duration + RetryDelay time.Duration + Codec EventCodec +} + +type PostgresProvider struct { + pool *pgxpool.Pool + owned bool + table string + codec EventCodec + + pollInterval time.Duration + retryDelay time.Duration + + schemaOnce sync.Once + schemaErr error + closed atomic.Bool +} + +func WithPostgres(config PostgresProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewPostgresProvider(config) }) +} + +func NewPostgresProvider(config PostgresProviderConfig) (*PostgresProvider, error) { + if config.Pool == nil && config.ConnectionString == "" { + return nil, errors.New("goeventbus: PostgreSQL pool or connection string is required") + } + if config.Table == "" { + config.Table = "goeventbus_events" + } + if !postgresIdentifierRE.MatchString(config.Table) { + return nil, fmt.Errorf("goeventbus: invalid PostgreSQL table name %q", config.Table) + } + if config.PollInterval <= 0 { + config.PollInterval = 250 * time.Millisecond + } + if config.RetryDelay <= 0 { + config.RetryDelay = time.Second + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + pool := config.Pool + owned := false + if pool == nil { + var err error + pool, err = pgxpool.New(context.Background(), config.ConnectionString) + if err != nil { + return nil, fmt.Errorf("goeventbus: create PostgreSQL pool: %w", err) + } + owned = true + } + + return &PostgresProvider{ + pool: pool, + owned: owned, + table: `"` + config.Table + `"`, + codec: config.Codec, + pollInterval: config.PollInterval, + retryDelay: config.RetryDelay, + }, nil +} + +func (p *PostgresProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + if err := p.ensureSchema(ctx); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + query := fmt.Sprintf(`INSERT INTO %s (event_id, projection, payload) VALUES ($1, $2, $3)`, p.table) + if _, err := p.pool.Exec(ctx, query, e.ID, projection, payload); err != nil { + return fmt.Errorf("goeventbus: publish PostgreSQL event: %w", err) + } + return nil +} + +func (p *PostgresProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if err := p.ensureSchema(ctx); err != nil { + return err + } + + for { + if p.closed.Load() { + return ErrProviderClosed + } + handled, err := p.consumeOne(ctx, consumer) + if err != nil { + return err + } + if handled { + continue + } + timer := time.NewTimer(p.pollInterval) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} + +func (p *PostgresProvider) consumeOne(ctx context.Context, consumer EventConsumer) (bool, error) { + tx, err := p.pool.Begin(ctx) + if err != nil { + return false, fmt.Errorf("goeventbus: begin PostgreSQL consume transaction: %w", err) + } + defer tx.Rollback(context.Background()) + + var ( + id int64 + payload []byte + ) + query := fmt.Sprintf(` + SELECT id, payload + FROM %s + WHERE available_at <= NOW() + ORDER BY id + FOR UPDATE SKIP LOCKED + LIMIT 1`, p.table) + err = tx.QueryRow(ctx, query).Scan(&id, &payload) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("goeventbus: select PostgreSQL event: %w", err) + } + + event, decodeErr := p.codec.Decode(payload) + if decodeErr != nil { + if err := p.deferPostgresEvent(ctx, tx, id); err != nil { + return true, err + } + return true, fmt.Errorf("goeventbus: decode PostgreSQL event: %w", decodeErr) + } + if err := consumer(ctx, event); err != nil { + if deferErr := p.deferPostgresEvent(ctx, tx, id); deferErr != nil { + return true, errors.Join(err, deferErr) + } + return true, err + } + + deleteQuery := fmt.Sprintf(`DELETE FROM %s WHERE id = $1`, p.table) + if _, err := tx.Exec(ctx, deleteQuery, id); err != nil { + return true, fmt.Errorf("goeventbus: delete PostgreSQL event: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return true, fmt.Errorf("goeventbus: commit PostgreSQL consume transaction: %w", err) + } + return true, nil +} + +func (p *PostgresProvider) deferPostgresEvent(ctx context.Context, tx pgx.Tx, id int64) error { + query := fmt.Sprintf(` + UPDATE %s + SET attempts = attempts + 1, + available_at = NOW() + ($2 * INTERVAL '1 millisecond') + WHERE id = $1`, p.table) + delayMS := p.retryDelay.Milliseconds() + if _, err := tx.Exec(ctx, query, id, delayMS); err != nil { + return fmt.Errorf("goeventbus: defer failed PostgreSQL event: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("goeventbus: commit PostgreSQL retry state: %w", err) + } + return nil +} + +func (p *PostgresProvider) ensureSchema(ctx context.Context) error { + p.schemaOnce.Do(func() { + query := fmt.Sprintf(` + CREATE TABLE IF NOT EXISTS %s ( + id BIGSERIAL PRIMARY KEY, + event_id TEXT NOT NULL DEFAULT '', + projection TEXT NOT NULL, + payload BYTEA NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + available_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX IF NOT EXISTS %s_available_idx + ON %s (available_at, id)`, + p.table, + p.table[1:len(p.table)-1], + p.table, + ) + _, p.schemaErr = p.pool.Exec(ctx, query) + if p.schemaErr != nil { + p.schemaErr = fmt.Errorf("goeventbus: initialize PostgreSQL event table: %w", p.schemaErr) + } + }) + return p.schemaErr +} + +func (p *PostgresProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + if p.owned { + p.pool.Close() + } + return nil +} + +var _ Provider = (*PostgresProvider)(nil) From 217a38f1902d3f98f1afc1adc95e7dd6368a216b Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:58:12 +0200 Subject: [PATCH 08/12] Add NSQ provider --- nsq_provider.go | 192 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 nsq_provider.go diff --git a/nsq_provider.go b/nsq_provider.go new file mode 100644 index 0000000..5b4c5eb --- /dev/null +++ b/nsq_provider.go @@ -0,0 +1,192 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + + "github.com/nsqio/go-nsq" +) + +type NSQProviderConfig struct { + NSQDAddress string + LookupdAddresses []string + Topic string + Channel string + Config *nsq.Config + Codec EventCodec +} + +type NSQProvider struct { + cfg NSQProviderConfig + codec EventCodec + + mu sync.Mutex + producer *nsq.Producer + consumer *nsq.Consumer + closed atomic.Bool +} + +func WithNSQ(config NSQProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { return NewNSQProvider(config) }) +} + +func NewNSQProvider(config NSQProviderConfig) (*NSQProvider, error) { + if config.NSQDAddress == "" { + return nil, errors.New("goeventbus: NSQ nsqd address must not be empty") + } + if config.Topic == "" { + return nil, errors.New("goeventbus: NSQ topic must not be empty") + } + if config.Config == nil { + config.Config = nsq.NewConfig() + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + return &NSQProvider{cfg: config, codec: config.Codec}, nil +} + +func (p *NSQProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + if err := ctx.Err(); err != nil { + return err + } + if _, err := remoteProjection(e); err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + producer, err := p.producerForUse() + if err != nil { + return err + } + if err := producer.Publish(p.cfg.Topic, payload); err != nil { + return fmt.Errorf("goeventbus: publish NSQ event: %w", err) + } + return nil +} + +func (p *NSQProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if p.cfg.Channel == "" { + return errors.New("goeventbus: NSQ channel is required to consume events") + } + if ctx == nil { + ctx = context.Background() + } + + nc, err := nsq.NewConsumer(p.cfg.Topic, p.cfg.Channel, p.cfg.Config) + if err != nil { + return fmt.Errorf("goeventbus: create NSQ consumer: %w", err) + } + nc.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error { + event, err := p.codec.Decode(message.Body) + if err != nil { + return fmt.Errorf("goeventbus: decode NSQ event: %w", err) + } + return consumer(ctx, event) + })) + + p.mu.Lock() + if p.closed.Load() { + p.mu.Unlock() + nc.Stop() + return ErrProviderClosed + } + if p.consumer != nil { + p.mu.Unlock() + nc.Stop() + return errors.New("goeventbus: NSQ provider already has an active consumer") + } + p.consumer = nc + p.mu.Unlock() + + if len(p.cfg.LookupdAddresses) > 0 { + err = nc.ConnectToNSQLookupds(p.cfg.LookupdAddresses) + } else { + err = nc.ConnectToNSQD(p.cfg.NSQDAddress) + } + if err != nil { + p.clearNSQConsumer(nc) + nc.Stop() + return fmt.Errorf("goeventbus: connect NSQ consumer: %w", err) + } + + select { + case <-ctx.Done(): + nc.Stop() + <-nc.StopChan + p.clearNSQConsumer(nc) + return ctx.Err() + case <-nc.StopChan: + p.clearNSQConsumer(nc) + if p.closed.Load() { + return ErrProviderClosed + } + if ctx.Err() != nil { + return ctx.Err() + } + return errors.New("goeventbus: NSQ consumer stopped") + } +} + +func (p *NSQProvider) producerForUse() (*nsq.Producer, error) { + p.mu.Lock() + defer p.mu.Unlock() + if p.closed.Load() { + return nil, ErrProviderClosed + } + if p.producer != nil { + return p.producer, nil + } + producer, err := nsq.NewProducer(p.cfg.NSQDAddress, p.cfg.Config) + if err != nil { + return nil, fmt.Errorf("goeventbus: create NSQ producer: %w", err) + } + p.producer = producer + return producer, nil +} + +func (p *NSQProvider) clearNSQConsumer(consumer *nsq.Consumer) { + p.mu.Lock() + if p.consumer == consumer { + p.consumer = nil + } + p.mu.Unlock() +} + +func (p *NSQProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.mu.Lock() + consumer := p.consumer + producer := p.producer + p.consumer = nil + p.producer = nil + p.mu.Unlock() + if consumer != nil { + consumer.Stop() + } + if producer != nil { + producer.Stop() + } + return nil +} + +var _ Provider = (*NSQProvider)(nil) From 425bd4e345dad5c73e3baabdbf95ba94ae29a93e Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:58:46 +0200 Subject: [PATCH 09/12] Add dependencies for remaining broker providers --- go.mod | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/go.mod b/go.mod index 635f0d1..e0bddf9 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,15 @@ go 1.23.0 toolchain go1.23.7 require ( + cloud.google.com/go/pubsub v1.48.0 + github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus v1.10.0 + github.com/apache/pulsar-client-go v0.16.0 + github.com/aws/aws-sdk-go-v2/service/sns v1.39.17 + github.com/aws/aws-sdk-go-v2/service/sqs v1.52.1 + github.com/eclipse/paho.mqtt.golang v1.5.1 github.com/fasthttp/router v1.5.4 + github.com/jackc/pgx/v5 v5.7.6 + github.com/nsqio/go-nsq v1.1.0 github.com/nats-io/nats.go v1.39.1 github.com/rabbitmq/amqp091-go v1.14.0 github.com/redis/go-redis/v9 v9.18.0 From 68092d2ce4233001ad5d145b3a375d58574988bd Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:59:03 +0200 Subject: [PATCH 10/12] Test remaining broker option validation --- broker_options_test.go | 53 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/broker_options_test.go b/broker_options_test.go index 5f1ffd5..10c73b7 100644 --- a/broker_options_test.go +++ b/broker_options_test.go @@ -3,6 +3,8 @@ package GoEventBus import ( "errors" "testing" + + mqtt "github.com/eclipse/paho.mqtt.golang" ) func TestWithNATSJetStreamRejectsNilConnectionLazily(t *testing.T) { @@ -63,3 +65,54 @@ func TestKafkaRejectsNonStringProjectionBeforeNetwork(t *testing.T) { t.Fatalf("expected ErrStringProjection, got %v", err) } } + + +func TestRemainingBrokerOptionsValidateLazily(t *testing.T) { + tests := []struct { + name string + opt EventStoreOption + want string + }{ + {"sqs", WithSQS(SQSProviderConfig{}), "goeventbus: SQS client must not be nil"}, + {"sns", WithSNS(SNSProviderConfig{}), "goeventbus: SNS client must not be nil"}, + {"gcp-pubsub", WithGCPPubSub(GCPPubSubProviderConfig{}), "goeventbus: GCP Pub/Sub client must not be nil"}, + {"azure-service-bus", WithAzureServiceBus(AzureServiceBusProviderConfig{}), "goeventbus: Azure Service Bus client must not be nil"}, + {"pulsar", WithPulsar(PulsarProviderConfig{}), "goeventbus: Pulsar client must not be nil"}, + {"mqtt", WithMQTT(MQTTProviderConfig{}), "goeventbus: MQTT client or client options are required"}, + {"postgres", WithPostgres(PostgresProviderConfig{}), "goeventbus: PostgreSQL pool or connection string is required"}, + {"nsq", WithNSQ(NSQProviderConfig{}), "goeventbus: NSQ nsqd address must not be empty"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var dispatcher Dispatcher + store := NewEventStore(&dispatcher, 8, Block, tt.opt) + err := store.PublishToProvider(nil, Event{Projection: "order.created"}) + if err == nil || err.Error() != tt.want { + t.Fatalf("expected %q, got %v", tt.want, err) + } + }) + } +} + +func TestPostgresRejectsUnsafeTableName(t *testing.T) { + _, err := NewPostgresProvider(PostgresProviderConfig{ + ConnectionString: "postgres://localhost/test", + Table: "events;drop table users", + }) + if err == nil { + t.Fatal("expected invalid table name error") + } +} + +func TestMQTTRejectsInvalidQoS(t *testing.T) { + _, err := NewMQTTProvider(MQTTProviderConfig{ + Client: nil, + Options: new(mqtt.ClientOptions), + Topic: "events", + QoS: 3, + }) + if err == nil { + t.Fatal("expected invalid MQTT QoS error") + } +} From b772638641785f2e7b9cad6882a5a2d0ccfa86ba Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:59:52 +0200 Subject: [PATCH 11/12] Document remaining official broker providers --- README.MD | 133 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 132 insertions(+), 1 deletion(-) diff --git a/README.MD b/README.MD index b4a1841..9fbf197 100644 --- a/README.MD +++ b/README.MD @@ -63,7 +63,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish() - **Lifecycle hooks** with `OnBefore`, `OnAfter`, and `OnError`. - **Back-pressure policies**: `DropOldest`, `Block`, or `ReturnError`. - **Dead-letter queue** with replay support and panic recovery. -- **Redis Streams, RabbitMQ, NATS JetStream, and Kafka EventStore options** for cross-process delivery. +- **12 official EventStore broker options**: Redis Streams, RabbitMQ, NATS JetStream, Kafka, SQS, SNS, GCP Pub/Sub, Azure Service Bus, Pulsar, MQTT, PostgreSQL, and NSQ. - **Transactions** with local buffering, transport-aware commit, and optional Rules/Cache/Jev event selection. - **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection. - **Metrics** for published, processed, and failed events. @@ -762,6 +762,137 @@ Kafka consumer offsets are committed synchronously only after successful event handling. With the default hash balancer, the string projection is used as the message key, preserving ordering for a projection within a partition. +## Remaining official brokers + +All remaining official transports use the same store-first API: + +```go +store := GoEventBus.NewEventStore( + &dispatcher, + 1024, + GoEventBus.Block, + GoEventBus.WithSQS(sqsConfig), // swap for any provider below +) +``` + +| Broker | EventStore option | Consume semantics | +|---|---|---| +| AWS SQS | `WithSQS` | long-poll; delete only after successful handling | +| AWS SNS | `WithSNS` | publish through SNS; consumption delegates to a real subscriber transport such as SQS | +| Google Cloud Pub/Sub | `WithGCPPubSub` | Ack on success, Nack on decode/handler failure | +| Azure Service Bus | `WithAzureServiceBus` | PeekLock; Complete on success, Abandon on failure | +| Apache Pulsar | `WithPulsar` | durable subscription; Ack on success, Nack on failure | +| MQTT | `WithMQTT` | QoS 0/1/2; broker ACK behavior follows MQTT/Paho semantics | +| PostgreSQL | `WithPostgres` | persistent table + `FOR UPDATE SKIP LOCKED`; delete on success, delayed retry on failure | +| NSQ | `WithNSQ` | channel-based consumption; handler errors are requeued by go-nsq | + +Broker-backed events still require string projections, custom `EventCodec` +implementations are supported by every provider, and `Drain` / `Close` +closes provider-owned resources. Injected cloud clients/connections remain +application-owned. + +### AWS SQS + +```go +store := GoEventBus.NewEventStore( + &dispatcher, 1024, GoEventBus.Block, + GoEventBus.WithSQS(GoEventBus.SQSProviderConfig{ + Client: sqsClient, + QueueURL: queueURL, + }), +) +``` + +FIFO queues can set `FIFO`, `MessageGroupID`, and an optional +`MessageDeduplicationID` callback. + +### AWS SNS + +```go +sqsSubscriber, _ := GoEventBus.NewSQSProvider(sqsConfig) + +store := GoEventBus.NewEventStore( + &dispatcher, 1024, GoEventBus.Block, + GoEventBus.WithSNS(GoEventBus.SNSProviderConfig{ + Client: snsClient, + TopicARN: topicARN, + Subscriber: sqsSubscriber, + }), +) +``` + +SNS is not modeled as a queue. `Consume` requires an actual subscriber +transport such as SQS, matching SNS delivery semantics. + +### Google Cloud Pub/Sub + +```go +GoEventBus.WithGCPPubSub(GoEventBus.GCPPubSubProviderConfig{ + Client: pubsubClient, + Topic: "events", + Subscription: "billing", + EnableOrdering: true, +}) +``` + +### Azure Service Bus + +```go +GoEventBus.WithAzureServiceBus(GoEventBus.AzureServiceBusProviderConfig{ + Client: sbClient, + Queue: "billing", +}) +``` + +For topic subscriptions, set `Topic` and `Subscription` instead of `Queue`. + +### Apache Pulsar + +```go +GoEventBus.WithPulsar(GoEventBus.PulsarProviderConfig{ + Client: pulsarClient, + Topic: "persistent://public/default/events", + Subscription: "billing", +}) +``` + +### MQTT + +```go +GoEventBus.WithMQTT(GoEventBus.MQTTProviderConfig{ + Options: mqttOptions, + TopicPrefix: "events/", + QoS: 1, +}) +``` + +An injected MQTT client remains caller-owned. When `Options` is supplied, +GoEventBus creates and owns the client. Automatic reconnect and persistent +sessions are configured through Paho's `ClientOptions`. + +### PostgreSQL + +```go +GoEventBus.WithPostgres(GoEventBus.PostgresProviderConfig{ + ConnectionString: os.Getenv("DATABASE_URL"), + Table: "goeventbus_events", +}) +``` + +The PostgreSQL provider uses a durable table-backed queue. Consumers claim one +row with `FOR UPDATE SKIP LOCKED`; successful processing deletes the row, +while failures increment the attempt count and delay the next delivery. + +### NSQ + +```go +GoEventBus.WithNSQ(GoEventBus.NSQProviderConfig{ + NSQDAddress: "127.0.0.1:4150", + Topic: "events", + Channel: "billing", +}) +``` + Providers use `JSONCodec` by default. Supply a custom `DecodePayload` when consumers need concrete payload types rather than generic JSON values. From 88f01471096a8f5a6cfc4e137478b3e863a97aae Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 20:02:41 +0200 Subject: [PATCH 12/12] Document all official broker providers --- website/dist/docs/index.html | 197 ++++++++++++++++++----------------- 1 file changed, 101 insertions(+), 96 deletions(-) diff --git a/website/dist/docs/index.html b/website/dist/docs/index.html index 8f265cb..648b801 100644 --- a/website/dist/docs/index.html +++ b/website/dist/docs/index.html @@ -3,7 +3,7 @@ - + Documentation — GoEventBus @@ -61,7 +61,7 @@

Routing & integrations

Rules, cache & Jev -Redis & RabbitMQ +12 broker providers

Operations

Dead letter queue @@ -88,7 +88,7 @@

Documentation#

A high-performance event bus for Go with deterministic rules, cached decisions, and optional Jev-powered event routing.

01 / SET UPInstall GoEventBusOne package. Go 1.23+.02 / BUILDRun your first eventRegister, enqueue, dispatch.
-

GoEventBus combines a bounded MPMC in-memory event bus with fan-out, ordered and batch handlers, middleware, lifecycle hooks, dead-letter handling, Redis Streams, RabbitMQ, and an optional decision layer for choosing which event type should execute.

+

GoEventBus combines a bounded MPMC in-memory event bus with fan-out, ordered and batch handlers, middleware, lifecycle hooks, dead-letter handling, 12 official broker providers, and an optional decision layer for choosing which event type should execute.

01Register02Subscribe03Publish
@@ -208,7 +208,7 @@

FeaturesLifecycle hooks with OnBefore, OnAfter, and OnError.
  • Back-pressure policies: DropOldest, Block, or ReturnError.
  • Dead-letter queue with replay support and panic recovery.
    • -
  • Redis Streams and RabbitMQ providers for cross-process delivery.
    • +
  • 12 official broker providers: Redis Streams, RabbitMQ, NATS JetStream, Kafka, AWS SQS, AWS SNS, Google Cloud Pub/Sub, Azure Service Bus, Apache Pulsar, MQTT, PostgreSQL, and NSQ.
  • Transactions with local buffering and transport-aware commit to local handlers or the configured broker.
  • Scheduling with absolute and relative timers.
  • Metrics for published, processed, and failed events.
    • @@ -479,111 +479,116 @@

    Rules, cache & Jev +

    Routing & integrations

    -

    Redis & RabbitMQ#

    -

    Redis Streams and RabbitMQ are configured directly on NewEventStore. The store owns the provider lifecycle, and broker construction is lazy until the first publish or consume operation.

    -

    When a broker is configured, DecideAndSubscribe selects the event type and publishes the selected event to that broker. Without a broker option, the same method keeps its normal local enqueue behavior.

    - -

    Redis Streams#

    -
    Go
    rdb := redis.NewClient(&redis.Options{
-    Addr: "localhost:6379",
-})
-
-store := GoEventBus.NewEventStore(
+
    External providers#
    
+
    GoEventBus exposes one application-facing broker API across twelve official transports. Configure exactly one provider on NewEventStore, then use the same DecideAndSubscribe, PublishToProvider, Consume, and shutdown lifecycle regardless of the broker.
    
+
    Go
    store := GoEventBus.NewEventStore(
     &dispatcher,
     1024,
     GoEventBus.Block,
-    GoEventBus.WithRedis(GoEventBus.RedisProviderConfig{
-        Client:   rdb,
-        Stream:   "events",
-        Group:    "billing",
-        Consumer: "billing-1",
-        StartID:  "0",
+    GoEventBus.WithSQS(sqsConfig), // swap only this option
+)
    
+
+
    Every official provider supports the shared EventCodec abstraction. JSONCodec is the default.
    
+
+
    Broker capability matrix#
    
+
    
+
+
+
+
+
+
+
+
+
+
+
+
+
    BrokerOptionSuccess acknowledgementFailure / retryOrdering / grouping
    Redis StreamsWithRedisXACK after handler successPending entry remains for retryConsumer groups; stream order
    RabbitMQWithRabbitMQManual Ack after successNack + requeueTopic routing; queue ordering
    NATS JetStreamWithNATSJetStreamExplicit AckNak for redeliveryDurable consumers; queue groups
    Apache KafkaWithKafkaOffset commit after successUncommitted offset is redeliveredPartitions; consumer groups
    AWS SQSWithSQSDelete after successVisibility timeout makes failed messages visible againStandard or FIFO queues
    AWS SNSWithSNSPublish acceptanceSubscriber transport owns retryTopics; FIFO when configured
    GCP Pub/SubWithGCPPubSubAck after successNack on decode/handler failureSubscriptions; optional ordering keys
    Azure Service BusWithAzureServiceBusComplete in PeekLock modeAbandon on failureQueues or topic subscriptions
    Apache PulsarWithPulsarAck after successNack for redeliveryShared / Exclusive / Failover / KeyShared
    MQTTWithMQTTProtocol QoS acknowledgementHandler errors cannot be translated to broker Nack after callback deliveryTopics; QoS 0/1/2
    PostgreSQLWithPostgresDelete row after successAttempts increment + delayed availabilityPersistent table; SKIP LOCKED
    NSQWithNSQFinish after handler successHandler error triggers requeueTopics + channels
    
+
    
+
+
    AWS SQS#
    
+
    Go
    store := GoEventBus.NewEventStore(
+    &dispatcher, 1024, GoEventBus.Block,
+    GoEventBus.WithSQS(GoEventBus.SQSProviderConfig{
+        Client:   sqsClient,
+        QueueURL: queueURL,
     }),
-)
-defer store.Close(context.Background())
+)
    
+
    SQS uses long polling. Messages are deleted only after the EventStore accepts and processes them successfully. FIFO queues can set FIFO, MessageGroupID, and MessageDeduplicationID.
    
+
    
 
-go func() {
-    if err := store.Consume(ctx); err != nil &&
-        !errors.Is(err, context.Canceled) &&
-        !errors.Is(err, GoEventBus.ErrProviderClosed) {
-        log.Println(err)
-    }
-}()
+
    AWS SNS#
    
+
    Go
    sqsSubscriber, _ := GoEventBus.NewSQSProvider(sqsConfig)
 
-brokerCandidates := []GoEventBus.EventCandidate{
-    {
-        Key:         "order_created",
-        Projection:  "order.created",
-        Description: "A new order should be created",
-    },
-}
+store := GoEventBus.NewEventStore(
+    &dispatcher, 1024, GoEventBus.Block,
+    GoEventBus.WithSNS(GoEventBus.SNSProviderConfig{
+        Client:     snsClient,
+        TopicARN:   topicARN,
+        Subscriber: sqsSubscriber,
+    }),
+)
    
+
    SNS is publish-oriented, not a queue. Consume therefore delegates to a supported subscriber transport such as SQS instead of inventing queue semantics.
    
+
    
 
-decision, err := store.DecideAndSubscribe(
-    ctx,
-    selector,
-    state,
-    GoEventBus.Event{
-        ID:   "evt-redis-1",
-        Data: order,
-    },
-    brokerCandidates,
-)
-if err != nil {
-    log.Fatal(err)
-}
+
    Google Cloud Pub/Sub#
    
+
    Go
    GoEventBus.WithGCPPubSub(GoEventBus.GCPPubSubProviderConfig{
+    Client:         pubsubClient,
+    Topic:          "events",
+    Subscription:   "billing",
+    EnableOrdering: true,
+})
    
+
    Successful handling calls Ack. Decode or handler failures call Nack. When ordering is enabled, the projection is the default ordering key.
    
+
    
 
-fmt.Println("published decision:", decision.Choice)
    -

    If the projection is already known and no decision step is needed, use store.PublishToProvider(ctx, event) instead.

    +

    Azure Service Bus#

    +
    Go
    GoEventBus.WithAzureServiceBus(GoEventBus.AzureServiceBusProviderConfig{
+    Client: sbClient,
+    Queue:  "billing",
+})
    +

    Queue consumers use PeekLock processing: successful messages are completed, while failures are abandoned for retry. For pub/sub, configure Topic and Subscription instead of Queue.

    -

    RabbitMQ#

    -
    Go
    store := GoEventBus.NewEventStore(
-    &dispatcher,
-    1024,
-    GoEventBus.Block,
-    GoEventBus.WithRabbitMQ(GoEventBus.RabbitMQProviderConfig{
-        URL:        "amqp://guest:guest@localhost:5672/",
-        Exchange:   "events",
-        Queue:      "billing",
-        BindingKey: "order.*",
-        Consumer:   "billing-1",
-    }),
-)
-defer store.Close(context.Background())
 
-go func() {
-    if err := store.Consume(ctx); err != nil &&
-        !errors.Is(err, context.Canceled) &&
-        !errors.Is(err, GoEventBus.ErrProviderClosed) {
-        log.Println(err)
-    }
-}()
+
    Apache Pulsar#
    
+
    Go
    GoEventBus.WithPulsar(GoEventBus.PulsarProviderConfig{
+    Client:       pulsarClient,
+    Topic:        "persistent://public/default/events",
+    Subscription: "billing",
+})
    
+
    The default subscription type is Shared. Configure SubscriptionType for Exclusive, Failover, or KeyShared behavior. Handler failures are Nacked for redelivery.
    
+
    
 
-brokerCandidates := []GoEventBus.EventCandidate{
-    {
-        Key:         "order_created",
-        Projection:  "order.created",
-        Description: "A new order should be created",
-    },
-}
+
    MQTT#
    
+
    Go
    GoEventBus.WithMQTT(GoEventBus.MQTTProviderConfig{
+    Options:     mqttOptions,
+    TopicPrefix: "events/",
+    QoS:         1,
+})
    
+
    QoS 0, 1, and 2 are supported. Automatic reconnect and persistent-session behavior are configured with Paho client options. Injected clients remain caller-owned; clients created from Options are owned by the provider.
    
+
    
 
-decision, err := store.DecideAndSubscribe(
-    ctx,
-    selector,
-    state,
-    GoEventBus.Event{
-        ID:   "evt-rabbit-1",
-        Data: order,
-    },
-    brokerCandidates,
-)
-if err != nil {
-    log.Fatal(err)
-}
+
    PostgreSQL#
    
+
    Go
    GoEventBus.WithPostgres(GoEventBus.PostgresProviderConfig{
+    ConnectionString: os.Getenv("DATABASE_URL"),
+    Table:            "goeventbus_events",
+})
    
+
    The PostgreSQL provider implements a persistent table-backed queue. Consumers claim rows with FOR UPDATE SKIP LOCKED; success deletes the row and failure increments attempts and moves available_at forward for retry. An injected pool remains caller-owned.
    
+
    
+
+
    NSQ#
    
+
    Go
    GoEventBus.WithNSQ(GoEventBus.NSQProviderConfig{
+    NSQDAddress: "127.0.0.1:4150",
+    Topic:       "events",
+    Channel:     "billing",
+})
    
+
    NSQ uses topics and channels. The provider returns handler errors to go-nsq, allowing the client to requeue failed messages according to NSQ retry behavior.
    
+
    
 
-fmt.Println("published decision:", decision.Choice)
    -

    Providers use JSONCodec by default. Supply a custom DecodePayload when consumers need concrete payload types rather than generic JSON values.

    +

    Redis, RabbitMQ, NATS, and Kafka#

    +

    The existing WithRedis, WithRabbitMQ, WithNATSJetStream, and WithKafka options use the same EventStore-first API and lifecycle. This means application code can swap transports without changing routing or execution code.

    @@ -671,7 +676,7 @@

    Transactions

    01Rules / Cache / Jev02DecideAndPublish03Transaction buffer04Commit

    DecideAndPublish resolves the event projection immediately and buffers only the selected event. Routing failures do not mutate the transaction, and Rollback discards intelligently selected events just like direct events.

    - +

    Local Commit stops at the first handler error. Broker-backed Commit publishes events in transaction order, stops at the first provider error, and removes only the prefix already confirmed by the provider. Retrying the transaction therefore starts from the first unpublished event instead of publishing the confirmed prefix again.

    Transactions are not database-style atomic transactions: handler side effects or broker publishes that already succeeded cannot be undone.