README.md

eda

Tier: Platform · Status: Partial (in-memory full; Kafka/RabbitMQ scaffolds) · Java original: firefly-common-eda · .NET project: FireflyFramework.Eda

Overview

eda is the framework's event-driven architecture port. It defines the Event envelope every Firefly event flows through, the Publisher / Subscriber / Broker interfaces, and an in-process fan-out InMemory broker. Production transports — Kafka via twmb/franz-go and RabbitMQ via rabbitmq/amqp091-go — share the same interfaces and slot in via eda.NewKafkaBroker(ctx, cfg) / eda.NewRabbitMQBroker(ctx, cfg) once the dedicated transport modules ship.

Until those land, NewKafkaBroker and NewRabbitMQBroker return typed sentinel errors (ErrKafkaUnavailable, ErrRabbitMQUnavailable) so a misconfigured deployment fails loud at startup rather than silently falling back to in-memory.

Public surface

type Event struct {
    ID            string
    Type          string
    Source        string
    Topic         string
    CorrelationID string
    Time          time.Time
    Headers       map[string]string
    Payload       []byte
}
func NewEvent(ctx, topic, eventType, source, payload) Event

type Publisher interface { Publish(ctx, ev Event) error; Close() error }
type Handler   func(ctx context.Context, ev Event) error
type Subscriber interface { Subscribe(ctx, topic, h Handler) error; Close() error }
type Broker interface { Publisher; Subscriber }

func NewInMemory() *InMemory   // fan-out broker, sync handler invocation
func NewKafkaBroker(ctx, KafkaConfig) (Broker, error)        // sentinel until wired
func NewRabbitMQBroker(ctx, RabbitMQConfig) (Broker, error)  // sentinel until wired

NewEvent extracts the correlation id from kernel.WithCorrelationID and stamps the event's CorrelationID automatically.

Quick start

import (
    "context"
    "github.com/fireflyframework/fireflyframework-go/eda"
)

b := eda.NewInMemory()
defer b.Close()

_ = b.Subscribe(ctx, "orders.created", func(ctx context.Context, ev eda.Event) error {
    log.Printf("got order %s", ev.ID)
    return nil
})
_ = b.Publish(ctx, eda.NewEvent(ctx, "orders.created", "OrderCreated", "orders-svc", []byte(`{...}`)))

For Kafka in production:

broker, err := eda.NewKafkaBroker(ctx, eda.KafkaConfig{
    Brokers:       []string{"kafka:9092"},
    ClientID:      "orders",
    ConsumerGroup: "orders-group",
})
// broker is a Broker satisfying both Publisher and Subscriber.

Testing

cd eda
go test ./...

Covers in-memory fan-out across multiple subscribers, correlation-id propagation through NewEvent, handler-error short-circuit, and the Kafka / RabbitMQ sentinel returns.