Documentation
¶
Overview ¶
Package server provides the owner-side primitives for building a service that pushes cache-invalidation events to coherent consumers.
It contains two building blocks, both transport-agnostic:
ConnectionManager broadcasts events to all currently-connected consumers with non-blocking sends, so one slow consumer cannot stall the others.
ReplayService replays events a reconnecting consumer missed, based on a timestamp watermark, over any RecordReader (Kafka, a log, etc.).
A correct subscribe handler must REGISTER the consumer with the ConnectionManager BEFORE it starts replay, then drain buffered live events, then stream live. Registering first guarantees no event is lost in the gap between reconnect and the end of replay. See the package example.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ConnectionManager ¶
type ConnectionManager[E any] struct { // contains filtered or unexported fields }
ConnectionManager maintains a registry of connected consumers and broadcasts events to all of them using non-blocking sends. An event destined for a consumer whose buffer is full is dropped and counted rather than blocking the broadcast; the consumer self-heals via cache-clear-on-reconnect and TTL.
ConnectionManager is safe for concurrent use.
func NewConnectionManager ¶
func NewConnectionManager[E any](bufSize int) *ConnectionManager[E]
NewConnectionManager returns a manager whose per-consumer channels have the given buffer size. A bufSize below 1 is raised to 1.
func (*ConnectionManager[E]) Active ¶
func (m *ConnectionManager[E]) Active() int
Active returns the number of connected consumers.
func (*ConnectionManager[E]) Broadcast ¶
func (m *ConnectionManager[E]) Broadcast(ev E)
Broadcast attempts a non-blocking send of ev to every connected consumer. Sends to full channels are dropped and counted (see Dropped).
func (*ConnectionManager[E]) Deregister ¶
func (m *ConnectionManager[E]) Deregister(id string)
Deregister removes id and closes its channel. It is safe to call more than once.
func (*ConnectionManager[E]) Dropped ¶
func (m *ConnectionManager[E]) Dropped() int64
Dropped returns the total number of broadcast sends dropped due to full consumer buffers.
func (*ConnectionManager[E]) Register ¶
func (m *ConnectionManager[E]) Register(id string) <-chan E
Register allocates a buffered channel for id and returns it for the caller to stream from. It MUST be called before starting replay for that consumer. If id is already registered, the previous channel is closed and replaced.
func (*ConnectionManager[E]) Sent ¶
func (m *ConnectionManager[E]) Sent() int64
Sent returns the total number of successful broadcast sends.
type Record ¶
Record is a single stored event to be replayed. TimestampMs is its source time in Unix milliseconds (informational); Value is the opaque encoded event.
type RecordReader ¶
type RecordReader interface {
// Seek positions the reader just after resumeAfterMs and reports whether a
// retention gap exists — that is, whether resumeAfterMs precedes the earliest
// still-available record, meaning the consumer's history cannot be fully
// reconstructed.
Seek(ctx context.Context, resumeAfterMs int64) (gap bool, err error)
// Next returns the next record. ok is false when the reader has caught up to
// the present (within its lag tolerance) or ctx is done; err is non-nil only
// on a genuine failure.
Next(ctx context.Context) (rec Record, ok bool, err error)
// Close releases reader resources.
Close() error
}
RecordReader streams stored records in timestamp order for replay. It abstracts the durable log (Kafka, a database, an object store, ...). Implementations should create an ephemeral, non-committing reader so replay does not disturb primary consumer offsets.
type ReplayService ¶
type ReplayService struct {
// contains filtered or unexported fields
}
ReplayService replays events a reconnecting consumer missed, based on a watermark, over a caller-provided RecordReader.
func NewReplayService ¶
func NewReplayService(newReader func() (RecordReader, error)) *ReplayService
NewReplayService returns a ReplayService that obtains a fresh RecordReader for each replay via newReader.
func (*ReplayService) Replay ¶
func (s *ReplayService) Replay( ctx context.Context, resumeAfterMs int64, send func(value []byte) error, sendClear func() error, ) error
Replay streams the records after resumeAfterMs to send, in order.
If Seek reports a retention gap (resumeAfterMs is older than the earliest retained record), the consumer's history cannot be reconstructed: sendClear is invoked once to instruct the consumer to flush its cache and lazily re-fill, and Replay returns without streaming further. Otherwise Replay streams every record after the resume point and returns nil once the reader has caught up.
Source Files