remove reflect in grpc server (#1200)

protocol/grpc/client.go

@@ -88,7 +88,8 @@ type Client struct {
 
 // NewClient creates a new gRPC client.
 func NewClient(url *common.URL) (*Client, error) {
-	// if global trace instance was set , it means trace function enabled. If not , will return Nooptracer
+	// If global trace instance was set, it means trace function enabled.
+	// If not, will return NoopTracer.
 	tracer := opentracing.GlobalTracer()
 	dialOpts := make([]grpc.DialOption, 0, 4)
 	maxMessageSize, _ := strconv.Atoi(url.GetParam(constant.MESSAGE_SIZE_KEY, "4"))
@@ -96,16 +97,22 @@ func NewClient(url *common.URL) (*Client, error) {
 	// consumer config client connectTimeout
 	connectTimeout := config.GetConsumerConfig().ConnectTimeout
 
-	dialOpts = append(dialOpts, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithTimeout(connectTimeout), grpc.WithUnaryInterceptor(
-		otgrpc.OpenTracingClientInterceptor(tracer, otgrpc.LogPayloads())),
+	dialOpts = append(dialOpts,
+		grpc.WithInsecure(),
+		grpc.WithBlock(),
+		grpc.WithTimeout(connectTimeout),
+		grpc.WithUnaryInterceptor(otgrpc.OpenTracingClientInterceptor(tracer, otgrpc.LogPayloads())),
+		grpc.WithStreamInterceptor(otgrpc.OpenTracingStreamClientInterceptor(tracer, otgrpc.LogPayloads())),
 		grpc.WithDefaultCallOptions(
 			grpc.CallContentSubtype(clientConf.ContentSubType),
 			grpc.MaxCallRecvMsgSize(1024*1024*maxMessageSize),
-			grpc.MaxCallSendMsgSize(1024*1024*maxMessageSize)))
+			grpc.MaxCallSendMsgSize(1024*1024*maxMessageSize),
+		),
+	)
 
 	conn, err := grpc.Dial(url.Location, dialOpts...)
 	if err != nil {
-		logger.Errorf("grpc dail error: %v", err)
+		logger.Errorf("grpc dial error: %v", err)
 		return nil, err
 	}
 

protocol/grpc/grpc_protocol.go

@@ -113,7 +113,7 @@ func (gp *GrpcProtocol) Destroy() {
 	}
 }
 
-// GetProtocol gets gRPC protocol , will create if null.
+// GetProtocol gets gRPC protocol, will create if null.
 func GetProtocol() protocol.Protocol {
 	if grpcProtocol == nil {
 		grpcProtocol = NewGRPCProtocol()

protocol/grpc/protoc-gen-dubbo/examples/Makefile

@@ -16,4 +16,4 @@
 grpc-gen:
 	protoc -I ./ helloworld.proto --go_out=plugins=grpc:.
 dubbo-gen:
-	protoc -I ./ helloworld.proto  --dubbo_out=plugins=grpc+dubbo:.
+	protoc -I ./ helloworld.proto --dubbo_out=plugins=grpc+dubbo:.

protocol/grpc/protoc-gen-dubbo/examples/helloworld.pb.go

@@ -21,14 +21,13 @@
 package main
 
 import (
-	"context"
-	"fmt"
-	"math"
-
-	"github.com/golang/protobuf/proto"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
+	context "context"
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	math "math"
 )
 
 import (
@@ -38,11 +37,9 @@ import (
 )
 
 // Reference imports to suppress errors if they are not otherwise used.
-var (
-	_ = proto.Marshal
-	_ = fmt.Errorf
-	_ = math.Inf
-)
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the proto package it is being compiled against.
@@ -68,19 +65,15 @@ func (*HelloRequest) Descriptor() ([]byte, []int) {
 func (m *HelloRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HelloRequest.Unmarshal(m, b)
 }
-
 func (m *HelloRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 	return xxx_messageInfo_HelloRequest.Marshal(b, m, deterministic)
 }
-
 func (m *HelloRequest) XXX_Merge(src proto.Message) {
 	xxx_messageInfo_HelloRequest.Merge(m, src)
 }
-
 func (m *HelloRequest) XXX_Size() int {
 	return xxx_messageInfo_HelloRequest.Size(m)
 }
-
 func (m *HelloRequest) XXX_DiscardUnknown() {
 	xxx_messageInfo_HelloRequest.DiscardUnknown(m)
 }
@@ -112,19 +105,15 @@ func (*HelloReply) Descriptor() ([]byte, []int) {
 func (m *HelloReply) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_HelloReply.Unmarshal(m, b)
 }
-
 func (m *HelloReply) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 	return xxx_messageInfo_HelloReply.Marshal(b, m, deterministic)
 }
-
 func (m *HelloReply) XXX_Merge(src proto.Message) {
 	xxx_messageInfo_HelloReply.Merge(m, src)
 }
-
 func (m *HelloReply) XXX_Size() int {
 	return xxx_messageInfo_HelloReply.Size(m)
 }
-
 func (m *HelloReply) XXX_DiscardUnknown() {
 	xxx_messageInfo_HelloReply.DiscardUnknown(m)
 }
@@ -162,14 +151,12 @@ var fileDescriptor_17b8c58d586b62f2 = []byte{
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
-var (
-	_ context.Context
-	_ grpc.ClientConn
-)
+var _ context.Context
+var _ grpc.ClientConnInterface
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the grpc package it is being compiled against.
-const _ = grpc.SupportPackageIsVersion4
+const _ = grpc.SupportPackageIsVersion6
 
 // GreeterClient is the client API for Greeter service.
 //
@@ -180,10 +167,10 @@ type GreeterClient interface {
 }
 
 type greeterClient struct {
-	cc *grpc.ClientConn
+	cc grpc.ClientConnInterface
 }
 
-func NewGreeterClient(cc *grpc.ClientConn) GreeterClient {
+func NewGreeterClient(cc grpc.ClientConnInterface) GreeterClient {
 	return &greeterClient{cc}
 }
 
@@ -203,7 +190,8 @@ type GreeterServer interface {
 }
 
 // UnimplementedGreeterServer can be embedded to have forward compatible implementations.
-type UnimplementedGreeterServer struct{}
+type UnimplementedGreeterServer struct {
+}
 
 func (*UnimplementedGreeterServer) SayHello(ctx context.Context, req *HelloRequest) (*HelloReply, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method SayHello not implemented")
@@ -282,7 +270,7 @@ func _DUBBO_Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec f
 	args = append(args, in)
 	invo := invocation.NewRPCInvocation("SayHello", args, nil)
 	if interceptor == nil {
-		result := base.GetProxyImpl().Invoke(context.Background(), invo)
+		result := base.GetProxyImpl().Invoke(ctx, invo)
 		return result.Result(), result.Error()
 	}
 	info := &grpc.UnaryServerInfo{
@@ -290,7 +278,7 @@ func _DUBBO_Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec f
 		FullMethod: "/main.Greeter/SayHello",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		result := base.GetProxyImpl().Invoke(context.Background(), invo)
+		result := base.GetProxyImpl().Invoke(ctx, invo)
 		return result.Result(), result.Error()
 	}
 	return interceptor(ctx, in, info, handler)

protocol/grpc/protoc-gen-dubbo/examples/helloworld.proto

 /*
-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.
-*/
+ * 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.
+ */
 syntax = "proto3";
 
 option java_multiple_files = true;

protocol/grpc/server.go

@@ -20,7 +20,6 @@ package grpc
 import (
 	"fmt"
 	"net"
-	"reflect"
 	"sync"
 	"time"
 )
@@ -39,6 +38,16 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/protocol"
 )
 
+// DubboGrpcService is gRPC service
+type DubboGrpcService interface {
+	// SetProxyImpl sets proxy.
+	SetProxyImpl(impl protocol.Invoker)
+	// GetProxyImpl gets proxy.
+	GetProxyImpl() protocol.Invoker
+	// ServiceDesc gets an RPC service's specification.
+	ServiceDesc() *grpc.ServiceDesc
+}
+
 // Server is a gRPC server
 type Server struct {
 	grpcServer *grpc.Server
@@ -50,16 +59,6 @@ func NewServer() *Server {
 	return &Server{}
 }
 
-// DubboGrpcService is gRPC service
-type DubboGrpcService interface {
-	// SetProxyImpl sets proxy.
-	SetProxyImpl(impl protocol.Invoker)
-	// GetProxyImpl gets proxy.
-	GetProxyImpl() protocol.Invoker
-	// ServiceDesc gets an RPC service's specification.
-	ServiceDesc() *grpc.ServiceDesc
-}
-
 func (s *Server) SetBufferSize(n int) {
 	s.bufferSize = n
 }
@@ -76,12 +75,15 @@ func (s *Server) Start(url *common.URL) {
 		panic(err)
 	}
 
-	// if global trace instance was set, then server tracer instance can be get. If not , will return Nooptracer
+	// If global trace instance was set, then server tracer instance
+	// can be get. If not, will return NoopTracer.
 	tracer := opentracing.GlobalTracer()
 	server := grpc.NewServer(
 		grpc.UnaryInterceptor(otgrpc.OpenTracingServerInterceptor(tracer)),
+		grpc.StreamInterceptor(otgrpc.OpenTracingStreamServerInterceptor(tracer)),
 		grpc.MaxRecvMsgSize(1024*1024*s.bufferSize),
-		grpc.MaxSendMsgSize(1024*1024*s.bufferSize))
+		grpc.MaxSendMsgSize(1024*1024*s.bufferSize),
+	)
 	s.grpcServer = server
 
 	go func() {
@@ -136,18 +138,12 @@ func waitGrpcExporter(providerServices map[string]*config.ServiceConfig) {
 func registerService(providerServices map[string]*config.ServiceConfig, server *grpc.Server) {
 	for key, providerService := range providerServices {
 		service := config.GetProviderService(key)
-
 		ds, ok := service.(DubboGrpcService)
 		if !ok {
 			panic("illegal service type registered")
 		}
 
-		m, ok := reflect.TypeOf(service).MethodByName("SetProxyImpl")
-		if !ok {
-			panic("method SetProxyImpl is necessary for grpc service")
-		}
 		serviceKey := common.ServiceKey(providerService.InterfaceName, providerService.Group, providerService.Version)
-
 		exporter, _ := grpcProtocol.ExporterMap().Load(serviceKey)
 		if exporter == nil {
 			panic(fmt.Sprintf("no exporter found for servicekey: %v", serviceKey))
@@ -156,12 +152,9 @@ func registerService(providerServices map[string]*config.ServiceConfig, server *
 		if invoker == nil {
 			panic(fmt.Sprintf("no invoker found for servicekey: %v", serviceKey))
 		}
-		in := []reflect.Value{reflect.ValueOf(service)}
-		in = append(in, reflect.ValueOf(invoker))
-		m.Func.Call(in)
 
+		ds.SetProxyImpl(invoker)
 		server.RegisterService(ds.ServiceDesc(), service)
-
 	}
 }