clientv3: non-recursive watch

Signed-off-by: Ted Yu <yuzhihong@gmail.com>

clientv3/watch.go

@@ -311,6 +311,8 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 	ok := false
 	ctxKey := streamKeyFromCtx(ctx)
 
+	var closeCh chan WatchResponse
+	for {
 		// find or allocate appropriate grpc watch stream
 		w.mu.Lock()
 		if w.streams == nil {
@@ -330,20 +332,24 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 		w.mu.Unlock()
 
 		// couldn't create channel; return closed channel
-	closeCh := make(chan WatchResponse, 1)
+		if closeCh == nil {
+			closeCh = make(chan WatchResponse, 1)
+		}
 
 		// submit request
 		select {
 		case reqc <- wr:
 			ok = true
 		case <-wr.ctx.Done():
+			ok = false
 		case <-donec:
+			ok = false
 			if wgs.closeErr != nil {
 				closeCh <- WatchResponse{Canceled: true, closeErr: wgs.closeErr}
 				break
 			}
 			// retry; may have dropped stream from no ctxs
-		return w.Watch(ctx, key, opts...)
+			continue
 		}
 
 		// receive channel
@@ -358,9 +364,11 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 					break
 				}
 				// retry; may have dropped stream from no ctxs
-			return w.Watch(ctx, key, opts...)
+				continue
 			}
 		}
+		break
+	}
 
 	close(closeCh)
 	return closeCh