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

update hooks calling and fix errors create #137

Merged
vtolstov merged 2 commits from devstigneev/micro-client-grpc:v3 into v3 2024-12-19 23:49:50 +03:00
Conversation 0 Commits 2 Files Changed 3 +12 -12
1 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit bec0e310e9 - Show all commits

24
grpc.go
View File

@ -103,7 +103,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
cf, err := g.newCodec(req.ContentType()) cf, err := g.newCodec(req.ContentType())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
maxRecvMsgSize := g.maxRecvMsgSizeValue() maxRecvMsgSize := g.maxRecvMsgSizeValue()
@ -147,7 +147,7 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...) cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", "Error sending request: %v", err)
} }
defer func() { defer func() {
// defer execution of release // defer execution of release
@ -217,7 +217,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
cf, err := g.newCodec(req.ContentType()) cf, err := g.newCodec(req.ContentType())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
var dialCtx context.Context var dialCtx context.Context
@ -258,7 +258,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...) cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", "Error sending request: %v", err)
} }
desc := &grpc.StreamDesc{ desc := &grpc.StreamDesc{
@ -291,7 +291,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// release the connection // release the connection
g.pool.Put(cc, err) g.pool.Put(cc, err)
// now return the error // now return the error
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err)) return errors.InternalServerError("go.micro.client", "Error creating stream: %v", err)
} }
// set request codec // set request codec
@ -421,7 +421,7 @@ func (g *grpcClient) Init(opts ...client.Option) error {
g.funcPublish = g.fnPublish g.funcPublish = g.fnPublish
g.funcBatchPublish = g.fnBatchPublish g.funcBatchPublish = g.fnBatchPublish
g.opts.Hooks.EachNext(func(hook options.Hook) { g.opts.Hooks.EachPrev(func(hook options.Hook) {
switch h := hook.(type) { switch h := hook.(type) {
case client.HookCall: case client.HookCall:
g.funcCall = h(g.funcCall) g.funcCall = h(g.funcCall)
@ -533,7 +533,7 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// 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)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
// only sleep if greater than 0 // only sleep if greater than 0
@ -548,7 +548,7 @@ func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interfa
// TODO apply any filtering here // TODO apply any filtering here
routes, err = g.opts.Lookup(ctx, req, callOpts) routes, err = g.opts.Lookup(ctx, req, callOpts)
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
// balance the list of nodes // balance the list of nodes
@ -675,7 +675,7 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
// 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)
if err != nil { if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error()) return nil, errors.InternalServerError("go.micro.client", "%+v", err)
} }
// only sleep if greater than 0 // only sleep if greater than 0
@ -690,7 +690,7 @@ func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...c
// TODO apply any filtering here // TODO apply any filtering here
routes, err = g.opts.Lookup(ctx, req, callOpts) routes, err = g.opts.Lookup(ctx, req, callOpts)
if err != nil { if err != nil {
return nil, errors.InternalServerError("go.micro.client", err.Error()) return nil, errors.InternalServerError("go.micro.client", "%+v", err)
} }
// balance the list of nodes // balance the list of nodes
@ -822,12 +822,12 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
// use codec for payload // use codec for payload
cf, err := g.newCodec(p.ContentType()) cf, err := g.newCodec(p.ContentType())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
// set the body // set the body
b, err := cf.Marshal(p.Payload()) b, err := cf.Marshal(p.Payload())
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", "%+v", err)
} }
body = b body = b
} }