kelindar/event

The kelindar/event package is a high-performance, in-process event dispatcher for Go. It is designed for decoupling modules within a single process using a generic pub/sub pattern.

Key Characteristics:

• High Throughput: Capable of processing millions of events per second (often 4x to 10x faster than standard Go channels).
• Asynchronous: Each subscriber is executed in its own goroutine.
• Generic: Works with any type implementing the Event interface.

When to use:

• Decoupling internal modules in a single Go process.
• Lightweight, high-throughput, low-latency event-driven patterns.

When NOT to use:

• For inter-process or service-to-service communication (use NATS, Kafka, etc.).
• When you need event persistence, durability, or replay capabilities.
• For scenarios involving heavy subscribe/unsubscribe churn.

To use kelindar/event, you must define event types that implement the Event interface. An event type must provide a Type() uint32 method that returns a unique identifier for that event type. This allows the dispatcher to route events to the correct subscribers.

// Define a unique event type ID
const EventA = 0x01

// Define your event structure
type myEvent struct{
    Data string
}

// Implement the Event interface
func (ev myEvent) Type() uint32 {
    return EventA
}

For better isolation or when managing multiple event buses, create a dedicated Dispatcher using event.NewDispatcher().

When using a specific dispatcher, use the generic event.Subscribe[T]() and event.Publish[T]() functions, passing the dispatcher instance as the first argument. This ensures that events are only routed within that specific bus instance.

bus := event.NewDispatcher()

// Subscribe to a specific event type on the custom bus
defer event.Subscribe(bus, func(e Event) {
    println("(consumer 1)", e.Data)
})()

// Publish an event to the custom bus
event.Publish(bus, newEventA("event 1"))

For simple use cases, you can use the package-level functions event.On() and event.Emit(). This uses a default global dispatcher.

event.On() returns an unsubscription function that should be called (typically via defer) to stop receiving events. Subscribers run in their own goroutines, so event handling is asynchronous and non-blocking.

// Subscribe to events using the global dispatcher
// The returned function is used to unsubscribe
defer event.On(func(e Event) {
    println("(consumer)", e.Data)
})()

// Publish events using the global dispatcher
event.Emit(newEventA("event 1"))
event.Emit(newEventA("event 2"))

Call Close() on the Dispatcher to signal that it is shutting down. This closes the internal done channel, which stops the background processing loops for all subscriber groups.

err := broker.Close()
if err != nil {
	// handle error
}

The Subscribe[T Event] function allows you to register a handler for a specific event type. The event type is automatically inferred from the type T provided. It returns a context.CancelFunc which, when called, unsubscribes the handler.

// Define your event
type UserCreated struct { Name string }
func (u UserCreated) Type() uint32 { return 101 }

// Subscribe
cancel := event.Subscribe(broker, func(ev UserCreated) {
	fmt.Printf("User created: %s\n", ev.Name)
})

// Later, to unsubscribe:
cancel()

Use NewDispatcher() to create a high-performance, in-process event dispatcher. By default, it uses a flush interval of 500 microseconds and a maximum queue size of 50,000 per consumer to handle high throughput (up to ~1 million events/second).

import "github.com/kelindar/kelindar/event"

broker := event.NewDispatcher()
defer broker.Close()

If you need to specify the event type explicitly (e.g., if you aren't using type inference), use SubscribeTo. This function takes the broker, the eventType (uint32), and a handler function of type func(T).

// Explicitly providing the event type
cancel := event.SubscribeTo(broker, 101, func(ev UserCreated) {
	fmt.Println(ev.Name)
})

Use Publish(broker, ev) to send an event to the dispatcher. The dispatcher will look up the corresponding subscriber group using the event's Type() and broadcast the event to all active consumers in that group.

ev := UserCreated{Name: "Alice"}
event.Publish(broker, ev)

The On[T Event](handler func(T)) function subscribes to an event using the Default dispatcher. The event type is automatically inferred from the type T provided in the handler function. For type inference to work correctly, the event type must be a constant.

Returns a context.CancelFunc which, when called, unsubscribes the handler from the dispatcher.

cancel := event.On(func(ev MyEventType) {
    // handle event
})