EventPool

EventPool is a lightweight, process-local event bus for Go. It provides two delivery models:

• Broadcast delivers each message to every registered listener.
• Partitioned queue routes a key consistently and lets consumer groups process messages independently.

EventPool is intentionally in-memory. It does not provide persistence, cross-process delivery, acknowledgements, or recovery after a process crash. Use a durable broker when those guarantees are required.

Features

• Concurrent broadcast listeners with configurable worker and buffer counts.
• XXH3 key partitioning with sequential processing inside each consumer partition.
• Configurable retries and error, panic-recovery, and close hooks.
• Runtime listener registration.
• Idempotent Run and Close operations.
• Graceful shutdown that drains every accepted message.
• Context-aware publishing and shutdown for bounded backpressure.
• Zero-allocation dispatch with immutable payloads through SendBytes.

Installation

go get github.com/yudhasubki/eventpool

Broadcast

package main

import (
	"context"
	"log"

	"github.com/yudhasubki/eventpool"
)

func main() {
	events := eventpool.New()
	events.Submit(
		eventpool.EventpoolListener{
			Name: "metrics",
			Subscriber: func(name string, message []byte) error {
				log.Printf("%s received %s", name, message)
				return nil
			},
		},
		eventpool.EventpoolListener{
			Name: "cache",
			Subscriber: func(name string, message []byte) error {
				return nil
			},
		},
	)
	events.Run()

	if err := events.TryPublish(eventpool.SendString("order-created")); err != nil {
		log.Fatal(err)
	}

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

Publish remains available as a fire-and-forget compatibility API. Prefer TryPublish when serialization or lifecycle errors need to be observed, and PublishContext when a producer must be able to cancel while waiting for buffer space.

Partitioned queue

events := eventpool.NewPartition(16)
events.Submit(4, eventpool.EventpoolListener{
	Name: "billing",
	Subscriber: func(name string, message []byte) error {
		return charge(message)
	},
})
events.Run()

// Messages with the same key are routed to the same billing consumer partition.
err := events.TryPublish("billing", "customer-42", eventpool.SendString("invoice-1"))
if err != nil {
	log.Fatal(err)
}

events.Close()

The first NewPartition argument controls routing partitions. The first Submit argument controls the number of sequential consumer partitions inside each consumer group. An empty consumer-group name or "*" broadcasts to all groups in the selected routing partition.

Runtime registration

SubmitOnFlight registers missing listeners and starts them immediately in both delivery models:

events.SubmitOnFlight(eventpool.EventpoolListener{
	Name:       "audit",
	Subscriber: audit,
})

Calling Run again is safe; it will not create duplicate workers. Calling Submit after Run also starts replacement listeners automatically and drains the listeners they replace.

Subscriber options

eventpool.EventpoolListener{
	Name:       "worker",
	Subscriber: handle,
	Opts: []eventpool.SubscriberConfigFunc{
		eventpool.BufferSize(1_000),
		eventpool.MaxWorker(20),
		eventpool.MaxRetry(3),
		eventpool.ErrorHook(sendToDeadLetterStore),
		eventpool.RecoverHook(reportPanic),
		eventpool.CloseHook(cleanup),
	},
}

Defaults are a buffer size of 100, 10 broadcast workers, and 3 retries after the initial attempt. Partitioned consumers always use one worker per consumer partition to preserve ordering. A returned handler error is retried; after the limit is exhausted, ErrorHook runs. A handler panic runs RecoverHook and is considered consumed rather than retried.

Hooks run synchronously inside worker or shutdown paths and should return promptly. Panics raised by hooks are recovered.

Lifecycle and delivery semantics

• Publishing applies blocking backpressure when a listener buffer is full.
• Close rejects new messages, drains queued and active messages, then stops workers. It is safe to call more than once.
• CloseContext may return when its context expires; cleanup continues in the background.
• CloseBy provides the same drain guarantee for selected listeners.
• Cap reports queued plus actively processing messages.
• Broadcast subscribers receive the same message byte slice and must treat it as read-only.
• SendBytes avoids payload construction allocations, but its input must stay immutable for the lifetime of the returned MessageFunc.
• Queued or active messages can be lost if the process terminates, and handlers can be invoked more than once when retries are configured.

Testing and benchmarks

go test ./...
go test -race ./...
go test -bench=. -benchmem

Benchmark results depend heavily on partition count, worker count, buffer size, handler cost, and hardware. Run the included benchmarks on the target workload instead of relying on a fixed cross-machine number. See BENCHMARKS.md for the measurement methodology, configuration, and a five-sample reference run.

Contributing

Contributions and bug reports are welcome through the GitHub repository.

License

MIT