Documentation
¶
Overview ¶
Package pubsub is the cross-instance message bus used by realtime to fan out database events, room messages and presence between server instances. Implementations: memory (single instance), redis (PUBLISH/SUBSCRIBE), postgres (LISTEN/NOTIFY). backends.Open builds one from configuration.
Index ¶
Constants ¶
const ( // MaxPayloadBytes is the largest payload Publish accepts. MaxPayloadBytes = 1 << 20 // MaxTopicBytes is the longest topic name Publish and Subscribe accept. MaxTopicBytes = 1024 )
Limits shared by every Broker implementation so that behaviour is identical across backends (a payload accepted by the memory broker in development is accepted by Redis or Postgres in production).
Variables ¶
var ( // ErrTooLarge is returned by Publish for payloads over MaxPayloadBytes. ErrTooLarge = errors.New("pubsub: payload too large") // ErrInvalidTopic is returned for empty, oversized or non-UTF-8 topics. ErrInvalidTopic = errors.New("pubsub: invalid topic") // ErrNilHandler is returned by Subscribe when the handler is nil. ErrNilHandler = errors.New("pubsub: nil handler") )
var ErrClosed = errors.New("pubsub: closed")
ErrClosed is returned after Close.
Functions ¶
func ValidatePublish ¶
ValidatePublish checks topic and payload size for Publish.
func ValidateTopic ¶
ValidateTopic reports whether topic is acceptable to every backend: non-empty, at most MaxTopicBytes, valid UTF-8 and free of control characters.
Types ¶
type Broker ¶
type Broker interface {
// Publish sends payload to every current subscriber of topic. Payloads may
// be up to 1 MiB (implementations handle transport limits transparently).
Publish(ctx context.Context, topic string, payload []byte) error
// Subscribe registers h for topic until cancel is called. Topics are exact
// strings (no wildcards), e.g. "rt.mysite.room.lobby".
Subscribe(ctx context.Context, topic string, h Handler) (cancel func(), err error)
// Close stops all subscriptions and releases resources.
Close() error
}
Broker publishes payloads to topics and delivers them to subscribers on all instances (including the publishing instance). Delivery is at-most-once and best-effort; ordering is preserved per topic per publisher.
type Handler ¶
type Handler func(payload []byte)
Handler receives a published payload. It is called sequentially per subscription (never concurrently for the same subscription) and must not block for long. The payload may be shared with other subscriptions of the same message and must be treated as read-only.
Each subscription has a bounded queue: a handler that falls far behind does not slow down publishers, but messages that overflow its queue are dropped (and logged).
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package backends constructs a pubsub.Broker from configuration.
|
Package backends constructs a pubsub.Broker from configuration. |
|
internal
|
|
|
fanout
Package fanout is the local delivery engine shared by the pubsub brokers.
|
Package fanout is the local delivery engine shared by the pubsub brokers. |
|
Package memory is the in-process pubsub.Broker used for single-instance deployments, development and tests.
|
Package memory is the in-process pubsub.Broker used for single-instance deployments, development and tests. |
|
Package postgres implements pubsub.Broker on PostgreSQL LISTEN/NOTIFY, so a horizontally scaled deployment that already uses Postgres for documents needs no extra infrastructure.
|
Package postgres implements pubsub.Broker on PostgreSQL LISTEN/NOTIFY, so a horizontally scaled deployment that already uses Postgres for documents needs no extra infrastructure. |
|
Package pubsubtest is the conformance suite every pubsub.Broker implementation must pass.
|
Package pubsubtest is the conformance suite every pubsub.Broker implementation must pass. |
|
Package redis implements pubsub.Broker on Redis (or Valkey, ElastiCache, Memorystore) PUBLISH/SUBSCRIBE.
|
Package redis implements pubsub.Broker on Redis (or Valkey, ElastiCache, Memorystore) PUBLISH/SUBSCRIBE. |