Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

clientv3: log warning in case of error sending request

Browse Source
This commit is contained in:
Ted Yu 2019-12-20 15:08:33 -08:00 committed by jingyih
parent 18dfb9cca3
commit e800c62eca
1 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
clientv3/watch.go
View File

@ -141,6 +141,7 @@ type watcher struct {
// streams holds all the active grpc streams keyed by ctx value. // streams holds all the active grpc streams keyed by ctx value.
streams map[string]*watchGrpcStream streams map[string]*watchGrpcStream
lg *zap.Logger
} }
// watchGrpcStream tracks all watch resources attached to a single grpc stream. // watchGrpcStream tracks all watch resources attached to a single grpc stream.
@ -243,6 +244,7 @@ func NewWatchFromWatchClient(wc pb.WatchClient, c *Client) Watcher {
} }
if c != nil { if c != nil {
w.callOpts = c.callOpts w.callOpts = c.callOpts
w.lg = c.lg
} }
return w return w
} }
@ -545,10 +547,14 @@ func (w *watchGrpcStream) run() {
w.resuming = append(w.resuming, ws) w.resuming = append(w.resuming, ws)
if len(w.resuming) == 1 { if len(w.resuming) == 1 {
// head of resume queue, can register a new watcher // head of resume queue, can register a new watcher
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
case *progressRequest: case *progressRequest:
wc.Send(wreq.toPB()) if err := wc.Send(wreq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
// new events from the watch client // new events from the watch client
@ -572,7 +578,9 @@ func (w *watchGrpcStream) run() {
} }
if ws := w.nextResume(); ws != nil { if ws := w.nextResume(); ws != nil {
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
// reset for next iteration // reset for next iteration
@ -633,7 +641,9 @@ func (w *watchGrpcStream) run() {
return return
} }
if ws := w.nextResume(); ws != nil { if ws := w.nextResume(); ws != nil {
wc.Send(ws.initReq.toPB()) if err := wc.Send(ws.initReq.toPB()); err != nil {
lg.Warningf("error when sending request: %v", err)
}
} }
cancelSet = make(map[int64]struct{}) cancelSet = make(map[int64]struct{})