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

Merge pull request #6331 from xiang90/fix_proxy

Browse Source 
grpcproxy: fix stream closing issue
This commit is contained in:
Xiang Li 2016-09-01 11:27:37 -07:00 committed by GitHub
commit 26999db927
1 changed files with 15 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
proxy/grpcproxy/watch_client_adapter.go
View File

@ -32,7 +32,11 @@ func (s *ws2wc) Watch(ctx context.Context, opts ...grpc.CallOption) (pb.Watch_Wa
headerc, trailerc := make(chan metadata.MD, 1), make(chan metadata.MD, 1)
wclient := &ws2wcClientStream{chanClientStream{headerc, trailerc, &chanStream{ch1, ch2, ctx}}}
wserver := &ws2wcServerStream{chanServerStream{headerc, trailerc, &chanStream{ch2, ch1, ctx}}}
go s.wserv.Watch(wserver)
go func() {
s.wserv.Watch(wserver)
// close the server side sender
close(ch1)
}()
return wclient, nil
}
@ -88,7 +92,7 @@ func (ss *chanServerStream) SetTrailer(md metadata.MD) {
type chanClientStream struct {
headerc <-chan metadata.MD
trailerc <-chan metadata.MD
grpc.Stream
*chanStream
}
func (cs *chanClientStream) Header() (metadata.MD, error) {
@ -109,7 +113,10 @@ func (cs *chanClientStream) Trailer() metadata.MD {
}
}
func (s *chanClientStream) CloseSend() error { return nil }
func (s *chanClientStream) CloseSend() error {
close(s.chanStream.sendc)
return nil
}
// chanStream implements grpc.Stream using channels
type chanStream struct {
@ -132,8 +139,11 @@ func (s *chanStream) SendMsg(m interface{}) error {
func (s *chanStream) RecvMsg(m interface{}) error {
v := m.(*interface{})
select {
case m = <-s.recvc:
*v = m
case msg, ok := <-s.recvc:
if !ok {
return grpc.ErrClientConnClosing
}
*v = msg
return nil
case <-s.ctx.Done():
}