pubsub

type Backend ¶

type Backend interface {
	Publish(context.Context, Channel, []byte) (uint64, error)
	Subscribe(context.Context, []Channel, Limits) (Stream, error)
}

Backend exposes ephemeral fan-out, separate from jobs or durable events. Publish returns the authority's subscriber count, not a processing acknowledgement: memory counts subscriptions, Redis counts subscribed connections (one client multiplexes all of its subscriptions over one connection). Subscribe's context bounds establishment only; the returned stream is explicitly owned until Close or adapter shutdown. No implicit publish retry or fallback.

type Broker ¶

type Broker struct {
	// contains filtered or unexported fields
}

Broker borrows its adapter and owns callbacks and typed subscriptions. Pure construction performs no I/O. Close cancels work and waits for actual exit.

func NewBroker(backend Backend, config Config) (*Broker, error)

func (b *Broker) Close(ctx context.Context) error

func (b *Broker) Done() <-chan struct{}

func (b *Broker) Namespace() keyspace.Namespace

func (b *Broker) Stats() Stats

type Brokers ¶

type Brokers struct {
	// contains filtered or unexported fields
}

Brokers is immutable; default and named selection alias the same instance.

func NewBrokers(selected ConnectionName, entries ...NamedBroker) (*Brokers, error)

func (r *Brokers) Broker(name ConnectionName) (*Broker, error)

func (r *Brokers) Default() (*Broker, error)

func (r *Brokers) DefaultName() ConnectionName

func (r *Brokers) Names() []ConnectionName

type Channel ¶

type Channel struct {
	// contains filtered or unexported fields
}

Channel is an opaque, versioned adapter address. Redis database numbers do not isolate pub/sub; application/environment identity is always part of this address.

func NewChannel(namespace keyspace.Namespace, name Name, version Version, logical string) (Channel, error)

func (c Channel) Namespace() keyspace.Namespace

func (c Channel) String() string

func (c Channel) Validate() error

type Config ¶

type Config struct {
	Namespace                                                     keyspace.Namespace
	MaxDeclarations, MaxKeyBytes, MaxConcurrent, MaxSubscriptions int
	Timeout                                                       time.Duration
	Buffer                                                        Limits
}

Config bounds publication/setup callbacks and separately bounds subscriptions, each of which permits one active receive/decode operation at a time.

func DefaultConfig(namespace keyspace.Namespace) Config

func (c Config) Validate() error

type ConnectionName ¶

type ConnectionName string

ConnectionName selects an instance, distinct from names of other service families.

func (n ConnectionName) Validate() error

type Declaration ¶

type Declaration[K, V any] struct {
	// contains filtered or unexported fields
}

Declaration preserves both resource-key and JSON payload types. Increment Version for incompatible payload contracts; define once and reuse across callers.

func Define[K, V any](name Name, version Version, keys keyspace.Codec[K]) Declaration[K, V]

func (d Declaration[K, V]) Bind(b *Broker) (Topic[K, V], error)

func (d Declaration[K, V]) Name() Name

func (d Declaration[K, V]) Validate() error

func (d Declaration[K, V]) Version() Version

type Delivery ¶

type Delivery[K, V any] struct {
	Key   K
	Value V
}

Delivery pairs a received payload with the typed key it was published under.

type Gap ¶

type Gap struct {
	Cause    error
	Attempts int
}

Gap reports one interruption after which delivery has resumed. Payloads published between Cause and the new subscription's readiness were lost: Redis pub/sub is at-most-once. Attempts counts failed re-subscriptions.

type KeyedSubscription ¶

type KeyedSubscription[K, V any] struct {
	// contains filtered or unexported fields
}

KeyedSubscription is one typed stream over several keys of the same topic. It shares Subscription's ownership, sequential receive and failure semantics.

func (s *KeyedSubscription[K, V]) Close(ctx context.Context) error

func (s *KeyedSubscription[K, V]) Done() <-chan struct{}

func (s *KeyedSubscription[K, V]) Err() error

func (s *KeyedSubscription[K, V]) Receive(ctx context.Context) (Delivery[K, V], error)

type Limits ¶

type Limits struct{ Messages, Bytes, PayloadBytes, Channels int }

Limits bound a subscription's payloads and pending messages. Queue bytes count payload bytes; fixed-size message metadata is additionally bounded by Messages.

func DefaultLimits() Limits

func (l Limits) Validate() error

type Message ¶

type Message struct {
	Channel Channel
	Data    []byte
}

Message owns its payload. Adapter streams may multiplex exact channels; the typed application API retains its concrete topic and resource key boundary.

type Name ¶

type Name string

type NamedBroker ¶

type NamedBroker struct {
	Name  ConnectionName
	Value *Broker
}

NamedBroker borrows one existing instance. Selection never creates another owner.

type Stats ¶

type Stats struct {
	Operations, Subscriptions int
	Closing                   bool
}

Stats includes setup, live and draining subscriptions until their resources and active callbacks have finished. Reading it performs no I/O.

type Stream ¶

type Stream interface {
	Next(context.Context) (Message, error)
	Close(context.Context) error
	Done() <-chan struct{}
	Err() error
}

Stream is ready when Subscribe returns. Next cancellation stops that wait; Close owns the subscription lifetime and must unblock pending reads. Done closes only after adapter work exits. Errors never imply an uninterrupted subscription. Implementations must not silently drop messages or hide reconnection gaps.

type Subscription ¶

type Subscription[V any] struct {
	// contains filtered or unexported fields
}

Subscription owns one typed stream. Receive is sequential, including JSON decoding; concurrent receives fail instead of creating unbounded waiting work. Terminal stream/decoding failures discard pending delivery and close the stream.

func (s *Subscription[V]) Close(ctx context.Context) error

func (s *Subscription[V]) Done() <-chan struct{}

func (s *Subscription[V]) Err() error

func (s *Subscription[V]) Receive(ctx context.Context) (V, error)

type SupervisePolicy ¶

type SupervisePolicy struct {
	InitialBackoff, MaxBackoff time.Duration
}

SupervisePolicy bounds automatic re-subscription after an interruption. The delay starts at InitialBackoff, doubles (with jitter) after each failed attempt up to MaxBackoff, and resets once delivery has resumed.

func DefaultSupervisePolicy() SupervisePolicy

func (p SupervisePolicy) Validate() error

type Topic ¶

type Topic[K, V any] struct {
	// contains filtered or unexported fields
}

Topic retains concrete keys and payloads through publication and subscription.

func (t Topic[K, V]) Name() Name

func (t Topic[K, V]) Publish(ctx context.Context, key K, input V) (uint64, error)

func (t Topic[K, V]) Subscribe(ctx context.Context, key K) (*Subscription[V], error)

func (t Topic[K, V]) SubscribeKeys(ctx context.Context, first K, rest ...K) (*KeyedSubscription[K, V], error)

func (t Topic[K, V]) Supervise(ctx context.Context, key K, policy SupervisePolicy, receive func(context.Context, V) error, gap func(context.Context, Gap) error) error

func (t Topic[K, V]) Validate() error

func (t Topic[K, V]) Version() Version

type Version ¶

type Version uint32