Pub/Sub

The cpubsub package (from gocopper/pkg) allows your packages to publish events without knowing who consumes them. Two backends are included: an in-process local backend, and a Redis backend for when multiple processes need to hear about events. To get started, install the package:

go get github.com/gocopper/pkg/cpubsub

The PubSub Interface

Both backends implement the same small interface:

type PubSub interface {
	Subscribe(ctx context.Context, topic string, handler Handler) error
	Publish(ctx context.Context, topic string, payload any) error
}

Choose a backend by adding the corresponding wire module to your app:

cpubsub.WireModuleLocal // in-process, no dependencies

cpubsub.WireModuleRedis // requires you to provide a *redis.Client

Publishing Events

You should define topics as package constants so that subscribers may import them:

const TopicRocketLaunched = "rockets.launched"

err := s.pubsub.Publish(ctx, TopicRocketLaunched, LaunchedEvent{
	RocketID: rocket.ID,
})

If you are publishing as part of a database transaction, you may wish to publish from an OnCommit callback so that the event is only sent once the transaction commits. To learn more, see Transactions.

Subscribing to Topics

It is most common to subscribe in a constructor so that the subscription is in place when the app starts:

func NewTelemetry(p NewTelemetryParams) (*Telemetry, error) {
	t := &Telemetry{logger: p.Logger}

	err := p.PubSub.Subscribe(context.Background(), rockets.TopicRocketLaunched, t.handleRocketLaunched)
	if err != nil {
		return nil, cerrors.New(err, "failed to subscribe to rocket launches", nil)
	}

	return t, nil
}

func (t *Telemetry) handleRocketLaunched(ctx context.Context, payload any) error {
	// ...
	return nil
}

Handlers run on managed background goroutines, and subscriptions remain active until the app shuts down.

Payloads Across Backends

The local backend hands your handler the exact Go value you published. The Redis backend, however, serializes payloads as JSON, so handlers receive the decoded form (a map[string]any for structs). If a handler needs the typed struct, you may convert it by marshaling and unmarshaling again:

func (t *Telemetry) handleRocketLaunched(ctx context.Context, payload any) error {
	var event rockets.LaunchedEvent

	raw, err := json.Marshal(payload)
	if err != nil {
		return cerrors.New(err, "failed to marshal payload", nil)
	}

	if err := json.Unmarshal(raw, &event); err != nil {
		return cerrors.New(err, "failed to unmarshal launched event", nil)
	}

	// ...
	return nil
}

cpubsub is fire-and-forget: handler errors are logged but not retried, and there are no delivery guarantees or acknowledgments. In addition, the Redis backend broadcasts to every subscribed process. For work that must not be lost, or must run exactly once, consider persisting a job row and letting a worker claim it instead. See Background Work.

Getting Started

The Basics

Architecture

HTTP

Database

Frontend

Digging Deeper

Reference

The PubSub Interface

Publishing Events

Subscribing to Topics

Payloads Across Backends

​The PubSub Interface

​Publishing Events

​Subscribing to Topics

​Payloads Across Backends

The PubSub Interface

Publishing Events

Subscribing to Topics

Payloads Across Backends