diff --git a/README.MD b/README.MD index 268a987..b4a1841 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 and RabbitMQ EventStore options** for cross-process delivery. +- **Redis Streams, RabbitMQ, NATS JetStream, and Kafka EventStore options** for cross-process delivery. - **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. @@ -718,6 +718,50 @@ if err != nil { fmt.Println("published decision:", decision.Choice) ``` +## NATS JetStream + +```go +nc, err := nats.Connect(nats.DefaultURL) +if err != nil { + log.Fatal(err) +} + +store := GoEventBus.NewEventStore( + &dispatcher, + 1024, + GoEventBus.Block, + GoEventBus.WithNATSJetStream(GoEventBus.NATSJetStreamProviderConfig{ + Conn: nc, + SubjectPrefix: "events.", + Subject: "events.>", + Durable: "billing", + Queue: "billing-workers", + }), +) +``` + +The injected NATS connection remains application-owned. JetStream messages use +explicit acknowledgements; handler failures are NAKed for broker redelivery. + +## Apache Kafka + +```go +store := GoEventBus.NewEventStore( + &dispatcher, + 1024, + GoEventBus.Block, + GoEventBus.WithKafka(GoEventBus.KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + }), +) +``` + +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. + Providers use `JSONCodec` by default. Supply a custom `DecodePayload` when consumers need concrete payload types rather than generic JSON values. diff --git a/broker_options_test.go b/broker_options_test.go new file mode 100644 index 0000000..5f1ffd5 --- /dev/null +++ b/broker_options_test.go @@ -0,0 +1,65 @@ +package GoEventBus + +import ( + "errors" + "testing" +) + +func TestWithNATSJetStreamRejectsNilConnectionLazily(t *testing.T) { + var dispatcher Dispatcher + store := NewEventStore(&dispatcher, 8, Block, WithNATSJetStream(NATSJetStreamProviderConfig{})) + err := store.PublishToProvider(nil, Event{Projection: "order.created"}) + if err == nil || err.Error() != "goeventbus: NATS connection must not be nil" { + t.Fatalf("expected nil NATS connection error, got %v", err) + } +} + +func TestWithKafkaValidatesConfigurationLazily(t *testing.T) { + var dispatcher Dispatcher + store := NewEventStore(&dispatcher, 8, Block, WithKafka(KafkaProviderConfig{})) + err := store.PublishToProvider(nil, Event{Projection: "order.created"}) + if err == nil || err.Error() != "goeventbus: Kafka brokers must not be empty" { + t.Fatalf("expected Kafka brokers error, got %v", err) + } +} + +func TestKafkaRejectsFixedPartitionWithGroup(t *testing.T) { + partition := 0 + _, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + Partition: &partition, + }) + if err == nil { + t.Fatal("expected group/partition validation error") + } +} + +func TestKafkaGroupUsesZeroValueConfigWithoutPartitionConflict(t *testing.T) { + provider, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + }) + if err != nil { + t.Fatalf("expected group config to be valid, got %v", err) + } + _ = provider.Close() +} + +func TestKafkaRejectsNonStringProjectionBeforeNetwork(t *testing.T) { + provider, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + }) + if err != nil { + t.Fatal(err) + } + defer provider.Close() + + err = provider.Publish(nil, Event{Projection: 123}) + if !errors.Is(err, ErrStringProjection) { + t.Fatalf("expected ErrStringProjection, got %v", err) + } +} diff --git a/go.mod b/go.mod index fb1265d..635f0d1 100644 --- a/go.mod +++ b/go.mod @@ -6,8 +6,10 @@ toolchain go1.23.7 require ( github.com/fasthttp/router v1.5.4 + 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 + github.com/segmentio/kafka-go v0.4.48 github.com/testcontainers/testcontainers-go v0.34.0 github.com/valyala/fasthttp v1.61.0 ) diff --git a/kafka_provider.go b/kafka_provider.go new file mode 100644 index 0000000..222ce87 --- /dev/null +++ b/kafka_provider.go @@ -0,0 +1,216 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/segmentio/kafka-go" +) + +// KafkaProviderConfig configures an Apache Kafka provider. +type KafkaProviderConfig struct { + Brokers []string + Topic string + Group string + + // Partition pins publishing and non-group consumption to one partition. + // Leave nil to let Kafka select/assign partitions. Partition and Group are + // mutually exclusive for consumers because group membership owns assignment. + Partition *int + + MinBytes int + MaxBytes int + MaxWait time.Duration + Codec EventCodec +} + +// KafkaProvider publishes through kafka-go Writer and consumes with synchronous +// offset commits after successful event handling. +type KafkaProvider struct { + writer *kafka.Writer + reader *kafka.Reader + + topic string + partition *int + codec EventCodec + + closeOnce sync.Once + closeErr error + closed atomic.Bool +} + +// WithKafka configures Apache Kafka as the EventStore provider. +func WithKafka(config KafkaProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { + return NewKafkaProvider(config) + }) +} + +// NewKafkaProvider builds Kafka clients. Network I/O begins on Publish/Consume. +func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { + if len(config.Brokers) == 0 { + return nil, errors.New("goeventbus: Kafka brokers must not be empty") + } + if config.Topic == "" { + return nil, errors.New("goeventbus: Kafka topic must not be empty") + } + if config.Partition != nil && *config.Partition < 0 { + return nil, errors.New("goeventbus: Kafka partition must be zero or greater") + } + if config.Group != "" && config.Partition != nil { + return nil, errors.New("goeventbus: Kafka partition cannot be fixed when a consumer group is configured") + } + if config.MinBytes <= 0 { + config.MinBytes = 1 + } + if config.MaxBytes <= 0 { + config.MaxBytes = 10e6 + } + if config.MaxWait <= 0 { + config.MaxWait = time.Second + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + var balancer kafka.Balancer = &kafka.Hash{} + if config.Partition != nil { + balancer = fixedKafkaPartitionBalancer(*config.Partition) + } + + writer := &kafka.Writer{ + Addr: kafka.TCP(config.Brokers...), + Topic: config.Topic, + Balancer: balancer, + RequiredAcks: kafka.RequireAll, + Async: false, + } + + readerConfig := kafka.ReaderConfig{ + Brokers: config.Brokers, + GroupID: config.Group, + Topic: config.Topic, + MinBytes: config.MinBytes, + MaxBytes: config.MaxBytes, + MaxWait: config.MaxWait, + CommitInterval: 0, + } + if config.Partition != nil { + readerConfig.Partition = *config.Partition + } + + return &KafkaProvider{ + writer: writer, + reader: kafka.NewReader(readerConfig), + topic: config.Topic, + partition: config.Partition, + codec: config.Codec, + }, nil +} + +// Publish writes one Kafka message. Projection is stored as the message key, +// preserving per-projection ordering with the default hash balancer. A fixed +// partition, when configured, overrides the balancer. +func (p *KafkaProvider) 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 := kafka.Message{ + Topic: p.topic, + Key: []byte(projection), + Value: payload, + Time: time.Now().UTC(), + } + if e.ID != "" { + msg.Headers = append(msg.Headers, kafka.Header{Key: "goeventbus-event-id", Value: []byte(e.ID)}) + } + + if err := p.writer.WriteMessages(ctx, msg); err != nil { + return fmt.Errorf("goeventbus: publish Kafka event: %w", err) + } + return nil +} + +// Consume fetches one message at a time and commits its offset only after the +// EventConsumer returns nil. Failed handling leaves the offset uncommitted for +// redelivery according to Kafka consumer-group semantics. +func (p *KafkaProvider) 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 + } + msg, err := p.reader.FetchMessage(ctx) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: fetch Kafka event: %w", err) + } + + event, err := p.codec.Decode(msg.Value) + if err != nil { + return fmt.Errorf("goeventbus: decode Kafka event at partition %d offset %d: %w", msg.Partition, msg.Offset, err) + } + if err := consumer(ctx, event); err != nil { + return err + } + if err := p.reader.CommitMessages(ctx, msg); err != nil { + return fmt.Errorf("goeventbus: commit Kafka offset %d: %w", msg.Offset, err) + } + } +} + +// Close closes the owned Kafka reader and writer. +func (p *KafkaProvider) Close() error { + p.closeOnce.Do(func() { + p.closed.Store(true) + p.closeErr = errors.Join(p.reader.Close(), p.writer.Close()) + }) + return p.closeErr +} + +type fixedKafkaPartitionBalancer int + +func (b fixedKafkaPartitionBalancer) Balance(_ kafka.Message, partitions ...int) int { + want := int(b) + for _, partition := range partitions { + if partition == want { + return want + } + } + if len(partitions) == 0 { + return 0 + } + return partitions[0] +} + +var _ Provider = (*KafkaProvider)(nil) diff --git a/nats_jetstream_provider.go b/nats_jetstream_provider.go new file mode 100644 index 0000000..7204c43 --- /dev/null +++ b/nats_jetstream_provider.go @@ -0,0 +1,232 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/nats-io/nats.go" +) + +// NATSJetStreamProviderConfig configures a NATS JetStream provider. +// +// Conn is injected and remains owned by the caller. The provider creates +// JetStream subscriptions lazily in Consume and never closes Conn. +type NATSJetStreamProviderConfig struct { + Conn *nats.Conn + + // Subject is the subscription subject. It defaults to SubjectPrefix + ">". + Subject string + // SubjectPrefix is prepended to the event projection on publish. + // It defaults to "events.". + SubjectPrefix string + // Durable configures a durable JetStream consumer. + Durable string + // Queue enables queue-group consumption for multiple consumers. + Queue string + + Codec EventCodec +} + +// NATSJetStreamProvider publishes to and consumes from NATS JetStream. +// Messages are acknowledged only after EventConsumer succeeds. +type NATSJetStreamProvider struct { + conn *nats.Conn + js nats.JetStreamContext + subject string + subjectPrefix string + durable string + queue string + codec EventCodec + + mu sync.Mutex + sub *nats.Subscription + closed atomic.Bool +} + +// WithNATSJetStream configures NATS JetStream as the EventStore provider. +func WithNATSJetStream(config NATSJetStreamProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { + return NewNATSJetStreamProvider(config) + }) +} + +// NewNATSJetStreamProvider creates a NATS JetStream provider without taking +// ownership of the injected connection. +func NewNATSJetStreamProvider(config NATSJetStreamProviderConfig) (*NATSJetStreamProvider, error) { + if config.Conn == nil { + return nil, errors.New("goeventbus: NATS connection must not be nil") + } + if config.SubjectPrefix == "" { + config.SubjectPrefix = "events." + } + if config.Subject == "" { + config.Subject = config.SubjectPrefix + ">" + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + js, err := config.Conn.JetStream() + if err != nil { + return nil, fmt.Errorf("goeventbus: create NATS JetStream context: %w", err) + } + + return &NATSJetStreamProvider{ + conn: config.Conn, + js: js, + subject: config.Subject, + subjectPrefix: config.SubjectPrefix, + durable: config.Durable, + queue: config.Queue, + codec: config.Codec, + }, nil +} + +// Publish publishes an event to SubjectPrefix + Projection. +func (p *NATSJetStreamProvider) 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 := &nats.Msg{ + Subject: p.subjectPrefix + projection, + Data: payload, + Header: nats.Header{}, + } + if e.ID != "" { + msg.Header.Set(nats.MsgIdHdr, e.ID) + } + + if _, err := p.js.PublishMsg(msg, nats.Context(ctx)); err != nil { + return fmt.Errorf("goeventbus: publish NATS JetStream event: %w", err) + } + return nil +} + +// Consume subscribes with explicit acknowledgements. Handler failures NAK the +// message so JetStream can redeliver it. +func (p *NATSJetStreamProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + + sub, err := p.subscribe() + if err != nil { + return err + } + defer func() { + p.mu.Lock() + if p.sub == sub { + p.sub = nil + } + p.mu.Unlock() + _ = sub.Unsubscribe() + }() + + for { + if p.closed.Load() { + return ErrProviderClosed + } + msg, err := sub.NextMsgWithContext(ctx) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: consume NATS JetStream event: %w", err) + } + + event, err := p.codec.Decode(msg.Data) + if err != nil { + _ = msg.Nak() + return fmt.Errorf("goeventbus: decode NATS JetStream event: %w", err) + } + if err := consumer(ctx, event); err != nil { + _ = msg.Nak() + return err + } + if err := msg.Ack(); err != nil { + return fmt.Errorf("goeventbus: acknowledge NATS JetStream event: %w", err) + } + } +} + +func (p *NATSJetStreamProvider) subscribe() (*nats.Subscription, error) { + p.mu.Lock() + defer p.mu.Unlock() + + if p.closed.Load() { + return nil, ErrProviderClosed + } + if p.sub != nil && p.sub.IsValid() { + return nil, errors.New("goeventbus: NATS JetStream provider already has an active consumer") + } + + opts := []nats.SubOpt{nats.ManualAck(), nats.AckExplicit()} + if p.durable != "" { + opts = append(opts, nats.Durable(p.durable)) + } + + var ( + sub *nats.Subscription + err error + ) + if strings.TrimSpace(p.queue) != "" { + sub, err = p.js.QueueSubscribeSync(p.subject, p.queue, opts...) + } else { + sub, err = p.js.SubscribeSync(p.subject, opts...) + } + if err != nil { + return nil, fmt.Errorf("goeventbus: subscribe NATS JetStream subject %q: %w", p.subject, err) + } + p.sub = sub + return sub, nil +} + +// Close stops active consumption. The injected NATS connection is not closed. +func (p *NATSJetStreamProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.mu.Lock() + defer p.mu.Unlock() + if p.sub == nil { + return nil + } + err := p.sub.Unsubscribe() + p.sub = nil + if err != nil && !errors.Is(err, nats.ErrBadSubscription) { + return fmt.Errorf("goeventbus: close NATS JetStream subscription: %w", err) + } + return nil +} + +// keep time imported for compatibility with nats context implementations that +// surface timeout errors through NextMsgWithContext on older clients. +var _ = time.Second + +var _ Provider = (*NATSJetStreamProvider)(nil)