func (w *watchGrpcStream) run() {
	var wc pb.Watch_WatchClient
	var closeErr error
	defer func() {
		w.owner.mu.Lock()
		w.closeErr = closeErr
		if w.owner.streams != nil {
			delete(w.owner.streams, w.ctxKey)
		}
		close(w.donec)
		w.owner.mu.Unlock()
		w.cancel()
	}()
	w.mu.RLock()
	stopc := w.stopc
	w.mu.RUnlock()
	if stopc == nil {
		return
	}
	if wc, closeErr = w.newWatchClient(); closeErr != nil {
		return
	}
	var pendingReq, failedReq *watchRequest
	curReqC := w.reqc
	cancelSet := make(map[int64]struct{})
	for {
		select {
		case pendingReq = <-curReqC:
			curReqC = nil
			if err := wc.Send(pendingReq.toPB()); err == nil {
				break
			}
			failedReq = pendingReq
		case pbresp := <-w.respc:
			switch {
			case pbresp.Created:
				w.addStream(pbresp, pendingReq)
				pendingReq = nil
				curReqC = w.reqc
			case pbresp.Canceled:
				delete(cancelSet, pbresp.WatchId)
				w.mu.Lock()
				if ws, ok := w.streams[pbresp.WatchId]; ok {
					close(ws.recvc)
					delete(w.streams, ws.id)
				}
				numStreams := len(w.streams)
				w.mu.Unlock()
				if numStreams == 0 {
					return
				}
			default:
				if ok := w.dispatchEvent(pbresp); ok {
					break
				}
				if _, ok := cancelSet[pbresp.WatchId]; ok {
					break
				}
				cancelSet[pbresp.WatchId] = struct{}{}
				cr := &pb.WatchRequest_CancelRequest{
					CancelRequest: &pb.WatchCancelRequest{
						WatchId: pbresp.WatchId,
					},
				}
				req := &pb.WatchRequest{RequestUnion: cr}
				wc.Send(req)
			}
		case err := <-w.errc:
			if toErr(w.ctx, err) == v3rpc.ErrNoLeader {
				closeErr = err
				return
			}
			if wc, closeErr = w.newWatchClient(); closeErr != nil {
				return
			}
			curReqC = w.reqc
			if pendingReq != nil {
				failedReq = pendingReq
			}
			cancelSet = make(map[int64]struct{})
		case <-stopc:
			return
		}
		if failedReq != nil {
			go func(wr *watchRequest) {
				select {
				case w.reqc <- wr:
				case <-wr.ctx.Done():
				case <-w.donec:
				}
			}(pendingReq)
			failedReq = nil
			pendingReq = nil
		}
	}
}
run is the root of the goroutines for managing a watcher client
package clientv3
import (
	"fmt"
	"sync"
	"time"
	v3rpc "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	mvccpb "github.com/coreos/etcd/mvcc/mvccpb"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
)
const (
	EventTypeDelete = mvccpb.DELETE
	EventTypePut    = mvccpb.PUT
	closeSendErrTimeout = 250 * time.Millisecond
)
type Event mvccpb.Event
type WatchChan <-chan WatchResponse
type Watcher interface {
	Watch(ctx context.Context, key string, opts ...OpOption) WatchChan
	Close() error
}
type WatchResponse struct {
	Header pb.ResponseHeader
	Events []*Event
	CompactRevision int64
	Canceled bool
	closeErr error
}
func (e *Event) IsCreate() bool {
	return e.Type == EventTypePut && e.Kv.CreateRevision == e.Kv.ModRevision
}
func (e *Event) IsModify() bool {
	return e.Type == EventTypePut && e.Kv.CreateRevision != e.Kv.ModRevision
}
func (wr *WatchResponse) Err() error {
	switch {
	case wr.closeErr != nil:
		return v3rpc.Error(wr.closeErr)
	case wr.CompactRevision != 0:
		return v3rpc.ErrCompacted
	case wr.Canceled:
		return v3rpc.ErrFutureRev
	}
	return nil
}
func (wr *WatchResponse) IsProgressNotify() bool {
	return len(wr.Events) == 0 && !wr.Canceled
}
type watcher struct {
	remote pb.WatchClient
	mu sync.RWMutex
	streams map[string]*watchGrpcStream
}
type watchGrpcStream struct {
	owner  *watcher
	remote pb.WatchClient
	ctx context.Context
	ctxKey string
	cancel context.CancelFunc
	mu sync.RWMutex
	streams map[int64]*watcherStream
	reqc chan *watchRequest
	respc chan *pb.WatchResponse
	stopc chan struct{}
	donec chan struct{}
	errc chan error
	closeErr error
}
type watchRequest struct {
	ctx context.Context
	key string
	end string
	rev int64
	progressNotify bool
	retc chan chan WatchResponse
}
type watcherStream struct {
	initReq watchRequest
	outc chan<- WatchResponse
	recvc chan *WatchResponse
	id    int64
	lastRev int64
	resumec chan int64
}
func NewWatcher(c *Client) Watcher {
	return &watcher{
		remote:  pb.NewWatchClient(c.conn),
		streams: make(map[string]*watchGrpcStream),
	}
}
var valCtxCh = make(chan struct{})
var zeroTime = time.Unix(0, 0)
type valCtx struct{ context.Context }
func (vc *valCtx) Deadline() (time.Time, bool) { return zeroTime, false }
func (vc *valCtx) Done() <-chan struct{}       { return valCtxCh }
func (vc *valCtx) Err() error                  { return nil }
func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream {
	ctx, cancel := context.WithCancel(&valCtx{inctx})
	wgs := &watchGrpcStream{
		owner:   w,
		remote:  w.remote,
		ctx:     ctx,
		ctxKey:  fmt.Sprintf("%v", inctx),
		cancel:  cancel,
		streams: make(map[int64]*watcherStream),
		respc: make(chan *pb.WatchResponse),
		reqc:  make(chan *watchRequest),
		stopc: make(chan struct{}),
		donec: make(chan struct{}),
		errc:  make(chan error, 1),
	}
	go wgs.run()
	return wgs
}
func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {
	ow := opWatch(key, opts...)
	retc := make(chan chan WatchResponse, 1)
	wr := &watchRequest{
		ctx:            ctx,
		key:            string(ow.key),
		end:            string(ow.end),
		rev:            ow.rev,
		progressNotify: ow.progressNotify,
		retc:           retc,
	}
	ok := false
	ctxKey := fmt.Sprintf("%v", ctx)
	w.mu.Lock()
	if w.streams == nil {
		w.mu.Unlock()
		ch := make(chan WatchResponse)
		close(ch)
		return ch
	}
	wgs := w.streams[ctxKey]
	if wgs == nil {
		wgs = w.newWatcherGrpcStream(ctx)
		w.streams[ctxKey] = wgs
	}
	donec := wgs.donec
	reqc := wgs.reqc
	w.mu.Unlock()
	closeCh := make(chan WatchResponse, 1)
	select {
	case reqc <- wr:
		ok = true
	case <-wr.ctx.Done():
		wgs.stopIfEmpty()
	case <-donec:
		if wgs.closeErr != nil {
			closeCh <- WatchResponse{closeErr: wgs.closeErr}
			break
		}
		return w.Watch(ctx, key, opts...)
	}
	if ok {
		select {
		case ret := <-retc:
			return ret
		case <-ctx.Done():
		case <-donec:
			if wgs.closeErr != nil {
				closeCh <- WatchResponse{closeErr: wgs.closeErr}
				break
			}
			return w.Watch(ctx, key, opts...)
		}
	}
	close(closeCh)
	return closeCh
}
func (w *watcher) Close() (err error) {
	w.mu.Lock()
	streams := w.streams
	w.streams = nil
	w.mu.Unlock()
	for _, wgs := range streams {
		if werr := wgs.Close(); werr != nil {
			err = werr
		}
	}
	return err
}
func (w *watchGrpcStream) Close() (err error) {
	w.mu.Lock()
	if w.stopc != nil {
		close(w.stopc)
		w.stopc = nil
	}
	w.mu.Unlock()
	<-w.donec
	select {
	case err = <-w.errc:
	default:
	}
	return toErr(w.ctx, err)
}
func (w *watchGrpcStream) addStream(resp *pb.WatchResponse, pendingReq *watchRequest) {
	if pendingReq == nil {
		return
	}
	if resp.Canceled || resp.CompactRevision != 0 {
		ret := make(chan WatchResponse, 1)
		ret <- WatchResponse{
			Header:          *resp.Header,
			CompactRevision: resp.CompactRevision,
			Canceled:        true}
		close(ret)
		pendingReq.retc <- ret
		return
	}
	ret := make(chan WatchResponse)
	if resp.WatchId == -1 {
		close(ret)
		pendingReq.retc <- ret
		return
	}
	ws := &watcherStream{
		initReq: *pendingReq,
		id:      resp.WatchId,
		outc:    ret,
		recvc:   make(chan *WatchResponse, 4),
		resumec: make(chan int64),
	}
	if pendingReq.rev == 0 {
		ws.initReq.rev = resp.Header.Revision
	}
	w.mu.Lock()
	w.streams[ws.id] = ws
	w.mu.Unlock()
	pendingReq.retc <- ret
	go w.serveStream(ws)
}
func (w *watchGrpcStream) closeStream(ws *watcherStream) {
	w.mu.Lock()
	close(ws.initReq.retc)
	close(ws.outc)
	delete(w.streams, ws.id)
	w.mu.Unlock()
}