unistack-org/micro-client-grpc
4
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 35 Wiki Activity

Apply wrappers to gRPC streams (#1675)

Browse Source 
* Add wrappers to grpc streams

* Fix typo
This commit is contained in:
ben-toogood 2020-06-02 17:56:26 +01:00 committed by Vasiliy Tolstov
parent ace96191fe
commit 9683d38d19
1 changed files with 33 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
grpc.go
View File

@ -6,6 +6,7 @@ import (
"crypto/tls" "crypto/tls"
"fmt" "fmt"
"net" "net"
"reflect"
"strings" "strings"
"sync/atomic" "sync/atomic"
"time" "time"
@ -173,7 +174,7 @@ func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.R
return grr return grr
} }
func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, opts client.CallOptions) (client.Stream, error) { func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
var header map[string]string var header map[string]string
address := node.Address address := node.Address
@ -199,7 +200,7 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
cf, err := g.newGRPCCodec(req.ContentType()) cf, err := g.newGRPCCodec(req.ContentType())
if err != nil { if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
} }
var dialCtx context.Context var dialCtx context.Context
@ -224,7 +225,7 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
cc, err := grpc.DialContext(dialCtx, address, grpcDialOptions...) cc, err := grpc.DialContext(dialCtx, address, grpcDialOptions...)
if err != nil { if err != nil {
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
desc := &grpc.StreamDesc{ desc := &grpc.StreamDesc{
@ -252,7 +253,7 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
// close the connection // close the connection
cc.Close() cc.Close()
// now return the error // now return the error
return nil, errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
} }
codec := &grpcCodec{ codec := &grpcCodec{
@ -265,21 +266,25 @@ func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client
r.codec = codec r.codec = codec
} }
rsp := &response{ // setup the stream response
stream := &grpcStream{
context: ctx,
request: req,
response: &response{
conn: cc, conn: cc,
stream: st, stream: st,
codec: cf, codec: cf,
gcodec: codec, gcodec: codec,
} },
return &grpcStream{
context: ctx,
request: req,
response: rsp,
stream: st, stream: st,
conn: cc, conn: cc,
cancel: cancel, cancel: cancel,
}, nil }
// set the stream as the response
val := reflect.ValueOf(rsp).Elem()
val.Set(reflect.ValueOf(stream).Elem())
return nil
} }
func (g *grpcClient) poolMaxStreams() int { func (g *grpcClient) poolMaxStreams() int {
@ -506,6 +511,14 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
default: default:
} }
// make a copy of stream
gstream := g.stream
// wrap the call in reverse
for i := len(callOpts.CallWrappers); i > 0; i-- {
gstream = callOpts.CallWrappers[i-1](gstream)
}
call := func(i int) (client.Stream, error) { call := func(i int) (client.Stream, error) {
// call backoff first. Someone may want an initial start delay // call backoff first. Someone may want an initial start delay
t, err := callOpts.Backoff(ctx, req, i) t, err := callOpts.Backoff(ctx, req, i)
@ -527,7 +540,10 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error()) return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
} }
stream, err := g.stream(ctx, node, req, callOpts) // make the call
stream := &grpcStream{}
err = g.stream(ctx, node, req, stream, callOpts)
g.opts.Selector.Mark(service, node, err) g.opts.Selector.Mark(service, node, err)
return stream, err return stream, err
} }