const EventTypeDelete … const EventTypePut … const closeSendErrTimeout … const AutoWatchID … const InvalidWatchID … type Event … type WatchChan … type Watcher … type WatchResponse … // IsCreate returns true if the event tells that the key is newly created. func (e *Event) IsCreate() bool { … } // IsModify returns true if the event tells that a new value is put on existing key. func (e *Event) IsModify() bool { … } // Err is the error value if this WatchResponse holds an error. func (wr *WatchResponse) Err() error { … } // IsProgressNotify returns true if the WatchResponse is progress notification. func (wr *WatchResponse) IsProgressNotify() bool { … } type watcher … type watchGrpcStream … type watchStreamRequest … type watchRequest … type progressRequest … type watcherStream … func NewWatcher(c *Client) Watcher { … } func NewWatchFromWatchClient(wc pb.WatchClient, c *Client) Watcher { … } var valCtxCh … var zeroTime … type valCtx … func (vc *valCtx) Deadline() (time.Time, bool) { … } func (vc *valCtx) Done() <-chan struct{ … } func (vc *valCtx) Err() error { … } func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream { … } // Watch posts a watch request to run() and waits for a new watcher channel func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan { … } func (w *watcher) Close() (err error) { … } // RequestProgress requests a progress notify response be sent in all watch channels. func (w *watcher) RequestProgress(ctx context.Context) (err error) { … } func (w *watchGrpcStream) close() (err error) { … } func (w *watcher) closeStream(wgs *watchGrpcStream) { … } func (w *watchGrpcStream) addSubstream(resp *pb.WatchResponse, ws *watcherStream) { … } func (w *watchGrpcStream) sendCloseSubstream(ws *watcherStream, resp *WatchResponse) { … } func (w *watchGrpcStream) closeSubstream(ws *watcherStream) { … } // run is the root of the goroutines for managing a watcher client func (w *watchGrpcStream) run() { … } // nextResume chooses the next resuming to register with the grpc stream. Abandoned // streams are marked as nil in the queue since the head must wait for its inflight registration. func (w *watchGrpcStream) nextResume() *watcherStream { … } // dispatchEvent sends a WatchResponse to the appropriate watcher stream func (w *watchGrpcStream) dispatchEvent(pbresp *pb.WatchResponse) bool { … } // broadcastResponse send a watch response to all watch substreams. func (w *watchGrpcStream) broadcastResponse(wr *WatchResponse) bool { … } // unicastResponse sends a watch response to a specific watch substream. func (w *watchGrpcStream) unicastResponse(wr *WatchResponse, watchId int64) bool { … } // serveWatchClient forwards messages from the grpc stream to run() func (w *watchGrpcStream) serveWatchClient(wc pb.Watch_WatchClient) { … } // serveSubstream forwards watch responses from run() to the subscriber func (w *watchGrpcStream) serveSubstream(ws *watcherStream, resumec chan struct{ … } func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) { … } func (w *watchGrpcStream) waitCancelSubstreams(stopc <-chan struct{ … } // joinSubstreams waits for all substream goroutines to complete. func (w *watchGrpcStream) joinSubstreams() { … } var maxBackoff … func (w *watchGrpcStream) backoffIfUnavailable(backoff time.Duration, err error) time.Duration { … } // openWatchClient retries opening a watch client until success or halt. // manually retry in case "ws==nil && err==nil" // TODO: remove FailFast=false func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error) { … } // toPB converts an internal watch request structure to its protobuf WatchRequest structure. func (wr *watchRequest) toPB() *pb.WatchRequest { … } // toPB converts an internal progress request structure to its protobuf WatchRequest structure. func (pr *progressRequest) toPB() *pb.WatchRequest { … } func streamKeyFromCtx(ctx context.Context) string { … }
Codebase Browser
kubernetes