Documentation
¶
Overview ¶
Package pubsub is a tiny message-fanout abstraction with pluggable backends. Standalone deployments use MemoryPubSub (same-process, zero-config); HA deployments can opt into a RESP-protocol backend (Redis or Valkey) for cross-node delivery.
The package is deliberately shaped like pkg/cache: one interface, one in-memory default, one or more opt-in HA backends, one factory. See Service for the entry point.
Index ¶
Constants ¶
This section is empty.
Variables ¶
ErrBackendUnavailable is logged (not returned) by Service.NewPubSub when an HA backend is requested but its dependency (a redis.UniversalClient) is nil. The factory then yields a MemoryPubSub with a nil error so a misconfigured HA deployment degrades to single-node behaviour instead of refusing to start. Callers cannot use errors.Is(err, ErrBackendUnavailable) to detect the fallback; inspect the log stream or the Service's configured backend instead.
var ErrClosed = errors.New("pubsub: closed")
ErrClosed is returned by PubSub operations on a closed instance.
Functions ¶
func NewClient ¶
func NewClient(cfg ClientConfig) (redis.UniversalClient, error)
NewClient builds a redis.UniversalClient from cfg. Returns nil without error when cfg has no addresses, so callers can treat "Redis section omitted" as "no HA pub/sub" without extra branching.
Types ¶
type Backend ¶
type Backend int
Backend selects which PubSub implementation Service.NewPubSub returns. BackendMemory is the zero value and the standalone default.
const ( // BackendMemory selects MemoryPubSub, a single-process bus. // Standalone deployments and tests use this. BackendMemory Backend = iota // BackendRedis selects a Redis Pub/Sub-backed bus for cross-node // delivery in HA mode. BackendRedis // BackendValkey selects a Valkey Pub/Sub-backed bus. Valkey speaks // the same RESP protocol as Redis, so the underlying implementation // is shared; the separate value exists so operators configure the // backend they actually run and so logs identify it correctly. BackendValkey )
func ParseBackend ¶
ParseBackend maps a config string ("memory"/"redis"/"valkey") to a Backend. Unknown and empty values map to BackendMemory so a missing config falls back to the standalone default without erroring.
type ClientConfig ¶
type ClientConfig struct {
// Addrs lists one or more "<host>:<port>" endpoints. A single entry
// yields a plain client; multiple entries yield a cluster client,
// matching redis.UniversalOptions semantics.
Addrs []string
// Username and Password carry ACL credentials (Redis 6+ / Valkey)
// or the single AUTH password on older servers.
Username string
Password string
// DB is the logical database number used by single-node mode
// (ignored in cluster mode).
DB int
// TLS enables TLS for the connection using the system root CAs.
// Finer-grained control (custom CA, mTLS) belongs behind a dedicated
// builder; this one is deliberately boring.
TLS bool
}
ClientConfig captures the fields needed to construct a redis.UniversalClient. Low-level builder input; the deployment-facing yaml-tagged shape lives in pubsub.Config, and cmd/verifier maps between the two.
type Config ¶
type Config struct {
// Backend selects the RESP backend; "redis" or "valkey". Defaults
// to "redis" when omitted.
Backend string `yaml:"backend" validate:"omitempty,oneof=redis valkey" default:"redis"`
// Addrs lists one or more "<host>:<port>" endpoints. A single entry
// yields a plain client; multiple entries yield a cluster client.
Addrs []string `yaml:"addrs" validate:"required,min=1,dive,hostname_port" doc_example:"[\"redis:6379\"]"`
// Username is the ACL username (Redis 6+ / Valkey). Optional.
Username string `yaml:"username,omitempty"`
// Password is the ACL password (Redis 6+ / Valkey) or the single
// AUTH password on older servers. Optional.
Password string `yaml:"password,omitempty"`
// DB is the logical database number used by single-node mode
// (ignored in cluster mode).
DB int `yaml:"db" default:"0"`
// TLS enables TLS for the client connection, using system roots.
TLS bool `yaml:"tls" default:"false"`
}
Config is the deployment-facing configuration for the pub/sub bus.
Supports Redis and Valkey today (both speak RESP, so the same client serves both; Backend only decides what logs and metrics identify it as). Omitted in standalone deployments; when omitted in HA, notifications stay in-process and do not cross nodes.
type Logger ¶
Logger mirrors pkg/cache.Logger: a minimal operational-error sink that both *logger.Log and logr.Logger satisfy. Nil is accepted everywhere and swapped for a silent no-op.
type MemoryPubSub ¶
type MemoryPubSub struct {
// contains filtered or unexported fields
}
MemoryPubSub is a single-process PubSub implementation: publish and subscribe share the same goroutine-safe map of topic groups. It is the standalone default and the test backend; cross-node delivery is outside its remit.
The lifecycle contract matches the needs of SSE fanout:
- Publish to a topic with no subscriber is a no-op (does not create a lingering group).
- A subscriber whose channel is not being drained drops messages rather than back-pressuring the publisher.
- Publish after Close returns ErrClosed; the per-group close path also makes a racing in-flight Publish observe the torn-down state before it tries to send on closed subscriber channels.
func NewMemoryPubSub ¶
func NewMemoryPubSub() *MemoryPubSub
NewMemoryPubSub returns a ready-to-use in-process PubSub.
func (*MemoryPubSub) Close ¶
func (m *MemoryPubSub) Close() error
Close terminates every topic group and lets outstanding Subscriptions observe end-of-stream on their next read. Idempotent.
func (*MemoryPubSub) HealthProbe ¶
func (m *MemoryPubSub) HealthProbe(_ context.Context) error
HealthProbe returns ErrClosed after Close has run, nil otherwise. The in-process bus has no network to probe.
func (*MemoryPubSub) Subscribe ¶
func (m *MemoryPubSub) Subscribe(_ context.Context, topic string) (Subscription, error)
Subscribe returns a fresh Subscription for topic. Each call allocates a new buffered channel; the Subscription owns it until Close.
type PubSub ¶
type PubSub interface {
// Publish sends payload to every live subscriber of topic. It is
// a no-op when no subscriber exists. Non-blocking: a slow
// subscriber loses the message rather than back-pressuring the
// publisher.
Publish(ctx context.Context, topic string, payload []byte) error
// Subscribe returns a Subscription delivering future publishes on
// topic. The caller MUST Close the returned Subscription when
// done to release backend resources.
Subscribe(ctx context.Context, topic string) (Subscription, error)
// Close releases backend resources and closes every outstanding
// Subscription. Safe to call multiple times.
Close() error
// HealthProbe reports whether the backing bus is reachable. Returns
// nil when healthy; a non-nil error surfaces unchanged to the
// aggregator (and to the SSE path, where a dead HA bus would
// otherwise leave every cross-node notification silently dropped).
HealthProbe(ctx context.Context) error
}
PubSub is a topic-based publish/subscribe bus. Payloads are opaque bytes; callers choose their own encoding.
Delivery is best-effort fire-and-forget: a message published to a topic with no live subscriber is dropped, and a subscriber whose channel is not being drained drops the message rather than blocking publishers. Durable delivery is a caller concern - see internal/verifier/apiv1 for the completion-marker pattern that handles reload survival for the verifier notify flow.
func NewRedisPubSub ¶
NewRedisPubSub creates a Redis-backed PubSub. prefix namespaces every channel name as "<prefix>:<topic>" so multiple buses may share one Redis keyspace without colliding. client ownership stays with the caller.
func NewValkeyPubSub ¶
NewValkeyPubSub creates a Valkey-backed PubSub. Valkey is the Linux-Foundation-maintained Redis fork and speaks the same RESP protocol, so the implementation is shared with NewRedisPubSub; the separate constructor exists so operators pick the backend they actually run and so logs identify it correctly.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service manages PubSub creation. One Service per process; one PubSub per logical bus (there is typically only one today). The factory shape mirrors pkg/cache.Service.
func New ¶
func New(backend Backend, redisClient redis.UniversalClient, log Logger) *Service
New creates a Service.
backend picks the implementation. When backend is BackendRedis or BackendValkey the supplied redis.UniversalClient MUST be non-nil; a nil client falls back to MemoryPubSub with a logged warning so a misconfigured HA deployment degrades gracefully. The caller owns the client lifecycle.
log is used for operational errors from the backing bus. Nil is accepted and swapped for a silent no-op.
func (*Service) NewPubSub ¶
NewPubSub builds a PubSub for the configured backend. prefix is used only by the RESP backends to namespace channel names so multiple buses can share one Redis/Valkey keyspace without colliding; it is ignored by MemoryPubSub.
RESP paths route through newRESP so the prefix validation enforced by the public NewRedisPubSub/NewValkeyPubSub constructors applies here too; a prefix containing glob metacharacters would otherwise slip past the factory and poison SUBSCRIBE pattern matching.
type Subscription ¶
type Subscription interface {
// C returns the delivery channel. It is closed by Close and by
// Close on the owning PubSub; readers should treat a closed
// channel as "subscription ended" rather than an error.
C() <-chan []byte
// Close cancels the subscription and closes C. Idempotent.
Close() error
}
Subscription is a single subscriber's handle to a topic. Each Subscribe call returns a fresh Subscription with its own channel; Close is the only path to release it.
Source Files