Documentation
¶
Overview ¶
Package watch implements the gRPC WatchService that streams ordered events from the control plane's OrderedLog to subscribers.
The server is a thin adapter: it delegates ordering, retention, and backfill semantics to the OrderedLog backend. Its only added responsibility is wire-protocol concerns: per-stream send buffer, SLOW_CONSUMER detection, COMPACTED status, and graceful shutdown.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is a resilient client for WatchService. It reconnects on transport errors with exponential backoff, advances fromSeq across reconnects so no event is delivered twice, and invokes the snapshot handler when the server reports COMPACTED.
func NewClient ¶
func NewClient(conn grpc.ClientConnInterface, startSeq uint64, handler EventHandler, snapshot SnapshotHandler, opts ClientOptions) *Client
NewClient constructs a Client over an existing grpc.ClientConn. startSeq is the seq immediately preceding the first event the client wants (i.e. delivery starts at startSeq+1). Pass 0 to receive everything in retention.
type ClientOptions ¶
type ClientOptions struct {
// ClientID is reported to the server for logs / metrics.
ClientID string
// InitialBackoff / MaxBackoff bound the reconnect backoff after
// transport errors. Defaults: 250ms / 30s.
InitialBackoff time.Duration
MaxBackoff time.Duration
Logger log.Logger
}
ClientOptions tunes the watch client.
type EventHandler ¶
type EventHandler func(ctx context.Context, ev orderedlog.Event) error
EventHandler processes events delivered to the client. It MUST NOT block for an unbounded amount of time: a slow handler will cause the server to drop the stream as a slow consumer. Handlers should either persist quickly or hand off to a worker. Returning a non-nil error terminates the watch loop with that error.
type Server ¶
type Server struct {
pb.UnimplementedWatchServiceServer
// SendTimeout is the per-event send deadline. If a send blocks
// longer than this, the client is declared slow and the stream is
// terminated with CODE_SLOW_CONSUMER.
SendTimeout time.Duration
// contains filtered or unexported fields
}
Server implements pb.WatchServiceServer over an orderedlog.OrderedLog.
func NewServer ¶
func NewServer(olog orderedlog.OrderedLog, logger log.Logger) *Server
NewServer constructs a Server backed by olog.
func (*Server) Close ¶
func (s *Server) Close()
Close signals all in-flight Subscribe streams to terminate with CODE_SHUTTING_DOWN. Idempotent.
func (*Server) Subscribe ¶
func (s *Server) Subscribe(req *pb.SubscribeRequest, stream pb.WatchService_SubscribeServer) error
Subscribe streams ordered events from req.FromSeq forward.
type SnapshotHandler ¶
SnapshotHandler is invoked when the server reports CODE_COMPACTED. The handler must take a fresh snapshot of state from the control plane (out-of-band) and return the seq it observed. The watch loop resumes from snapshotSeq + 1.
Source Files