Merge pull request #20541 from apullo777/watcher-respch

cache: change watcher.eventQueue to respCh (clientv3.WatchResponse)

Author: serathius

cache/cache.go

@@ -119,7 +119,7 @@ func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption
 				return
 			case <-c.internalCtx.Done():
 				return
-			case events, ok := <-w.eventQueue:
+			case resp, ok := <-w.respCh:
 				if !ok {
 					if w.cancelResp != nil {
 						select {
@@ -135,7 +135,7 @@ func (c *Cache) Watch(ctx context.Context, key string, opts ...clientv3.OpOption
 					return
 				case <-c.internalCtx.Done():
 					return
-				case responseChan <- clientv3.WatchResponse{Events: events}:
+				case responseChan <- resp:
 				}
 			}
 		}
@@ -272,11 +272,17 @@ func (c *Cache) applyStorage(storeW *watcher) error {
 		select {
 		case <-c.internalCtx.Done():
 			return nil
-		case events, ok := <-storeW.eventQueue:
+		case resp, ok := <-storeW.respCh:
 			if !ok {
 				return nil
 			}
-			if err := c.store.Apply(events); err != nil {
+			if resp.Canceled {
+				return nil
+			}
+			if len(resp.Events) == 0 {
+				continue
+			}
+			if err := c.store.Apply(resp.Events); err != nil {
 				return err
 			}
 		}

cache/demux_test.go

@@ -154,9 +154,15 @@ func readBatches(t *testing.T, w *watcher, n int) (revs []int64, sizes []int) {
 	timeout := time.After(2 * time.Second)
 	for len(revs) < n {
 		select {
-		case batch := <-w.eventQueue:
-			revs = append(revs, batch[0].Kv.ModRevision)
-			sizes = append(sizes, len(batch))
+		case resp := <-w.respCh:
+			if resp.Canceled {
+				t.Fatalf("unexpected canceled response in test: %v", resp.CancelReason)
+			}
+			if len(resp.Events) == 0 {
+				continue
+			}
+			revs = append(revs, resp.Events[0].Kv.ModRevision)
+			sizes = append(sizes, len(resp.Events))
 		case <-timeout:
 			t.Fatalf("timed out waiting for %d batches; got %d", n, len(revs))
 		}

cache/watcher.go

@@ -23,16 +23,16 @@ import (
 
 // watcher holds one client’s buffered stream of events.
 type watcher struct {
-	eventQueue chan []*clientv3.Event
+	respCh     chan clientv3.WatchResponse
 	cancelResp *clientv3.WatchResponse
 	keyPred    KeyPredicate
 	stopOnce   sync.Once
 }
 
 func newWatcher(bufSize int, pred KeyPredicate) *watcher {
 	return &watcher{
-		eventQueue: make(chan []*clientv3.Event, bufSize),
-		keyPred:    pred,
+		respCh:  make(chan clientv3.WatchResponse, bufSize),
+		keyPred: pred,
 	}
 }
 
@@ -52,7 +52,7 @@ func (w *watcher) enqueueEvent(eventBatch []*clientv3.Event) bool {
 		eventBatch = filtered
 	}
 	select {
-	case w.eventQueue <- eventBatch:
+	case w.respCh <- clientv3.WatchResponse{Events: eventBatch}:
 		return true
 	default:
 		return false
@@ -67,13 +67,13 @@ func (w *watcher) Compact(compactRev int64) {
 	}
 	w.stopOnce.Do(func() {
 		w.cancelResp = resp
-		close(w.eventQueue)
+		close(w.respCh)
 	})
 }
 
 // Stop closes the event channel atomically.
 func (w *watcher) Stop() {
 	w.stopOnce.Do(func() {
-		close(w.eventQueue)
+		close(w.respCh)
 	})
 }