Documentation
¶
Overview ¶
Package streaming holds the machinery every long-lived gRPC stream client in this module shares: a stream whose underlying client is replaced on reconnect, the receive loop that drives it with backoff, and the error classification that decides between reconnecting and stopping.
Index ¶
- Constants
- Variables
- func Listen[C any, E any](ctx context.Context, stream *ReconnectingStream[C], recv func(C) (E, error), ...) error
- func SendListenerError(ctx context.Context, errCh chan<- error, err error)
- type Classifier
- type ReconnectingStream
- func (s *ReconnectingStream[C]) Close() error
- func (s *ReconnectingStream[C]) CloseStream() error
- func (s *ReconnectingStream[C]) ConnectOnce(ctx context.Context) error
- func (s *ReconnectingStream[C]) ConnectSync(ctx context.Context) error
- func (s *ReconnectingStream[C]) IsClosed() bool
- func (s *ReconnectingStream[C]) LifecycleContext() context.Context
- func (s *ReconnectingStream[C]) Name() string
- func (s *ReconnectingStream[C]) RetrySend(ctx context.Context, send func(C) error) error
- func (s *ReconnectingStream[C]) SendOnce(send func(C) error) error
- func (s *ReconnectingStream[C]) SetInitialClient(client C)
- func (s *ReconnectingStream[C]) SetSleep(sleep func(ctx context.Context, attempt int) error)
- func (s *ReconnectingStream[C]) Snapshot() (client C, generation uint64, ok bool)
- type Verdict
Constants ¶
const MaxConsecutiveNoProgress = 10
MaxConsecutiveNoProgress caps consecutive no-progress failures (recv or reconnect) before a listen loop surfaces the error to its owner.
Variables ¶
var ErrListenerClosed = errors.New("listener is closed")
ErrListenerClosed is returned by operations on a stream or listener after it was closed.
var ErrStreamNotConnected = errors.New("client is not connected")
ErrStreamNotConnected is returned by a send on a stream that has no client installed.
Functions ¶
func Listen ¶
func Listen[C any, E any]( ctx context.Context, stream *ReconnectingStream[C], recv func(C) (E, error), handle func(E) error, classify Classifier, ) error
Listen runs the receive loop for a reconnecting stream. Reconnects use full-jitter backoff; after MaxConsecutiveNoProgress consecutive no-progress failures the loop returns an error; classify decides clean vs error exits.
Context contract: ctx scopes the loop itself (recv classification and backoff sleeps; cancellation is a clean exit returning nil, because background loops shut down cleanly). Reconnect attempts always use stream.LifecycleContext(), which only Close() cancels, so one caller's ctx cannot destroy the shared stream's ability to reconnect.
If no client is installed yet, the first iteration connects (this serves StreamByAdditionalMetadata's initial connect and makes Listen() usable on a never-connected listener).
Types ¶
type Classifier ¶
Classifier maps a stream error to a verdict. The listen loop consults it once per error, so implementations may carry per-error side effects (e.g. the action listener's V2→V1 strategy fallback).
func NewClassifier ¶
func NewClassifier(reconnectOnEOF func(ctx context.Context) bool) Classifier
NewClassifier builds the default classifier. reconnectOnEOF is consulted every time io.EOF is observed because handler registration can change between errors.
type ReconnectingStream ¶
type ReconnectingStream[C any] struct { // contains filtered or unexported fields }
ReconnectingStream is one logical gRPC stream whose underlying client can be replaced on reconnect. connectGroup (singleflight) coalesces concurrent connect attempts so only one replacement stream opens. Connect runs outside mu so snapshots do not block behind network I/O. replay re-sends current subscriptions on a fresh stream before install. lifecycleCtx lasts until Close; caller contexts must not kill shared listeners. NOTE: field order follows govet fieldalignment (enforced by the pre-commit autofixer); mu guards client, generation, hasClient, and closed.
func NewReconnectingStream ¶
func NewReconnectingStream[C any]( l *zerolog.Logger, name string, constructor func(context.Context) (C, error), closeSend func(C) error, replay func(context.Context, C) error, ) *ReconnectingStream[C]
NewReconnectingStream builds a stream whose lifecycle ends only with Close. constructor opens a new client, closeSend half-closes a retired one, and replay (optional) brings a fresh client up to date before it is published.
func NewReconnectingStreamWithLifecycle ¶
func NewReconnectingStreamWithLifecycle[C any]( parent context.Context, l *zerolog.Logger, name string, constructor func(context.Context) (C, error), closeSend func(C) error, replay func(context.Context, C) error, ) *ReconnectingStream[C]
NewReconnectingStreamWithLifecycle is NewReconnectingStream with a parent for the lifecycle context, so cancelling parent also ends the stream.
func (*ReconnectingStream[C]) Close ¶
func (s *ReconnectingStream[C]) Close() error
Close ends the stream for good: the lifecycle context is cancelled and the current client is half-closed.
func (*ReconnectingStream[C]) CloseStream ¶
func (s *ReconnectingStream[C]) CloseStream() error
CloseStream half-closes the current client without closing the stream, so the receive loop reconnects.
func (*ReconnectingStream[C]) ConnectOnce ¶
func (s *ReconnectingStream[C]) ConnectOnce(ctx context.Context) error
ConnectOnce makes one connect attempt, coalesced with any concurrent one, and publishes the new client after replay.
func (*ReconnectingStream[C]) ConnectSync ¶
func (s *ReconnectingStream[C]) ConnectSync(ctx context.Context) error
ConnectSync connects with bounded retries and backoff, returning the last error when every attempt fails.
func (*ReconnectingStream[C]) IsClosed ¶
func (s *ReconnectingStream[C]) IsClosed() bool
IsClosed reports whether Close has been called.
func (*ReconnectingStream[C]) LifecycleContext ¶
func (s *ReconnectingStream[C]) LifecycleContext() context.Context
LifecycleContext is the context that only Close cancels. Reconnects and background loops that must outlive any one caller use it.
func (*ReconnectingStream[C]) Name ¶
func (s *ReconnectingStream[C]) Name() string
Name is the stream's name in log messages.
func (*ReconnectingStream[C]) RetrySend ¶
func (s *ReconnectingStream[C]) RetrySend(ctx context.Context, send func(C) error) error
RetrySend sends with bounded retries. Each failed attempt makes at most one reconnect attempt (ConnectOnce, coalesced with any concurrent reconnect via singleflight) before backing off, so the total budget is StreamSyncMaxAttempts sends, at most StreamSyncMaxAttempts reconnects, and at most StreamSyncMaxAttempts-1 backoff sleeps. A reconnect failure that is permanent (ErrListenerClosed or classified StreamDecisionStop) short-circuits immediately.
func (*ReconnectingStream[C]) SendOnce ¶
func (s *ReconnectingStream[C]) SendOnce(send func(C) error) error
SendOnce performs one send on the current client under sendMu and never reconnects. Callers that keep their own record of what was sent use it so a reconnect's replay is the only path that sends the same message again.
func (*ReconnectingStream[C]) SetInitialClient ¶
func (s *ReconnectingStream[C]) SetInitialClient(client C)
SetInitialClient installs client without a connect when none is installed yet; a later ConnectOnce replaces it like any other.
func (*ReconnectingStream[C]) SetSleep ¶
func (s *ReconnectingStream[C]) SetSleep(sleep func(ctx context.Context, attempt int) error)
SetSleep replaces the backoff sleep used between reconnect and send attempts. Tests use it to disable backoff; it is not safe to call once the stream is in use.
func (*ReconnectingStream[C]) Snapshot ¶
func (s *ReconnectingStream[C]) Snapshot() (client C, generation uint64, ok bool)
Snapshot returns the current client, its generation, and whether one is installed. It never blocks behind network I/O.
type Verdict ¶
type Verdict int
Verdict is the single classification of an error observed on a long-lived stream, for both recv and reconnect failures.
const ( // VerdictRetry reconnects without counting toward the no-progress cap. VerdictRetry Verdict = iota // VerdictNoProgress reconnects and counts toward MaxConsecutiveNoProgress. VerdictNoProgress // VerdictStopClean exits the listen loop returning nil. VerdictStopClean // VerdictStopError exits the listen loop returning the error. VerdictStopError )
Source Files