GoForjUnreleased

Appearance

Support this libraryStar on GitHub

events logo

events is a typed event bus library for local dispatch and distributed pub/sub.

Go ReferenceLicense: MITGo TestGo versionLatest tagCodecovUnit tests (executed count)Integration tests (executed count)

What events is ​

events is a typed event bus for Go and handles event publication and fan-out. Durable background work such as retries and worker queues belongs in queue.

It lets applications publish and subscribe to events using normal Go types, with delivery handled either in-process or through distributed backends like NATS, Redis, Kafka, or Google Pub/Sub.

Installation ​

bash
go get github.com/goforj/events

All published modules support Go 1.24 and newer.

Quick Start ​

go
package main

import (
	"context"
	"fmt"

	"github.com/goforj/events"
)

type UserCreated struct {
	ID string `json:"id"`
}

func main() {
	bus, _ := events.NewSync()
	_, _ = bus.Subscribe(func(ctx context.Context, event UserCreated) error {
		fmt.Println("received", event.ID, ctx != nil)
		return nil
	})
	_ = bus.Publish(UserCreated{ID: "123"})
}

Topic Override ​

go
type UserCreated struct {
	ID string `json:"id"`
}

func (UserCreated) Topic() string { return "users.created" }

Driver Matrix ​

Optional distributed backends are separate modules. Keep dependencies lean and install only what you use:

bash
go get github.com/goforj/events/driver/natsevents
go get github.com/goforj/events/driver/redisevents
go get github.com/goforj/events/driver/kafkaevents
go get github.com/goforj/events/driver/natsjetstreamevents
go get github.com/goforj/events/driver/gcppubsubevents
go get github.com/goforj/events/driver/snsevents

Drivers ​

Each driver is thoroughly tested against the shared test suite using testcontainers or emulators where appropriate.

Driver / BackendModeFan-outDurableQueue SemanticsNotes
SyncIn-process✓xxRoot-backed synchronous dispatch in the caller path.
NullDrop-onlyxxxRoot-backed no-op transport for disabled eventing and tests.
NATSDistributed pub/sub✓xxSubject-based transport with live integration coverage.
NATS JetStreamDistributed stream✓PartialxEphemeral JetStream consumers preserve subscribe/close semantics while adding durable stream storage.
RedisDistributed pub/sub✓xxRedis pub/sub transport; Streams are intentionally deferred.
KafkaDistributed topic/log✓PartialxCurrent driver validates topic-based fan-out compatibility, not full consumer-group semantics.
SNSDistributed topic plus queue✓PartialxSNS fan-out with per-subscription SQS queues to preserve bus-style delivery semantics.
Google Pub/SubDistributed topic/subscription✓PartialxEmulator-backed Google Pub/Sub integration with per-subscription fan-out mapping.
SQSQueue targetPlanned✓✓Deferred until a separate async capability surface is intentionally introduced.

Driver Constructor Quick Examples ​

Use root constructors for local backends, and driver-module constructors for distributed backends. Driver backends live in separate modules so applications only import/link the optional dependencies they actually use.

go
package main

import (
	"context"

	"github.com/goforj/events"
	"github.com/goforj/events/driver/gcppubsubevents"
	"github.com/goforj/events/driver/kafkaevents"
	"github.com/goforj/events/driver/natsjetstreamevents"
	"github.com/goforj/events/driver/natsevents"
	"github.com/goforj/events/driver/redisevents"
	"github.com/goforj/events/driver/snsevents"
)

func main() {
	ctx := context.Background()

	events.NewSync()
	events.NewNull()

	natsevents.New(natsevents.Config{URL: "nats://127.0.0.1:4222"})
	natsjetstreamevents.New(natsjetstreamevents.Config{URL: "nats://127.0.0.1:4222"})
	redisevents.New(redisevents.Config{Addr: "127.0.0.1:6379"})
	kafkaevents.New(kafkaevents.Config{Brokers: []string{"127.0.0.1:9092"}})
	gcppubsubevents.New(ctx, gcppubsubevents.Config{
		ProjectID: "events-project",
		URI:       "127.0.0.1:8085",
	})
	snsevents.New(snsevents.Config{
		Region:   "us-east-1",
		Endpoint: "http://127.0.0.1:4566",
	})
}

Benchmarks ​

To refresh the live benchmark snapshot and regenerate the charts:

bash
sh scripts/refresh-bench-snapshot.sh

These charts compare one publish-plus-delivery round trip for sync and each enabled distributed driver fixture.

Note: sns and gcppubsub run through local emulators in this repo, so read those results as development approximations rather than direct managed-service latency comparisons.

Events backend latency chart

Events backend throughput chart

Events backend bytes chart

Events backend allocations chart

These checks are for obvious regression detection, not for noisy micro-optimism or hard CI performance gates.

API Index ​

GroupFunctions
BusDriver · Ready · WithContext
ConfigConfig · gcppubsubevents.Config · kafkaevents.Config · natsevents.Config · natsjetstreamevents.Config · redisevents.Config · snsevents.Config
ConstructionNew · NewNull · NewSync
Driver Constructorsgcppubsubevents.New · kafkaevents.New · natsevents.New · natsjetstreamevents.New · redisevents.New · snsevents.New
LifecycleClose
OptionsOption · WithCodec
PublishPublish · TopicEvent · TopicEvent.Topic
SubscribeSubscribe · Subscription
TestingFake · Fake.Bus · Fake.Count · Fake.Records · Fake.Reset · NewFake · Record

Bus ​

Driver ​

Driver reports the active backend.

go
bus, _ := events.NewSync()
fmt.Println(bus.Driver())
// Output: sync

Ready ​

Ready reports whether the bus is ready.

go
bus, _ := events.NewSync()
fmt.Println(bus.Ready() == nil)
// Output: true

WithContext ​

WithContext returns a derived bus handle bound to ctx for subsequent operations.

Config ​

Config ​

Config configures root bus construction.

Example: define bus construction config

go
cfg := events.Config{Driver: eventscore.DriverSync}

Example: define bus construction config with all fields

go
cfg := events.Config{
	Driver:    eventscore.DriverSync, // default: "sync" when empty and no Transport is provided
	Codec:     nil,                   // default: nil uses the built-in JSON codec
	Transport: nil,                   // default: nil keeps dispatch in-process
}

gcppubsubevents.Config ​

Config configures Google Pub/Sub transport construction.

Example: define Google Pub/Sub driver config

go
cfg := gcppubsubevents.Config{
	ProjectID: "events-project",
	URI:       "127.0.0.1:8085",
}

Example: define Google Pub/Sub driver config with all fields

go
cfg := gcppubsubevents.Config{
	ProjectID: "events-project",
	URI:       "127.0.0.1:8085", // default: "" is invalid unless Client is provided
	Client:    nil,              // default: nil creates a client from ProjectID and URI
}

kafkaevents.Config ​

Config configures Kafka transport construction.

Example: define Kafka driver config

go
cfg := kafkaevents.Config{Brokers: []string{"127.0.0.1:9092"}}

Example: define Kafka driver config with all fields

go
cfg := kafkaevents.Config{
	Brokers: []string{"127.0.0.1:9092"},
	Dialer:  nil, // default: nil uses a zero-value kafka.Dialer
	Writer:  nil, // default: nil builds a writer with single-message, auto-topic defaults
}

natsevents.Config ​

Config configures NATS transport construction.

Example: define NATS driver config

go
cfg := natsevents.Config{URL: "nats://127.0.0.1:4222"}

Example: define NATS driver config with all fields

go
cfg := natsevents.Config{
	URL:  "nats://127.0.0.1:4222",
	Conn: nil, // default: nil dials URL instead of reusing an existing connection
}

natsjetstreamevents.Config ​

Config configures NATS JetStream transport construction.

Example: define NATS JetStream driver config

go
cfg := natsjetstreamevents.Config{URL: "nats://127.0.0.1:4222"}

Example: define NATS JetStream driver config with all fields

go
cfg := natsjetstreamevents.Config{
	URL:               "nats://127.0.0.1:4222",
	Conn:              nil,                    // default: nil dials URL instead of reusing an existing connection
	SubjectPrefix:     "events.",              // default: "events."
	StreamNamePrefix:  "EVENTS_",              // default: "EVENTS_"
	InactiveThreshold: 30 * time.Second,       // default: 30s
	AckWait:           30 * time.Second,       // default: 30s
	FetchMaxWait:      250 * time.Millisecond, // default: 250ms
	Storage:           jetstream.MemoryStorage,// default: MemoryStorage
}

redisevents.Config ​

Config configures Redis transport construction.

Example: define Redis driver config

go
cfg := redisevents.Config{Addr: "127.0.0.1:6379"}

Example: define Redis driver config with all fields

go
cfg := redisevents.Config{
	Addr:   "127.0.0.1:6379",
	Client: nil, // default: nil constructs a client from Addr
}

snsevents.Config ​

Config configures SNS transport construction.

Example: define SNS driver config

go
cfg := snsevents.Config{
	Region:   "us-east-1",
	Endpoint: "http://127.0.0.1:4566",
}

Example: define SNS driver config with all fields

go
cfg := snsevents.Config{
	Region:            "us-east-1",
	Endpoint:          "http://127.0.0.1:4566", // default: "" uses normal AWS resolution
	SNSClient:         nil,                      // default: nil creates a client from Region and Endpoint
	SQSClient:         nil,                      // default: nil creates a client from Region and Endpoint
	TopicNamePrefix:   "events-",                // default: ""
	QueueNamePrefix:   "events-",                // default: ""
	WaitTimeSeconds:   1,                        // default: 1
	VisibilityTimeout: 30,                       // default: 30
}

Construction ​

New ​

New constructs a root bus for the requested driver.

go
bus, _ := events.New(events.Config{Driver: "sync"})
fmt.Println(bus.Driver())
// Output: sync

NewNull ​

NewNull constructs the root null bus.

go
bus, _ := events.NewNull()
fmt.Println(bus.Driver())
// Output: null

NewSync ​

NewSync constructs the root sync bus.

go
bus, _ := events.NewSync()
fmt.Println(bus.Driver())
// Output: sync

Driver Constructors ​

gcppubsubevents.New ​

New constructs a Google Pub/Sub-backed driver.

go
driver, _ := gcppubsubevents.New(context.Background(), gcppubsubevents.Config{
	ProjectID: "events-project",
	URI:       "127.0.0.1:8085",
})

kafkaevents.New ​

New constructs a Kafka-backed driver.

go
driver, _ := kafkaevents.New(kafkaevents.Config{Brokers: []string{"127.0.0.1:9092"}})

natsevents.New ​

New connects a NATS-backed driver from config.

go
driver, _ := natsevents.New(natsevents.Config{URL: "nats://127.0.0.1:4222"})

natsjetstreamevents.New ​

New connects a NATS JetStream-backed driver from config.

go
driver, _ := natsjetstreamevents.New(natsjetstreamevents.Config{URL: "nats://127.0.0.1:4222"})

redisevents.New ​

New constructs a Redis pub/sub-backed driver.

go
driver, _ := redisevents.New(redisevents.Config{Addr: "127.0.0.1:6379"})

snsevents.New ​

New constructs an SNS-backed driver.

go
driver, _ := snsevents.New(snsevents.Config{
	Region:   "us-east-1",
	Endpoint: "http://127.0.0.1:4566",
})

Lifecycle ​

Close ​

Close closes the underlying Pub/Sub client.

go
driver, _ := redisevents.New(redisevents.Config{Addr: "127.0.0.1:6379"})

Options ​

Option ​

Option configures root bus behavior.

WithCodec ​

WithCodec overrides the default event codec.

go
bus, _ := events.NewSync(events.WithCodec(nil))
fmt.Println(bus.Driver())
// Output: sync

Publish ​

Publish ​

Publish publishes an event using the background context.

go
type UserCreated struct {
	ID string `json:"id"`
}

bus, _ := events.NewSync()
_, _ = bus.Subscribe(func(event UserCreated) {
	fmt.Println(event.ID)
})
_ = bus.Publish(UserCreated{ID: "123"})
// Output: 123

TopicEvent ​

TopicEvent overrides the derived topic for an event.

TopicEvent.Topic ​

Topic returns the stable routing key shared by publishers and subscribers.

Subscribe ​

Subscribe ​

Subscribe registers a handler using the background context.

go
type UserCreated struct {
	ID string `json:"id"`
}

bus, _ := events.NewSync()
sub, _ := bus.Subscribe(func(ctx context.Context, event UserCreated) error {
	fmt.Println(event.ID)
	return nil
})
defer sub.Close()
_ = bus.Publish(UserCreated{ID: "123"})
// Output: 123

Subscription ​

Subscription releases a subscription when closed.

go
type UserCreated struct {
	ID string `json:"id"`
}

bus, _ := events.NewSync()
sub, _ := bus.Subscribe(func(event UserCreated) {
	fmt.Println("received", event.ID)
})
_ = bus.Publish(UserCreated{ID: "123"})
_ = sub.Close()
_ = bus.Publish(UserCreated{ID: "456"})
// Output: received 123

Testing ​

Fake ​

Fake provides a root-package testing helper that records published events.

go
fake := events.NewFake()
fmt.Println(fake.Count())
// Output: 0

Fake.Bus ​

Bus returns the wrapped API to inject into code under test.

go
fake := events.NewFake()
bus := fake.Bus()
fmt.Println(bus.Ready() == nil)
// Output: true

Fake.Count ​

Count returns the total number of recorded publishes.

go
type UserCreated struct {
	ID string `json:"id"`
}

fake := events.NewFake()
_ = fake.Bus().Publish(UserCreated{ID: "123"})
fmt.Println(fake.Count())
// Output: 1

Fake.Records ​

Records returns a copy of recorded publishes.

go
type UserCreated struct {
	ID string `json:"id"`
}

fake := events.NewFake()
_ = fake.Bus().Publish(UserCreated{ID: "123"})
fmt.Println(len(fake.Records()))
// Output: 1

Fake.Reset ​

Reset clears recorded publishes.

go
type UserCreated struct {
	ID string `json:"id"`
}

fake := events.NewFake()
_ = fake.Bus().Publish(UserCreated{ID: "123"})
fake.Reset()
fmt.Println(fake.Count())
// Output: 0

NewFake ​

NewFake creates a new fake event harness backed by the root sync bus.

go
fake := events.NewFake()
fmt.Println(fake.Count())
// Output: 0

Record ​

Record captures one published event observed by a Fake bus.

go
type UserCreated struct {
	ID string `json:"id"`
}

record := events.Record{Event: UserCreated{ID: "123"}}
fmt.Printf("%T\n", record.Event)
// Output: main.UserCreated

Development ​

Use make test for root-module tests, make vet for static checks, make generate to refresh generated documentation, and make test-integration for the separate integration module. Pass a driver such as make test-integration nats to narrow the matrix. Integration tests may require local services. Driver, docs, examples, and integration directories are independent Go modules; test each changed module from its directory.

Run make docs-watch to regenerate documentation as source files change.

Using with GoForj ​

GoForj Apps expose named event buses through generated accessors. Publish through those accessors and keep driver selection in event configuration.

For the App workflow, see Events.