diff --git a/protocol/grpc/client.go b/protocol/grpc/client.go index 0c7499a179571d623eccc607dd4cc8f1950f3239..97cd2216db64260c945e27204e7321085a84cefd 100644 --- a/protocol/grpc/client.go +++ b/protocol/grpc/client.go @@ -25,15 +25,64 @@ import ( "github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc" "github.com/opentracing/opentracing-go" "google.golang.org/grpc" + "gopkg.in/yaml.v2" ) import ( "github.com/apache/dubbo-go/common" "github.com/apache/dubbo-go/common/constant" + "github.com/apache/dubbo-go/common/logger" "github.com/apache/dubbo-go/config" ) -// Client ... +var ( + clientConf *ClientConfig +) + +func init() { + // load clientconfig from consumer_config + consumerConfig := config.GetConsumerConfig() + + clientConfig := GetClientConfig() + clientConf = &clientConfig + + // check client config and decide whether to use the default config + defer func() { + if clientConf == nil || len(clientConf.ContentSubType) == 0 { + defaultClientConfig := GetDefaultClientConfig() + clientConf = &defaultClientConfig + } + if err := clientConf.Validate(); err != nil { + panic(err) + } + }() + + if consumerConfig.ApplicationConfig == nil { + return + } + protocolConf := config.GetConsumerConfig().ProtocolConf + + if protocolConf == nil { + logger.Info("protocol_conf default use dubbo config") + } else { + grpcConf := protocolConf.(map[interface{}]interface{})[GRPC] + if grpcConf == nil { + logger.Warnf("grpcConf is nil") + return + } + grpcConfByte, err := yaml.Marshal(grpcConf) + if err != nil { + panic(err) + } + err = yaml.Unmarshal(grpcConfByte, clientConf) + if err != nil { + panic(err) + } + } + +} + +// Client return grpc connection and warp service stub type Client struct { *grpc.ClientConn invoker reflect.Value @@ -43,9 +92,11 @@ type Client struct { func NewClient(url common.URL) *Client { // if global trace instance was set , it means trace function enabled. If not , will return Nooptracer tracer := opentracing.GlobalTracer() - conn, err := grpc.Dial(url.Location, grpc.WithInsecure(), grpc.WithBlock(), - grpc.WithUnaryInterceptor( - otgrpc.OpenTracingClientInterceptor(tracer, otgrpc.LogPayloads()))) + dailOpts := make([]grpc.DialOption, 0, 4) + dailOpts = append(dailOpts, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithUnaryInterceptor( + otgrpc.OpenTracingClientInterceptor(tracer, otgrpc.LogPayloads())), + grpc.WithDefaultCallOptions(grpc.CallContentSubtype(clientConf.ContentSubType))) + conn, err := grpc.Dial(url.Location, dailOpts...) if err != nil { panic(err) } diff --git a/protocol/grpc/codec.go b/protocol/grpc/codec.go new file mode 100644 index 0000000000000000000000000000000000000000..7235a3941baefcdce5964ae21dbdbfe9d6ca9cc8 --- /dev/null +++ b/protocol/grpc/codec.go @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package grpc + +import ( + "bytes" + "encoding/json" +) + +import ( + "github.com/golang/protobuf/jsonpb" + "github.com/golang/protobuf/proto" + "google.golang.org/grpc/encoding" +) + +const ( + codecJson = "json" + codecProto = "proto" +) + +func init() { + encoding.RegisterCodec(grpcJson{ + Marshaler: jsonpb.Marshaler{ + EmitDefaults: true, + OrigName: true, + }, + }) +} + +type grpcJson struct { + jsonpb.Marshaler + jsonpb.Unmarshaler +} + +// Name implements grpc encoding package Codec interface method, +// returns the name of the Codec implementation. +func (_ grpcJson) Name() string { + return codecJson +} + +// Marshal implements grpc encoding package Codec interface method,returns the wire format of v. +func (j grpcJson) Marshal(v interface{}) (out []byte, err error) { + if pm, ok := v.(proto.Message); ok { + b := new(bytes.Buffer) + err := j.Marshaler.Marshal(b, pm) + if err != nil { + return nil, err + } + return b.Bytes(), nil + } + return json.Marshal(v) +} + +// Unmarshal implements grpc encoding package Codec interface method,Unmarshal parses the wire format into v. +func (j grpcJson) Unmarshal(data []byte, v interface{}) (err error) { + if pm, ok := v.(proto.Message); ok { + b := bytes.NewBuffer(data) + return j.Unmarshaler.Unmarshal(b, pm) + } + return json.Unmarshal(data, v) +} diff --git a/protocol/grpc/config.go b/protocol/grpc/config.go new file mode 100644 index 0000000000000000000000000000000000000000..e8a9baac8999a009bab435a5bd4e70d102f77b56 --- /dev/null +++ b/protocol/grpc/config.go @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package grpc + +import ( + perrors "github.com/pkg/errors" +) + +type ( + // ServerConfig currently is empty struct,for future expansion + ServerConfig struct { + } + + // ClientConfig wrap client call parameters + ClientConfig struct { + // content type, more information refer by https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests + ContentSubType string `default:"proto" yaml:"content_sub_type" json:"content_sub_type,omitempty"` + } +) + +// GetDefaultClientConfig return grpc client default call options +func GetDefaultClientConfig() ClientConfig { + return ClientConfig{ + ContentSubType: codecProto, + } +} + +// GetDefaultServerConfig currently return empty struct,for future expansion +func GetDefaultServerConfig() ServerConfig { + return ServerConfig{} +} + +// GetClientConfig return grpc client custom call options +func GetClientConfig() ClientConfig { + return ClientConfig{} +} + +// Validate check if custom config encoding is supported in dubbo grpc +func (c *ClientConfig) Validate() error { + if c.ContentSubType != codecJson && c.ContentSubType != codecProto { + return perrors.Errorf(" dubbo-go grpc codec currently only support proto銆乯son, %s isn't supported,"+ + " please check protocol content_sub_type config", c.ContentSubType) + } + return nil +} + +// Validate currently return empty struct,for future expansion +func (c *ServerConfig) Validate() error { + return nil +} diff --git a/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go b/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go index 83d28519f6f8e28bf471ce2ea6807603c1324911..1af4fafdc606783e937ede63f99e5a08f0b2419e 100644 --- a/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go +++ b/protocol/grpc/protoc-gen-dubbo/plugin/dubbo/dubbo.go @@ -107,7 +107,6 @@ func (g *dubboGrpc) GenerateImports(file *generator.FileDescriptor) { g.P(`dgrpc "github.com/apache/dubbo-go/protocol/grpc"`) g.P(`"github.com/apache/dubbo-go/protocol/invocation"`) g.P(`"github.com/apache/dubbo-go/protocol"`) - g.P(`"github.com/apache/dubbo-go/config"`) g.P(` ) `) } @@ -266,7 +265,7 @@ func (g *dubboGrpc) generateServerMethod(servName, fullServName string, method * g.P(`invo := invocation.NewRPCInvocation("`, methName, `", args, nil)`) g.P("if interceptor == nil {") - g.P("result := base.GetProxyImpl().Invoke(invo)") + g.P("result := base.GetProxyImpl().Invoke(context.Background(), invo)") g.P("return result.Result(), result.Error()") g.P("}") @@ -276,7 +275,7 @@ func (g *dubboGrpc) generateServerMethod(servName, fullServName string, method * g.P("}") g.P("handler := func(ctx ", contextPkg, ".Context, req interface{}) (interface{}, error) {") - g.P("result := base.GetProxyImpl().Invoke(invo)") + g.P("result := base.GetProxyImpl().Invoke(context.Background(), invo)") g.P("return result.Result(), result.Error()") g.P("}")