kubernetes/vendor/go.etcd.io/etcd/server/v3/etcdserver/api/v3rpc/watch.go

const minWatchProgressInterval …

type watchServer …

// NewWatchServer returns a new watch server.
func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer { … }

var progressReportInterval …

var progressReportIntervalMu …

// GetProgressReportInterval returns the current progress report interval (for testing).
func GetProgressReportInterval() time.Duration { … }

// SetProgressReportInterval updates the current progress report interval (for testing).
func SetProgressReportInterval(newTimeout time.Duration) { … }

const ctrlStreamBufLen …

type serverWatchStream …

func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) { … }

func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) error { … }

func (sws *serverWatchStream) recvLoop() error { … }

func (sws *serverWatchStream) sendLoop() { … }

func IsCreateEvent(e mvccpb.Event) bool { … }

func sendFragments(
	wr *pb.WatchResponse,
	maxRequestBytes int,
	sendFunc func(*pb.WatchResponse) error) error { … }

func (sws *serverWatchStream) close() { … }

func (sws *serverWatchStream) newResponseHeader(rev int64) *pb.ResponseHeader { … }

func filterNoDelete(e mvccpb.Event) bool { … }

func filterNoPut(e mvccpb.Event) bool { … }

// FiltersFromRequest returns "mvcc.FilterFunc" from a given watch create request.
func FiltersFromRequest(creq *pb.WatchCreateRequest) []mvcc.FilterFunc { … }