opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix the streaming rpc case

Browse Source
This commit is contained in:
iamqizhao
2016-07-28 16:45:48 -07:00
parent b2f30e082e
commit 80572b2739
2 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
stream.go
View File

@ -165,7 +165,7 @@ func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
put() put()
put = nil put = nil
} }
if _, ok := err.(transport.ConnectionError); ok { if _, ok := err.(transport.ConnectionError); ok || err == transport.ErrStreamDrain {
if c.failFast { if c.failFast {
cs.finish(err) cs.finish(err)
return nil, toRPCErr(err) return nil, toRPCErr(err)

5
transport/http2_client.go
View File

@ -284,6 +284,10 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
t.mu.Unlock() t.mu.Unlock()
return nil, ErrConnClosing return nil, ErrConnClosing
} }
if t.state == draining {
t.mu.Unlock()
return nil, ErrStreamDrain
}
if t.state != reachable { if t.state != reachable {
t.mu.Unlock() t.mu.Unlock()
return nil, ErrConnClosing return nil, ErrConnClosing
@ -310,7 +314,6 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
t.mu.Lock() t.mu.Lock()
if t.state == draining { if t.state == draining {
t.mu.Unlock() t.mu.Unlock()
// t has been drained. Return the quota.
t.streamsQuota.add(1) t.streamsQuota.add(1)
// Need to make t writable again so that the rpc in flight can still proceed. // Need to make t writable again so that the rpc in flight can still proceed.
t.writableChan <- 0 t.writableChan <- 0