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

allow to send server metadata via header #134

Merged
vtolstov merged 1 commits from metadata into v3 2022-11-14 14:54:48 +03:00
Conversation 0 Commits 1 Files Changed 1 +16 -4
1 changed files with 16 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
grpc.go
View File

@ -283,7 +283,7 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) (err err
// get peer from context // get peer from context
if p, ok := peer.FromContext(stream.Context()); ok { if p, ok := peer.FromContext(stream.Context()); ok {
md["Remote"] = p.Addr.String() md.Set("Remote", p.Addr.String())
ctx = peer.NewContext(ctx, p) ctx = peer.NewContext(ctx, p)
} }
@ -388,7 +388,7 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
} }
// define the handler func // define the handler func
fn := func(ctx context.Context, req server.Request, rsp interface{}) (err error) { fn := func(ctx context.Context, req server.Request, rsp interface{}) (err error) {
returnValues = function.Call([]reflect.Value{service.rcvr, mtype.prepareContext(ctx), reflect.ValueOf(argv.Interface()), reflect.ValueOf(rsp)}) returnValues = function.Call([]reflect.Value{service.rcvr, mtype.prepareContext(ctx), argv, reflect.ValueOf(rsp)})
// The return value for the method is an error. // The return value for the method is an error.
if rerr := returnValues[0].Interface(); rerr != nil { if rerr := returnValues[0].Interface(); rerr != nil {
@ -406,7 +406,13 @@ func (g *grpcServer) processRequest(ctx context.Context, stream grpc.ServerStrea
statusCode := codes.OK statusCode := codes.OK
statusDesc := "" statusDesc := ""
// execute the handler // execute the handler
if appErr := fn(ctx, r, replyv.Interface()); appErr != nil { appErr := fn(ctx, r, replyv.Interface())
if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
if err = stream.SendHeader(gmetadata.New(outmd)); err != nil {
return err
}
}
if appErr != nil {
var errStatus *status.Status var errStatus *status.Status
switch verr := appErr.(type) { switch verr := appErr.(type) {
case *errors.Error: case *errors.Error:
@ -526,7 +532,13 @@ func (g *grpcServer) processStream(ctx context.Context, stream grpc.ServerStream
statusCode := codes.OK statusCode := codes.OK
statusDesc := "" statusDesc := ""
if appErr := fn(ctx, r, ss); appErr != nil { appErr := fn(ctx, r, ss)
if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
if err := stream.SendHeader(gmetadata.New(outmd)); err != nil {
return err
}
}
if appErr != nil {
var err error var err error
var errStatus *status.Status var errStatus *status.Status
switch verr := appErr.(type) { switch verr := appErr.(type) {