proxy/grpcproxy/watcher.go
func (w *watcher) send(wr clientv3.WatchResponse) {
	if wr.IsProgressNotify() && !w.progress {
		return
	}
	if w.nextrev > wr.Header.Revision && len(wr.Events) > 0 {
		return
	}
	if w.nextrev == 0 {
		// current watch; expect updates following this revision
		w.nextrev = wr.Header.Revision + 1
	}

	events := make([]*mvccpb.Event, 0, len(wr.Events))

	var lastRev int64
	for i := range wr.Events {
		ev := (*mvccpb.Event)(wr.Events[i])
		if ev.Kv.ModRevision < w.nextrev {
			continue
		} else {
			// We cannot update w.rev here.
			// txn can have multiple events with the same rev.
			// If w.nextrev updates here, it would skip events in the same txn.
			lastRev = ev.Kv.ModRevision
		}

		filtered := false
		for _, filter := range w.filters {
			if filter(*ev) {
				filtered = true
				break
			}
		}
		if filtered {
			continue
		}

		if !w.prevKV {
			evCopy := *ev
			evCopy.PrevKv = nil
			ev = &evCopy
		}
		events = append(events, ev)
	}

	if lastRev >= w.nextrev {
		w.nextrev = lastRev + 1
	}

	// all events are filtered out?
	if !wr.IsProgressNotify() && !wr.Created && len(events) == 0 && wr.CompactRevision == 0 {
		return
	}

	w.lastHeader = wr.Header
	w.post(&pb.WatchResponse{
		Header:          &wr.Header,
		Created:         wr.Created,
		CompactRevision: wr.CompactRevision,
		Canceled:        wr.Canceled,
		WatchId:         w.id,
		Events:          events,
	})
}
func (w *watcher) send(wr clientv3.WatchResponse) {
	if wr.IsProgressNotify() && !w.progress {
		return
	}
	if w.nextrev > wr.Header.Revision && len(wr.Events) > 0 {
		return
	}
	if w.nextrev == 0 {
		// current watch; expect updates following this revision
		w.nextrev = wr.Header.Revision + 1
	}

	events := make([]*mvccpb.Event, 0, len(wr.Events))

	var lastRev int64
	for i := range wr.Events {
		ev := (*mvccpb.Event)(wr.Events[i])
		if ev.Kv.ModRevision < w.nextrev {
			continue
		} else {
			// We cannot update w.rev here.
			// txn can have multiple events with the same rev.
			// If w.nextrev updates here, it would skip events in the same txn.
			lastRev = ev.Kv.ModRevision
		}

		filtered := false
		for _, filter := range w.filters {
			if filter(*ev) {
				filtered = true
				break
			}
		}
		if filtered {
			continue
		}

		if !w.prevKV {
			evCopy := *ev
			evCopy.PrevKv = nil
			ev = &evCopy
		}
		events = append(events, ev)
	}

	if lastRev >= w.nextrev {
		w.nextrev = lastRev + 1
	}

	// all events are filtered out?
	if !wr.IsProgressNotify() && !wr.Created && len(events) == 0 && wr.CompactRevision == 0 {
		return
	}

	w.lastHeader = wr.Header
	w.post(&pb.WatchResponse{
		Header:          &wr.Header,
		Created:         wr.Created,
		CompactRevision: wr.CompactRevision,
		Canceled:        wr.Canceled,
		WatchId:         w.id,
		Events:          events,
	})
}
func ( w * watcher ) send ( wr clientv3 . WatchResponse ) { if wr . IsProgressNotify ( ) && ! w . progress { return } if w . nextrev > wr . Header . Revision && len ( wr . Events ) > 0 { return } if w . nextrev == 0 { // current watch; expect updates following this revision w . nextrev = wr . Header . Revision + 1 } events := make ( [ ] * mvccpb . Event , 0 , len ( wr . Events ) ) var lastRev int64 for i := range wr . Events { ev := ( * mvccpb . Event ) ( wr . Events [ i ] ) if ev . Kv . ModRevision < w . nextrev { continue } else { // We cannot update w.rev here. // txn can have multiple events with the same rev. // If w.nextrev updates here, it would skip events in the same txn. lastRev = ev . Kv . ModRevision } filtered := false for _ , filter := range w . filters { if filter ( * ev ) { filtered = true break } } if filtered { continue } if ! w . prevKV { evCopy := * ev evCopy . PrevKv = nil ev = & evCopy } events = append ( events , ev ) } if lastRev >= w . nextrev { w . nextrev = lastRev + 1 } // all events are filtered out? if ! wr . IsProgressNotify ( ) && ! wr . Created && len ( events ) == 0 && wr . CompactRevision == 0 { return } w . lastHeader = wr . Header w . post ( & pb . WatchResponse { Header : & wr . Header , Created : wr . Created , CompactRevision : wr . CompactRevision , Canceled : wr . Canceled , WatchId : w . id , Events : events , } ) }
// send filters out repeated events by discarding revisions older
// than the last one sent over the watch channel.
send filters out repeated events by discarding revisions older than the last one sent over the watch channel .