Getting started

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, 12 official broker providers, and an optional decision layer for choosing which event type should execute.

01Register02Subscribe03Publish

Getting started

Installation#

go get github.com/Protocol-Lattice/GoEventBus

GoEventBus currently targets Go 1.23+.

Getting started

Quick start#

Plain event bus#

Use the direct path when the event type is already known.

package main

import (
    "context"
    "fmt"
    "log"

    GoEventBus "github.com/Protocol-Lattice/GoEventBus"
)

type UserCreated struct {
    ID string
}

func main() {
    dispatcher := GoEventBus.Dispatcher{}

    dispatcher.Register("user.created", func(
        ctx context.Context,
        ev GoEventBus.Event,
    ) (GoEventBus.Result, error) {
        payload := ev.Data.(UserCreated)
        fmt.Println("user created:", payload.ID)
        return GoEventBus.Result{Message: "ok"}, nil
    })

    store := GoEventBus.NewEventStore(
        &dispatcher,
        1<<16,
        GoEventBus.DropOldest,
    )
    store.Async = true

    if err := store.Subscribe(context.Background(), GoEventBus.Event{
        ID:         "evt-1",
        Projection: "user.created",
        Data:       UserCreated{ID: "u-42"},
    }); err != nil {
        log.Fatal(err)
    }

    store.Publish()

    if err := store.Drain(context.Background()); err != nil {
        log.Fatal(err)
    }
}

Projection may be any comparable value for local dispatch, including a string or a typed struct.

Getting started

Events & handlers#

type Event struct {
    ID         string
    Projection interface{}
    Data       any
    Args       map[string]any // deprecated
}

Prefer Data for payloads. Args remains for backwards compatibility.

Handlers use:

type HandlerFunc func(
    context.Context,
    Event,
) (Result, error)

dispatcher := GoEventBus.Dispatcher{}

dispatcher.Register(
    "order.created",
    handleBilling,
    handleAnalytics,
)

Calling Register repeatedly for the same projection appends handlers rather than replacing them.

Getting started

Why GoEventBus?#

Most events do not need model inference. Some do.

GoEventBus keeps those paths separate:

Known event type → enqueue it directly.
Obvious rule → route deterministically.
Repeated decision → reuse the cached choice.
Ambiguous input → ask Jev to choose from a bounded candidate set.
Execution → always stays inside GoEventBus.

That gives you intelligent routing without putting an LLM call in the Publish() hot path.

Getting started

Features#

Bounded MPMC ring buffer built with atomics and cache-line padding.
Rules → cache → Jev routing for optional intelligent event selection.
Typed or string projections for local dispatch.
Fan-out handlers with independent execution.
Ordered handlers with FIFO delivery per ordering key.
Batch handlers for bulk writes and high-throughput pipelines.
Sync and async dispatch with a fixed worker pool.
Middleware for per-handler cross-cutting behavior.
Lifecycle hooks with OnBefore, OnAfter, and OnError.
Back-pressure policies: DropOldest, Block, or ReturnError.
Dead-letter queue with replay support and panic recovery.
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.
Pluggable decision cache through the DecisionCache interface.

Event handling

Fan-out#

Multiple handlers may subscribe to the same projection.

dispatcher.Register(
    "order.placed",
    auditLogger,
    inventoryReducer,
    notificationSender,
)

Each handler is independent. One handler returning an error does not prevent the other fan-out handlers from running.

In async mode, each regular handler invocation becomes its own worker-pool item.

Event handling

Ordered Handlers#

Use ordered handlers when events for the same entity must be processed sequentially while unrelated entities can still run concurrently.

type OrderEvent struct {
    OrderID  string
    Sequence int
}

store.RegisterOrdered(
    "order.event",
    func(ev GoEventBus.Event) string {
        return ev.Data.(OrderEvent).OrderID
    },
    func(
        ctx context.Context,
        ev GoEventBus.Event,
    ) (GoEventBus.Result, error) {
        return applyOrderEvent(ctx, ev.Data.(OrderEvent))
    },
)

For a given ordering key, async delivery preserves FIFO order.

Different keys remain concurrent.

Event handling

Batch Handlers#

Batch handlers collect pending events for a projection during a Publish cycle and deliver them in chunks.

store.RegisterBatch(
    "metrics.recorded",
    100,
    func(
        ctx context.Context,
        events []GoEventBus.Event,
    ) ([]GoEventBus.Result, error) {
        return nil, writeMetricsBatch(ctx, events)
    },
)

Regular and batch handlers may coexist on the same projection.

Multiple batch handlers may also be registered for one projection; each receives the full chunk independently.

Middleware is not applied to batch handlers. Lifecycle hooks are.

Event handling

Middleware#

Middleware wraps regular handlers.

store.Use(func(next GoEventBus.HandlerFunc) GoEventBus.HandlerFunc {
    return func(
        ctx context.Context,
        ev GoEventBus.Event,
    ) (GoEventBus.Result, error) {
        started := time.Now()
        result, err := next(ctx, ev)
        log.Printf(
            "projection=%v duration=%s err=%v",
            ev.Projection,
            time.Since(started),
            err,
        )
        return result, err
    }
})

Middleware is applied independently to every handler in a fan-out.

Event handling

Lifecycle Hooks#

Use hooks for observability without changing handler logic.

store.OnBefore(func(ctx context.Context, ev GoEventBus.Event) {
    metrics.Inc("handler.started")
})

store.OnAfter(func(
    ctx context.Context,
    ev GoEventBus.Event,
    result GoEventBus.Result,
    err error,
) {
    metrics.Inc("handler.finished")
})

store.OnError(func(
    ctx context.Context,
    ev GoEventBus.Event,
    err error,
) {
    log.Printf("event=%s err=%v", ev.ID, err)
})

For batch handlers, hooks are emitted per event.

Event handling

Back-pressure#

Choose how Subscribe behaves when the ring buffer is full.

Policy	Behavior
`DropOldest`	Evict the oldest queued event and accept the new one
`Block`	Wait for capacity while respecting the caller context
`ReturnError`	Return `ErrBufferFull` immediately

Example:

store := GoEventBus.NewEventStore(
    &dispatcher,
    1<<14,
    GoEventBus.Block,
)

ctx, cancel := context.WithTimeout(
    context.Background(),
    50*time.Millisecond,
)
defer cancel()

if err := store.Subscribe(ctx, event); err != nil {
    log.Println("enqueue failed:", err)
}

Event handling

Async & shutdown#

Enable worker-pool dispatch with:

store.Async = true

The store uses a fixed worker pool sized to runtime.NumCPU().

Publish() submits regular, ordered, and batch work to that pool.

Call Drain or Close when shutting down:

ctx, cancel := context.WithTimeout(
    context.Background(),
    5*time.Second,
)
defer cancel()

if err := store.Drain(ctx); err != nil {
    log.Println("drain failed:", err)
}

Once shutdown begins, new subscriptions return ErrEventStoreClosed.

Routing & integrations

Rules, cache & Jev#

GoEventBus can choose an event type before enqueueing it.

The recommended pipeline is:

rules → cache → Jev → cache write → DecideAndSubscribe → local queue or configured broker

This keeps the fast path deterministic and only calls Jev when local logic cannot resolve the event type.

1. Define candidates#

type HouseWasSold struct{}

candidates := []GoEventBus.EventCandidate{
    {
        Key:         "user_created",
        Projection:  "user.created",
        Description: "A new user account was created",
    },
    {
        Key:         "house_sold",
        Projection:  HouseWasSold{},
        Description: "A property sale was completed",
    },
    {
        Key:         "order_cancelled",
        Projection:  "order.cancelled",
        Description: "An existing order should be cancelled",
    },
}

Key is the stable identifier exposed to the decision layer. Projection is the real GoEventBus dispatcher key.

That separation lets Jev choose between simple string keys while the event bus can still dispatch to typed projections.

2. Add deterministic rules#

Rules run first. The first matching rule wins.

rules := []GoEventBus.EventRule{
    {
        Name:   "explicit-cancel",
        Choice: "order_cancelled",
        Match: func(
            _ context.Context,
            state any,
            _ []GoEventBus.EventCandidate,
        ) bool {
            input, ok := state.(map[string]any)
            return ok && input["action"] == "cancel"
        },
    },
}

Rules are intentionally evaluated before cache. Adding a new rule can therefore override an older cached model decision immediately.

3. Add a decision cache#

cache := GoEventBus.NewMemoryDecisionCache(5 * time.Minute)

The built-in cache is goroutine-safe and lazily expires entries.

The default cache key is a SHA-256 hash of:

the JSON-serializable decision state,
candidate keys,
candidate descriptions.

Candidate ordering does not change the key. Projection values are deliberately excluded because typed projections may not be JSON-serializable.

Need a different strategy? Supply RuleCacheSelector.CacheKey.

4. Use Jev as the fallback#

jev := &GoEventBus.JevSelector{
    APIKey: os.Getenv("OPENROUTER_API_KEY"),
}

The default model is:

typesafe/jev-1.13

The adapter calls OpenRouter's Decisions API using the Go standard library, so the Jev integration adds no SDK dependency.

5. Compose the router#

selector := &GoEventBus.RuleCacheSelector{
    Rules:    rules,
    Cache:    cache,
    Fallback: jev,
}

Then route and deliver:

state := map[string]any{
    "message": "The property at 1 Main St was sold for $500000",
}

decision, err := store.DecideAndSubscribe(
    context.Background(),
    selector,
    state,
    GoEventBus.Event{
        ID: "evt-2",
    },
    candidates,
)
if err != nil {
    log.Fatal(err)
}

fmt.Printf(
    "choice=%s confidence=%.2f probabilities=%v\n",
    decision.Choice,
    decision.Confidence,
    decision.Probabilities,
)

store.Publish()

With a normal three-argument NewEventStore, DecideAndSubscribe enqueues locally and Publish() dispatches the event. With WithRedis(...) or WithRabbitMQ(...), the same call publishes the selected event to the configured broker instead.

If a rule matches, Jev is never called. If the same decision state is already cached, Jev is never called. Only a cache miss reaches the model.

Cache behavior#

DecisionCache is intentionally small:

type DecisionCache interface {
    Get(context.Context, string) (EventDecision, bool, error)
    Set(context.Context, string, EventDecision) error
}

That makes it straightforward to implement Redis, distributed, persistent, or application-specific caches.

Cache failures are fail-open by default. Routing continues through the fallback selector because cache is treated as an optimization.

Set:

StrictCache: true

when cache failures should fail the routing call instead.

Routing & integrations

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.

store := GoEventBus.NewEventStore(
    &dispatcher,
    1024,
    GoEventBus.Block,
    GoEventBus.WithSQS(sqsConfig), // swap only this option
)

Every official provider supports the shared EventCodec abstraction. JSONCodec is the default.

Broker capability matrix#

Broker	Option	Success acknowledgement	Failure / retry	Ordering / grouping
Redis Streams	`WithRedis`	`XACK` after handler success	Pending entry remains for retry	Consumer groups; stream order
RabbitMQ	`WithRabbitMQ`	Manual Ack after success	Nack + requeue	Topic routing; queue ordering
NATS JetStream	`WithNATSJetStream`	Explicit Ack	Nak for redelivery	Durable consumers; queue groups
Apache Kafka	`WithKafka`	Offset commit after success	Uncommitted offset is redelivered	Partitions; consumer groups
AWS SQS	`WithSQS`	Delete after success	Visibility timeout makes failed messages visible again	Standard or FIFO queues
AWS SNS	`WithSNS`	Publish acceptance	Subscriber transport owns retry	Topics; FIFO when configured
GCP Pub/Sub	`WithGCPPubSub`	Ack after success	Nack on decode/handler failure	Subscriptions; optional ordering keys
Azure Service Bus	`WithAzureServiceBus`	Complete in PeekLock mode	Abandon on failure	Queues or topic subscriptions
Apache Pulsar	`WithPulsar`	Ack after success	Nack for redelivery	Shared / Exclusive / Failover / KeyShared
MQTT	`WithMQTT`	Protocol QoS acknowledgement	Handler errors cannot be translated to broker Nack after callback delivery	Topics; QoS 0/1/2
PostgreSQL	`WithPostgres`	Delete row after success	Attempts increment + delayed availability	Persistent table; `SKIP LOCKED`
NSQ	`WithNSQ`	Finish after handler success	Handler error triggers requeue	Topics + channels

AWS SQS#

store := GoEventBus.NewEventStore(
    &dispatcher, 1024, GoEventBus.Block,
    GoEventBus.WithSQS(GoEventBus.SQSProviderConfig{
        Client:   sqsClient,
        QueueURL: queueURL,
    }),
)

SQS uses long polling. Messages are deleted only after the EventStore accepts and processes them successfully. FIFO queues can set FIFO, MessageGroupID, and MessageDeduplicationID.

AWS SNS#

sqsSubscriber, _ := GoEventBus.NewSQSProvider(sqsConfig)

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.

Google Cloud Pub/Sub#

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.

Azure Service Bus#

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.

Apache Pulsar#

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.

MQTT#

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.

PostgreSQL#

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#

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.

Redis Streams#

store := GoEventBus.NewEventStore(
    &dispatcher,
    1024,
    GoEventBus.Block,
    GoEventBus.WithRedis(GoEventBus.RedisProviderConfig{
        Client:   redisClient,
        Stream:   "events",
        Group:    "billing",
        Consumer: "billing-1",
        StartID:  "0",
    }),
)

Redis Streams uses consumer groups. Messages are acknowledged with XACK only after successful handling. Failed messages remain pending so the same consumer can retry them on a later consume cycle.

The injected Redis client remains application-owned and is not closed by the provider. Stream retention can be controlled with MaxLen, and Block / Count tune read behavior.

RabbitMQ#

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",
    }),
)

RabbitMQ publishes to a durable topic exchange and waits for publisher confirms. Consumers use manual acknowledgements: successful processing calls Ack, while decode or handler failures Nack and requeue the delivery.

The provider owns its AMQP connection and publisher channel. PrefetchCount controls the number of unacknowledged deliveries per consumer.

NATS JetStream#

store := GoEventBus.NewEventStore(
    &dispatcher,
    1024,
    GoEventBus.Block,
    GoEventBus.WithNATSJetStream(GoEventBus.NATSJetStreamProviderConfig{
        Conn:          nc,
        SubjectPrefix: "events.",
        Subject:       "events.>",
        Durable:       "billing",
        Queue:         "billing-workers",
    }),
)

JetStream publishes each event to SubjectPrefix + Projection. Durable consumers and queue groups are supported. Messages use explicit acknowledgements: Ack after successful handling and Nak after decode or handler failure for redelivery.

The injected NATS connection remains application-owned. GoEventBus closes only the subscription it creates.

Apache Kafka#

store := GoEventBus.NewEventStore(
    &dispatcher,
    1024,
    GoEventBus.Block,
    GoEventBus.WithKafka(GoEventBus.KafkaProviderConfig{
        Brokers: []string{"localhost:9092"},
        Topic:   "events",
        Group:   "billing",
    }),
)

Kafka uses the event projection as the default message key, so events with the same projection stay ordered within a partition. Consumer groups are supported, and offsets are committed synchronously only after successful event handling.

Handler or decode failures leave the offset uncommitted for redelivery according to Kafka consumer-group semantics. An optional fixed partition can be used when no consumer group is configured.

Operations

Dead letter queue#

Attach a DLQ to retain handler failures and recovered panics.

store.DLQ = GoEventBus.NewDeadLetterQueue()

Inspect failures:

for _, dead := range store.DLQ.Entries() {
    log.Printf(
        "event=%s attempts=%d err=%v",
        dead.Event.ID,
        dead.Attempts,
        dead.Err,
    )
}

Replay:

if err := store.DLQ.Replay(ctx, store); err != nil {
    log.Println("replay failed:", err)
}

Handler panics are recovered and converted into errors so a single handler cannot kill the worker pool.

Operations

Transactions#

Transactions are local event buffers, not database-style atomic transactions.

Direct transactions#

tx := store.BeginTransaction()

tx.Publish(GoEventBus.Event{
    ID:         "evt-1",
    Projection: "order.created",
    Data:       order,
})

tx.Publish(GoEventBus.Event{
    ID:         "evt-2",
    Projection: "invoice.created",
    Data:       invoice,
})

if err := tx.Commit(ctx); err != nil {
    tx.Rollback()
    log.Fatal(err)
}

Jev-aware transactions#

Use DecideAndPublish when a transaction entry needs the same Rules / Cache / Jev selection pipeline as DecideAndSubscribe.

tx := store.BeginTransaction()

decision, err := tx.DecideAndPublish(
    ctx,
    selector,
    map[string]any{
        "message": "Create an invoice for order 42",
    },
    GoEventBus.Event{
        ID:   "evt-3",
        Data: invoice,
    },
    []GoEventBus.EventCandidate{
        {
            Key:         "order_created",
            Projection:  "order.created",
            Description: "A new order should be created",
        },
        {
            Key:         "invoice_created",
            Projection:  "invoice.created",
            Description: "An invoice should be created",
        },
    },
)
if err != nil {
    tx.Rollback()
    log.Fatal(err)
}

fmt.Println("selected:", decision.Choice)

// No handler has run yet.
if err := tx.Commit(ctx); err != nil {
    tx.Rollback()
    log.Fatal(err)
}

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.

Rollback discards only events still buffered locally and never rewinds the shared EventStore ring.

Operations

Scheduling#

Direct scheduling#

Schedule a known event at a specific time:

timer := store.Schedule(
    ctx,
    time.Now().Add(10*time.Second),
    event,
)

Or after a duration:

timer := store.ScheduleAfter(
    ctx,
    30*time.Second,
    event,
)

Jev-aware scheduling#

Use the same Rules / Cache / Jev selector pipeline when the event type must be chosen before scheduling.

decision, timer, err := store.DecideAndSchedule(
    ctx,
    time.Now().Add(10*time.Second),
    selector,
    state,
    GoEventBus.Event{
        ID:   "evt-scheduled-1",
        Data: payload,
    },
    candidates,
)
if err != nil {
    log.Fatal(err)
}

fmt.Println("scheduled:", decision.Choice)

Or after a duration:

decision, timer, err := store.DecideAndScheduleAfter(
    ctx,
    30*time.Second,
    selector,
    state,
    GoEventBus.Event{
        ID:   "evt-scheduled-2",
        Data: payload,
    },
    candidates,
)
if err != nil {
    log.Fatal(err)
}

01Rules / Cache / Jev02Selected event03Timer04Handler

The selector runs when scheduling is requested, not when the timer fires. Selection errors are returned immediately and no timer is created. A successfully scheduled timer therefore contains an already-resolved event.

The returned *time.Timer may be stopped before it fires. Past times and non-positive durations execute immediately and return nil.

Operations

Metrics#

published, processed, failures := store.Metrics()

fmt.Printf(
    "published=%d processed=%d errors=%d\n",
    published,
    processed,
    failures,
)

These counters cover event execution. Decision-layer metrics such as cache hit rate or Jev latency are intentionally not baked into the core API yet.

Reference

API reference#

Event routing#

type EventCandidate struct {
    Key         string
    Projection  interface{}
    Description string
}

type EventDecision struct {
    Choice        string
    Projection    interface{}
    Confidence    float64
    Probabilities map[string]float64
    Model         string
    RequestID     string
}

type EventSelector interface {
    SelectEvent(
        context.Context,
        any,
        []EventCandidate,
    ) (EventDecision, error)
}

Main implementations:

JevSelector
RuleCacheSelector
MemoryDecisionCache

EventStore#

func NewEventStore(
    dispatcher *Dispatcher,
    bufferSize uint64,
    policy OverrunPolicy,
    options ...EventStoreOption,
) *EventStore

bufferSize must be a non-zero power of two.

Important methods:

Method	Purpose
`Subscribe`	Enqueue a known event
`DecideAndSubscribe`	Select a candidate and enqueue locally or publish to the configured broker
`Publish`	Dispatch pending events
`RegisterOrdered`	Preserve FIFO per ordering key
`RegisterBatch`	Process projection events in chunks
`Consume`	Feed configured or explicit provider events into the local store
`PublishToProvider`	Publish a known event through the provider configured on `NewEventStore`
`Use`	Register middleware
`OnBefore` / `OnAfter` / `OnError`	Register lifecycle hooks
`Metrics`	Read event counters
`Schedule` / `ScheduleAfter`	Schedule known events
`DecideAndSchedule` / `DecideAndScheduleAfter`	Select with Rules/Cache/Jev, then schedule the resolved event
`BeginTransaction`	Buffer events until commit; local stores execute regular handlers, provider-backed stores publish to the configured broker
`Drain` / `Close`	Stop accepting events and finish in-flight work

Reference

Examples#

The repository includes runnable examples for both the classic event-bus path and the intelligent routing layer.

Example	What it demonstrates
`routing_rules`	Deterministic first-match routing with `EventRule`; no model or cache required
`routing_cache`	Reusing a previous decision so the fallback selector is called only once
`routing_jev`	Full `rules → cache → Jev → GoEventBus` routing with OpenRouter
`transaction_jev`	Rules/Cache/Jev selection buffered until commit, then dispatched locally or published through the configured provider
`schedule_jev`	Rules/Cache/Jev event selection before delayed scheduling
`hello_world`	Minimal direct `Subscribe → Publish` flow
`middleware`	Handler middleware and lifecycle behavior
`goroutines-subscribe-publisher`	Concurrent producers and publishing
`drop_oldest`	`DropOldest` back-pressure
`return_error`	`ReturnError` back-pressure
`handler_timeout`	Handler context timeout
`publisher_timeout`	Blocking publisher timeout
`fasthttp`	HTTP integration

Run the new routing examples directly:

go run ./examples/routing_rules
go run ./examples/routing_cache
OPENROUTER_API_KEY=... go run ./examples/routing_jev
OPENROUTER_API_KEY=... go run ./examples/transaction_jev
OPENROUTER_API_KEY=... go run ./examples/schedule_jev

See examples/README.md for a compact guide to all examples.

Reference

Benchmarks#

Repository benchmarks on Apple M-series:

go test -bench . -benchtime=3s

Benchmark	Iterations	ns/op
`BenchmarkSubscribe`	27,080,376	40.37
`BenchmarkSubscribeParallel`	26,418,999	38.42
`BenchmarkPublish`	295,661,464	3.91
`BenchmarkPublishAfterPrefill`	252,943,526	4.59
`BenchmarkSubscribeLargePayload`	1,613,017	771.5
`BenchmarkPublishLargePayload`	296,434,225	3.91
`BenchmarkEventStore_Async`	2,816,988	436.5
`BenchmarkEventStore_Sync`	2,638,519	428.5
`BenchmarkFastHTTPSync`	6,275,112	163.8
`BenchmarkFastHTTPAsync`	1,954,884	662.0
`BenchmarkFastHTTPParallel`	4,489,274	262.3

The intelligent routing path is intentionally outside these core dispatch benchmarks because rules, cache, and Jev have very different latency profiles.

Reference

Architecture#

GoEventBus keeps decision-making and execution separate.

Decision layer                    Execution layer

rules ─┐
       ├─► choice ───────────────► EventStore
cache ─┤                           ├─ ring buffer
       │                           ├─ back-pressure
Jev ───┘                           ├─ worker pool
                                   ├─ ordering
                                   ├─ batching
                                   ├─ middleware/hooks
                                   └─ DLQ

This separation means:

the event bus does not depend on Jev,
deterministic workloads do not pay model latency,
model failures do not change core dispatch semantics,
alternative selectors can implement the same EventSelector interface,
alternative caches can implement DecisionCache,
application code can mix direct and intelligent routing in the same store.

Reference

Contributing#

Issues and pull requests are welcome.

Run the test suite with:

go test -race ./...

Integration tests:

go test -race -tags=integration -timeout=5m ./...

Quality checks used by CI include go vet and staticcheck.

Reference

License#

Distributed under the MIT License. See LICENSE.