update for latest micro

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
2022-01-12 17:40:58 +03:00
parent f247a8c906
commit a10275ad1f
3 changed files with 56 additions and 21 deletions

View File

@@ -42,6 +42,14 @@ func (g *grpcStream) Send(msg interface{}) error {
return nil
}
func (g *grpcStream) SendMsg(msg interface{}) error {
if err := g.ClientStream.SendMsg(msg); err != nil {
g.setError(err)
return err
}
return nil
}
func (g *grpcStream) Recv(msg interface{}) (err error) {
defer g.setError(err)
@@ -60,6 +68,24 @@ func (g *grpcStream) Recv(msg interface{}) (err error) {
return
}
func (g *grpcStream) RecvMsg(msg interface{}) (err error) {
defer g.setError(err)
if err = g.ClientStream.RecvMsg(msg); err != nil {
// #202 - inconsistent gRPC stream behavior
// the only way to tell if the stream is done is when we get a EOF on the Recv
// here we should close the underlying gRPC ClientConn
closeErr := g.Close()
if err == io.EOF && closeErr != nil {
err = closeErr
}
return err
}
return
}
func (g *grpcStream) Error() error {
g.RLock()
defer g.RUnlock()
@@ -90,3 +116,17 @@ func (g *grpcStream) Close() error {
g.close(g.err)
return g.ClientStream.CloseSend()
}
func (g *grpcStream) CloseSend() error {
g.Lock()
defer g.Unlock()
if g.closed {
return nil
}
// close the connection
g.closed = true
g.close(g.err)
return g.ClientStream.CloseSend()
}