Merge pull request #275 from xujianhai666/feature/grpc-client

feat(grpc): 支持client grpc 调用

common/proxy/proxy.go

@@ -72,10 +72,11 @@ func (p *Proxy) Implement(v common.RPCService) {
 	makeDubboCallProxy := func(methodName string, outs []reflect.Type) func(in []reflect.Value) []reflect.Value {
 		return func(in []reflect.Value) []reflect.Value {
 			var (
-				err   error
-				inv   *invocation_impl.RPCInvocation
-				inArr []interface{}
-				reply reflect.Value
+				err    error
+				inv    *invocation_impl.RPCInvocation
+				inIArr []interface{}
+				inVArr []reflect.Value
+				reply  reflect.Value
 			)
 			if methodName == "Echo" {
 				methodName = "$echo"
@@ -104,21 +105,25 @@ func (p *Proxy) Implement(v common.RPCService) {
 			}
 
 			if end-start <= 0 {
-				inArr = []interface{}{}
+				inIArr = []interface{}{}
+				inVArr = []reflect.Value{}
 			} else if v, ok := in[start].Interface().([]interface{}); ok && end-start == 1 {
-				inArr = v
+				inIArr = v
+				inVArr = []reflect.Value{in[start]}
 			} else {
-				inArr = make([]interface{}, end-start)
+				inIArr = make([]interface{}, end-start)
+				inVArr = make([]reflect.Value, end-start)
 				index := 0
 				for i := start; i < end; i++ {
-					inArr[index] = in[i].Interface()
+					inIArr[index] = in[i].Interface()
+					inVArr[index] = in[i]
 					index++
 				}
 			}
 
 			inv = invocation_impl.NewRPCInvocationWithOptions(invocation_impl.WithMethodName(methodName),
-				invocation_impl.WithArguments(inArr), invocation_impl.WithReply(reply.Interface()),
-				invocation_impl.WithCallBack(p.callBack))
+				invocation_impl.WithArguments(inIArr), invocation_impl.WithReply(reply.Interface()),
+				invocation_impl.WithCallBack(p.callBack), invocation_impl.WithParameterValues(inVArr))
 
 			for k, value := range p.attachments {
 				inv.SetAttachments(k, value)

config/config_loader.go

@@ -91,7 +91,7 @@ func Load() {
 				continue
 			}
 			ref.id = key
-			ref.Refer()
+			ref.Refer(rpcService)
 			ref.Implement(rpcService)
 		}
 		//wait for invoker is available, if wait over default 3s, then panic

config/reference_config.go

@@ -88,7 +88,7 @@ func (refconfig *ReferenceConfig) UnmarshalYAML(unmarshal func(interface{}) erro
 	return nil
 }
 
-func (refconfig *ReferenceConfig) Refer() {
+func (refconfig *ReferenceConfig) Refer(impl interface{}) {
 	url := common.NewURLWithOptions(common.WithPath(refconfig.id), common.WithProtocol(refconfig.Protocol), common.WithParams(refconfig.getUrlMap()))
 
 	//1. user specified URL, could be peer-to-peer address, or register center's address.
@@ -122,12 +122,12 @@ func (refconfig *ReferenceConfig) Refer() {
 		}
 	}
 	if len(refconfig.urls) == 1 {
-		refconfig.invoker = extension.GetProtocol(refconfig.urls[0].Protocol).Refer(*refconfig.urls[0])
+		refconfig.invoker = extension.GetProtocol(refconfig.urls[0].Protocol).Refer(*refconfig.urls[0], impl)
 	} else {
 		invokers := []protocol.Invoker{}
 		var regUrl *common.URL
 		for _, u := range refconfig.urls {
-			invokers = append(invokers, extension.GetProtocol(u.Protocol).Refer(*u))
+			invokers = append(invokers, extension.GetProtocol(u.Protocol).Refer(*u, impl))
 			if u.Protocol == constant.REGISTRY_PROTOCOL {
 				regUrl = u
 			}
@@ -207,7 +207,7 @@ func (refconfig *ReferenceConfig) GenericLoad(id string) {
 	genericService := NewGenericService(refconfig.id)
 	SetConsumerService(genericService)
 	refconfig.id = id
-	refconfig.Refer()
+	refconfig.Refer(genericService)
 	refconfig.Implement(genericService)
 	return
 }

config/reference_config_test.go

@@ -184,7 +184,7 @@ func Test_ReferMultireg(t *testing.T) {
 	extension.SetCluster("registryAware", cluster_impl.NewRegistryAwareCluster)
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
 	}
@@ -197,7 +197,7 @@ func Test_Refer(t *testing.T) {
 	extension.SetCluster("registryAware", cluster_impl.NewRegistryAwareCluster)
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.Equal(t, "soa.mock", reference.Params["serviceid"])
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
@@ -211,7 +211,7 @@ func Test_ReferAsync(t *testing.T) {
 	extension.SetCluster("registryAware", cluster_impl.NewRegistryAwareCluster)
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.Equal(t, "soa.mock", reference.Params["serviceid"])
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
@@ -227,7 +227,7 @@ func Test_ReferP2P(t *testing.T) {
 	m.Url = "dubbo://127.0.0.1:20000"
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
 	}
@@ -241,7 +241,7 @@ func Test_ReferMultiP2P(t *testing.T) {
 	m.Url = "dubbo://127.0.0.1:20000;dubbo://127.0.0.2:20000"
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
 	}
@@ -256,7 +256,7 @@ func Test_ReferMultiP2PWithReg(t *testing.T) {
 	m.Url = "dubbo://127.0.0.1:20000;registry://127.0.0.2:20000"
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		assert.NotNil(t, reference.invoker)
 		assert.NotNil(t, reference.pxy)
 	}
@@ -268,7 +268,7 @@ func Test_Implement(t *testing.T) {
 	extension.SetProtocol("registry", GetProtocol)
 	extension.SetCluster("registryAware", cluster_impl.NewRegistryAwareCluster)
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		reference.Implement(&MockService{})
 		assert.NotNil(t, reference.GetRPCService())
 
@@ -284,7 +284,7 @@ func Test_Forking(t *testing.T) {
 	m.Url = "dubbo://127.0.0.1:20000;registry://127.0.0.2:20000"
 
 	for _, reference := range consumerConfig.References {
-		reference.Refer()
+		reference.Refer(nil)
 		forks := int(reference.invoker.GetUrl().GetParamInt(constant.FORKS_KEY, constant.DEFAULT_FORKS))
 		assert.Equal(t, 5, forks)
 		assert.NotNil(t, reference.pxy)
@@ -301,7 +301,7 @@ func Test_Sticky(t *testing.T) {
 	m.Url = "dubbo://127.0.0.1:20000;registry://127.0.0.2:20000"
 
 	reference := consumerConfig.References["MockService"]
-	reference.Refer()
+	reference.Refer(nil)
 	referenceSticky := reference.invoker.GetUrl().GetParam(constant.STICKY_KEY, "false")
 	assert.Equal(t, "false", referenceSticky)
 
@@ -324,7 +324,7 @@ func newRegistryProtocol() protocol.Protocol {
 
 type mockRegistryProtocol struct{}
 
-func (*mockRegistryProtocol) Refer(url common.URL) protocol.Invoker {
+func (*mockRegistryProtocol) Refer(url common.URL, impl interface{}) protocol.Invoker {
 	return protocol.NewBaseInvoker(url)
 }
 

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/go-errors/errors v1.0.1 // indirect
 	github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
 	github.com/golang/mock v1.3.1
+	github.com/golang/protobuf v1.3.2
 	github.com/google/btree v1.0.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect

go.sum

@@ -449,6 +449,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/tebeka/strftime v0.1.3 h1:5HQXOqWKYRFfNyBMNVc9z5+QzuBtIXy03psIhtdJYto=
 github.com/tebeka/strftime v0.1.3/go.mod h1:7wJm3dZlpr4l/oVK0t1HYIc4rMzQ2XJlOMIUJUJH6XQ=
 github.com/tent/http-link-go v0.0.0-20130702225549-ac974c61c2f9/go.mod h1:RHkNRtSLfOK7qBTHaeSX1D6BNpI3qw7NTxsmNr4RvN8=
+github.com/tevid/gohamcrest v1.1.1 h1:ou+xSqlIw1xfGTg1uq1nif/htZ2S3EzRqLm2BP+tYU0=
 github.com/tevid/gohamcrest v1.1.1/go.mod h1:3UvtWlqm8j5JbwYZh80D/PVBt0mJ1eJiYgZMibh0H/k=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 h1:LnC5Kc/wtumK+WB441p7ynQJzVuNRJiqddSIE3IlSEQ=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=

protocol/dubbo/dubbo_invoker_test.go

@@ -40,8 +40,8 @@ func TestDubboInvoker_Invoke(t *testing.T) {
 		pendingResponses: new(sync.Map),
 		conf:             *clientConf,
 		opts: Options{
-			ConnectTimeout: 3e9,
-			RequestTimeout: 6e9,
+			ConnectTimeout: 3 * time.Second,
+			RequestTimeout: 6 * time.Second,
 		},
 	}
 	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))

protocol/dubbo/dubbo_protocol.go

@@ -66,7 +66,7 @@ func (dp *DubboProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
 	return exporter
 }
 
-func (dp *DubboProtocol) Refer(url common.URL) protocol.Invoker {
+func (dp *DubboProtocol) Refer(url common.URL, impl interface{}) protocol.Invoker {
 	invoker := NewDubboInvoker(url, NewClient(Options{
 		ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
 		RequestTimeout: config.GetConsumerConfig().RequestTimeout,

protocol/dubbo/dubbo_protocol_test.go

@@ -85,7 +85,7 @@ func TestDubboProtocol_Refer(t *testing.T) {
 		"side=provider&timeout=3000&timestamp=1556509797245")
 	assert.NoError(t, err)
 	clientConf = &ClientConfig{}
-	invoker := proto.Refer(url)
+	invoker := proto.Refer(url, nil)
 
 	// make sure url
 	eq := invoker.GetUrl().URLEqual(url)

protocol/grpc/client.go

+/*
+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 (
+	"reflect"
+)
+
+import (
+	"google.golang.org/grpc"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+)
+
+type Client struct {
+	*grpc.ClientConn
+	invoker reflect.Value
+}
+
+func NewClient(impl interface{}, url common.URL) *Client {
+	conn, err := grpc.Dial(url.Location, grpc.WithInsecure(), grpc.WithBlock())
+	if err != nil {
+		panic(err)
+	}
+
+	invoker := getInvoker(impl, conn)
+
+	return &Client{
+		ClientConn: conn,
+		invoker:    reflect.ValueOf(invoker),
+	}
+}
+
+func getInvoker(impl interface{}, conn *grpc.ClientConn) interface{} {
+	in := []reflect.Value{}
+	in = append(in, reflect.ValueOf(conn))
+	method := reflect.ValueOf(impl).MethodByName("GetDubboStub")
+	res := method.Call(in)
+	return res[0].Interface()
+}

protocol/grpc/client_test.go

+/*
+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 (
+	"context"
+	"reflect"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol/grpc/internal"
+)
+
+func TestGetInvoker(t *testing.T) {
+	var conn *grpc.ClientConn
+	var impl *internal.GrpcGreeterImpl
+	invoker := getInvoker(impl, conn)
+
+	i := reflect.TypeOf(invoker)
+	expected := reflect.TypeOf(internal.NewGreeterClient(nil))
+	assert.Equal(t, i, expected)
+}
+
+func TestNewClient(t *testing.T) {
+	go internal.InitGrpcServer()
+	defer internal.ShutdownGrpcServer()
+
+	var impl *internal.GrpcGreeterImpl
+	url, err := common.NewURL(context.Background(), "grpc://127.0.0.1:30000/GrpcGreeterImpl?accesslog=&anyhost=true&app.version=0.0.1&application=BDTService&async=false&bean.name=GrpcGreeterImpl&category=providers&cluster=failover&dubbo=dubbo-provider-golang-2.6.0&environment=dev&execute.limit=&execute.limit.rejected.handler=&generic=false&group=&interface=io.grpc.examples.helloworld.GreeterGrpc%24IGreeter&ip=192.168.1.106&loadbalance=random&methods.SayHello.loadbalance=random&methods.SayHello.retries=1&methods.SayHello.tps.limit.interval=&methods.SayHello.tps.limit.rate=&methods.SayHello.tps.limit.strategy=&methods.SayHello.weight=0&module=dubbogo+say-hello+client&name=BDTService&organization=ikurento.com&owner=ZX&pid=49427&reference.filter=cshutdown&registry.role=3&remote.timestamp=1576923717&retries=&service.filter=echo%2Ctoken%2Caccesslog%2Ctps%2Cexecute%2Cpshutdown&side=provider&timestamp=1576923740&tps.limit.interval=&tps.limit.rate=&tps.limit.rejected.handler=&tps.limit.strategy=&tps.limiter=&version=&warmup=100!")
+	assert.Nil(t, err)
+	cli := NewClient(impl, url)
+	assert.NotNil(t, cli)
+}

protocol/grpc/common_test.go

+/*
+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 (
+	"context"
+	"fmt"
+)
+
+import (
+	native_grpc "google.golang.org/grpc"
+)
+
+import (
+	"github.com/apache/dubbo-go/config"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/grpc/internal"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+// userd grpc-dubbo biz service
+func addService() {
+	config.SetProviderService(newGreeterProvider())
+}
+
+type greeterProvider struct {
+	*greeterProviderBase
+}
+
+func newGreeterProvider() *greeterProvider {
+	return &greeterProvider{
+		greeterProviderBase: &greeterProviderBase{},
+	}
+}
+
+func (g *greeterProvider) SayHello(ctx context.Context, req *internal.HelloRequest) (reply *internal.HelloReply, err error) {
+	fmt.Printf("req: %v", req)
+	return &internal.HelloReply{Message: "this is message from reply"}, nil
+}
+
+func (g *greeterProvider) Reference() string {
+	return "GrpcGreeterImpl"
+}
+
+// code generated by greeter.go
+type greeterProviderBase struct {
+	proxyImpl protocol.Invoker
+}
+
+func (g *greeterProviderBase) SetProxyImpl(impl protocol.Invoker) {
+	g.proxyImpl = impl
+}
+
+func (g *greeterProviderBase) GetProxyImpl() protocol.Invoker {
+	return g.proxyImpl
+}
+
+func (g *greeterProviderBase) ServiceDesc() *native_grpc.ServiceDesc {
+	return &native_grpc.ServiceDesc{
+		ServiceName: "helloworld.Greeter",
+		HandlerType: (*internal.GreeterServer)(nil),
+		Methods: []native_grpc.MethodDesc{
+			{
+				MethodName: "SayHello",
+				Handler:    _DUBBO_Greeter_SayHello_Handler,
+			},
+		},
+		Streams:  []native_grpc.StreamDesc{},
+		Metadata: "helloworld.proto",
+	}
+}
+
+func _DUBBO_Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor native_grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(internal.HelloRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	base := srv.(DubboGrpcService)
+
+	args := []interface{}{}
+	args = append(args, in)
+	invo := invocation.NewRPCInvocation("SayHello", args, nil)
+
+	if interceptor == nil {
+		result := base.GetProxyImpl().Invoke(invo)
+		return result.Result(), result.Error()
+	}
+	info := &native_grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/helloworld.Greeter/SayHello",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		result := base.GetProxyImpl().Invoke(invo)
+		return result.Result(), result.Error()
+	}
+	return interceptor(ctx, in, info, handler)
+}

protocol/grpc/grpc_exporter.go

+/*
+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 (
+	"sync"
+)
+
+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/protocol"
+)
+
+type GrpcExporter struct {
+	*protocol.BaseExporter
+}
+
+func NewGrpcExporter(key string, invoker protocol.Invoker, exporterMap *sync.Map) *GrpcExporter {
+	return &GrpcExporter{
+		BaseExporter: protocol.NewBaseExporter(key, invoker, exporterMap),
+	}
+}
+
+func (gg *GrpcExporter) Unexport() {
+	serviceId := gg.GetInvoker().GetUrl().GetParam(constant.BEAN_NAME_KEY, "")
+	gg.BaseExporter.Unexport()
+	err := common.ServiceMap.UnRegister(GRPC, serviceId)
+	if err != nil {
+		logger.Errorf("[GrpcExporter.Unexport] error: %v", err)
+	}
+}

protocol/grpc/grpc_invoker.go

+/*
+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 (
+	"context"
+	"reflect"
+	"sync"
+)
+
+import (
+	"github.com/pkg/errors"
+	"google.golang.org/grpc/connectivity"
+)
+
+import (
+	hessian2 "github.com/apache/dubbo-go-hessian2"
+
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+var ErrNoReply = errors.New("request need @response")
+
+type GrpcInvoker struct {
+	protocol.BaseInvoker
+	quitOnce sync.Once
+	client   *Client
+}
+
+func NewGrpcInvoker(url common.URL, client *Client) *GrpcInvoker {
+	return &GrpcInvoker{
+		BaseInvoker: *protocol.NewBaseInvoker(url),
+		client:      client,
+	}
+}
+
+func (gi *GrpcInvoker) Invoke(invocation protocol.Invocation) protocol.Result {
+	var (
+		result protocol.RPCResult
+	)
+
+	if invocation.Reply() == nil {
+		result.Err = ErrNoReply
+	}
+
+	in := []reflect.Value{}
+	in = append(in, reflect.ValueOf(context.Background()))
+	in = append(in, invocation.ParameterValues()...)
+
+	methodName := invocation.MethodName()
+	method := gi.client.invoker.MethodByName(methodName)
+	res := method.Call(in)
+
+	result.Rest = res[0]
+	// check err
+	if !res[1].IsNil() {
+		result.Err = res[1].Interface().(error)
+	} else {
+		_ = hessian2.ReflectResponse(res[0], invocation.Reply())
+	}
+
+	return &result
+}
+
+func (gi *GrpcInvoker) IsAvailable() bool {
+	return gi.BaseInvoker.IsAvailable() && gi.client.GetState() != connectivity.Shutdown
+}
+
+func (gi *GrpcInvoker) IsDestroyed() bool {
+	return gi.BaseInvoker.IsDestroyed() && gi.client.GetState() == connectivity.Shutdown
+}
+
+func (gi *GrpcInvoker) Destroy() {
+	gi.quitOnce.Do(func() {
+		gi.BaseInvoker.Destroy()
+
+		if gi.client != nil {
+			_ = gi.client.Close()
+		}
+	})
+}

protocol/grpc/grpc_invoker_test.go

+/*
+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 (
+	"context"
+	"reflect"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol/grpc/internal"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestInvoke(t *testing.T) {
+	go internal.InitGrpcServer()
+	defer internal.ShutdownGrpcServer()
+
+	var impl *internal.GrpcGreeterImpl
+	url, err := common.NewURL(context.Background(), "grpc://127.0.0.1:30000/GrpcGreeterImpl?accesslog=&anyhost=true&app.version=0.0.1&application=BDTService&async=false&bean.name=GrpcGreeterImpl&category=providers&cluster=failover&dubbo=dubbo-provider-golang-2.6.0&environment=dev&execute.limit=&execute.limit.rejected.handler=&generic=false&group=&interface=io.grpc.examples.helloworld.GreeterGrpc%24IGreeter&ip=192.168.1.106&loadbalance=random&methods.SayHello.loadbalance=random&methods.SayHello.retries=1&methods.SayHello.tps.limit.interval=&methods.SayHello.tps.limit.rate=&methods.SayHello.tps.limit.strategy=&methods.SayHello.weight=0&module=dubbogo+say-hello+client&name=BDTService&organization=ikurento.com&owner=ZX&pid=49427&reference.filter=cshutdown&registry.role=3&remote.timestamp=1576923717&retries=&service.filter=echo%2Ctoken%2Caccesslog%2Ctps%2Cexecute%2Cpshutdown&side=provider&timestamp=1576923740&tps.limit.interval=&tps.limit.rate=&tps.limit.rejected.handler=&tps.limit.strategy=&tps.limiter=&version=&warmup=100!")
+	assert.Nil(t, err)
+
+	cli := NewClient(impl, url)
+
+	invoker := NewGrpcInvoker(url, cli)
+
+	args := []reflect.Value{}
+	args = append(args, reflect.ValueOf(&internal.HelloRequest{Name: "request name"}))
+	bizReply := &internal.HelloReply{}
+	invo := invocation.NewRPCInvocationWithOptions(invocation.WithMethodName("SayHello"),
+		invocation.WithParameterValues(args), invocation.WithReply(bizReply))
+	res := invoker.Invoke(invo)
+	assert.Nil(t, res.Error())
+	assert.NotNil(t, res.Result())
+	assert.Equal(t, "Hello request name", bizReply.Message)
+}

protocol/grpc/grpc_protocol.go

+/*
+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 (
+	"sync"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const GRPC = "grpc"
+
+func init() {
+	extension.SetProtocol(GRPC, GetProtocol)
+}
+
+var grpcProtocol *GrpcProtocol
+
+type GrpcProtocol struct {
+	protocol.BaseProtocol
+	serverMap  map[string]*Server
+	serverLock sync.Mutex
+}
+
+func NewGRPCProtocol() *GrpcProtocol {
+	return &GrpcProtocol{
+		BaseProtocol: protocol.NewBaseProtocol(),
+		serverMap:    make(map[string]*Server),
+	}
+}
+
+func (gp *GrpcProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
+	url := invoker.GetUrl()
+	serviceKey := url.ServiceKey()
+	exporter := NewGrpcExporter(serviceKey, invoker, gp.ExporterMap())
+	gp.SetExporterMap(serviceKey, exporter)
+	logger.Infof("Export service: %s", url.String())
+	gp.openServer(url)
+	return exporter
+}
+
+func (gp *GrpcProtocol) openServer(url common.URL) {
+	_, ok := gp.serverMap[url.Location]
+	if !ok {
+		_, ok := gp.ExporterMap().Load(url.ServiceKey())
+		if !ok {
+			panic("[GrpcProtocol]" + url.Key() + "is not existing")
+		}
+
+		gp.serverLock.Lock()
+		_, ok = gp.serverMap[url.Location]
+		if !ok {
+			srv := NewServer()
+			gp.serverMap[url.Location] = srv
+			srv.Start(url)
+		}
+		gp.serverLock.Unlock()
+	}
+}
+
+func (gp *GrpcProtocol) Refer(url common.URL, impl interface{}) protocol.Invoker {
+	invoker := NewGrpcInvoker(url, NewClient(impl, url))
+	gp.SetInvokers(invoker)
+	logger.Infof("Refer service: %s", url.String())
+	return invoker
+}
+
+func (gp *GrpcProtocol) Destroy() {
+	logger.Infof("GrpcProtocol destroy.")
+
+	gp.BaseProtocol.Destroy()
+
+	for key, server := range gp.serverMap {
+		delete(gp.serverMap, key)
+		server.Stop()
+	}
+}
+
+func GetProtocol() protocol.Protocol {
+	if grpcProtocol == nil {
+		grpcProtocol = NewGRPCProtocol()
+	}
+	return grpcProtocol
+}

protocol/grpc/grpc_protocol_test.go

+/*
+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 (
+	"context"
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/grpc/internal"
+)
+
+func TestGrpcProtocol_Export(t *testing.T) {
+	// Export
+	addService()
+
+	proto := GetProtocol()
+	url, err := common.NewURL(context.Background(), "grpc://127.0.0.1:40000/GrpcGreeterImpl?accesslog=&app.version=0.0.1&application=BDTService&bean.name=GrpcGreeterImpl&cluster=failover&environment=dev&execute.limit=&execute.limit.rejected.handler=&group=&interface=io.grpc.examples.helloworld.GreeterGrpc%24IGreeter&loadbalance=random&methods.SayHello.loadbalance=random&methods.SayHello.retries=1&methods.SayHello.tps.limit.interval=&methods.SayHello.tps.limit.rate=&methods.SayHello.tps.limit.strategy=&methods.SayHello.weight=0&module=dubbogo+say-hello+client&name=BDTService&organization=ikurento.com&owner=ZX&registry.role=3&retries=&service.filter=echo%2Ctoken%2Caccesslog%2Ctps%2Cexecute%2Cpshutdown&timestamp=1576923717&tps.limit.interval=&tps.limit.rate=&tps.limit.rejected.handler=&tps.limit.strategy=&tps.limiter=&version=&warmup=100")
+	assert.NoError(t, err)
+	exporter := proto.Export(protocol.NewBaseInvoker(url))
+	time.Sleep(time.Second)
+
+	// make sure url
+	eq := exporter.GetInvoker().GetUrl().URLEqual(url)
+	assert.True(t, eq)
+
+	// make sure exporterMap after 'Unexport'
+	_, ok := proto.(*GrpcProtocol).ExporterMap().Load(url.ServiceKey())
+	assert.True(t, ok)
+	exporter.Unexport()
+	_, ok = proto.(*GrpcProtocol).ExporterMap().Load(url.ServiceKey())
+	assert.False(t, ok)
+
+	// make sure serverMap after 'Destroy'
+	_, ok = proto.(*GrpcProtocol).serverMap[url.Location]
+	assert.True(t, ok)
+	proto.Destroy()
+	_, ok = proto.(*GrpcProtocol).serverMap[url.Location]
+	assert.False(t, ok)
+}
+
+func TestGrpcProtocol_Refer(t *testing.T) {
+	go internal.InitGrpcServer()
+	defer internal.ShutdownGrpcServer()
+	time.Sleep(time.Second)
+
+	proto := GetProtocol()
+	url, err := common.NewURL(context.Background(), "grpc://127.0.0.1:30000/GrpcGreeterImpl?accesslog=&anyhost=true&app.version=0.0.1&application=BDTService&async=false&bean.name=GrpcGreeterImpl&category=providers&cluster=failover&dubbo=dubbo-provider-golang-2.6.0&environment=dev&execute.limit=&execute.limit.rejected.handler=&generic=false&group=&interface=io.grpc.examples.helloworld.GreeterGrpc%24IGreeter&ip=192.168.1.106&loadbalance=random&methods.SayHello.loadbalance=random&methods.SayHello.retries=1&methods.SayHello.tps.limit.interval=&methods.SayHello.tps.limit.rate=&methods.SayHello.tps.limit.strategy=&methods.SayHello.weight=0&module=dubbogo+say-hello+client&name=BDTService&organization=ikurento.com&owner=ZX&pid=49427&reference.filter=cshutdown&registry.role=3&remote.timestamp=1576923717&retries=&service.filter=echo%2Ctoken%2Caccesslog%2Ctps%2Cexecute%2Cpshutdown&side=provider&timestamp=1576923740&tps.limit.interval=&tps.limit.rate=&tps.limit.rejected.handler=&tps.limit.strategy=&tps.limiter=&version=&warmup=100!")
+	assert.NoError(t, err)
+	var impl *internal.GrpcGreeterImpl
+	invoker := proto.Refer(url, impl)
+
+	// make sure url
+	eq := invoker.GetUrl().URLEqual(url)
+	assert.True(t, eq)
+
+	// make sure invokers after 'Destroy'
+	invokersLen := len(proto.(*GrpcProtocol).Invokers())
+	assert.Equal(t, 1, invokersLen)
+	proto.Destroy()
+	invokersLen = len(proto.(*GrpcProtocol).Invokers())
+	assert.Equal(t, 0, invokersLen)
+}

protocol/grpc/internal/client.go

+/*
+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 internal
+
+import (
+	"context"
+)
+
+import (
+	"google.golang.org/grpc"
+)
+
+// used for dubbo-grpc biz client
+type GrpcGreeterImpl struct {
+	SayHello func(ctx context.Context, in *HelloRequest, out *HelloReply) error
+}
+
+func (u *GrpcGreeterImpl) Reference() string {
+	return "GrpcGreeterImpl"
+}
+
+func (u *GrpcGreeterImpl) GetDubboStub(cc *grpc.ClientConn) GreeterClient {
+	return NewGreeterClient(cc)
+}

protocol/grpc/internal/doc.go

+/*
+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.
+*/
+
+// just for test, never use internal for production.
+package internal

protocol/grpc/internal/helloworld.pb.go

+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: helloworld.proto
+
+package internal
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/golang/protobuf/proto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+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.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+// The request message containing the user's name.
+type HelloRequest struct {
+	Name                 string   `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *HelloRequest) Reset()         { *m = HelloRequest{} }
+func (m *HelloRequest) String() string { return proto.CompactTextString(m) }
+func (*HelloRequest) ProtoMessage()    {}
+func (*HelloRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_17b8c58d586b62f2, []int{0}
+}
+
+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)
+}
+
+var xxx_messageInfo_HelloRequest proto.InternalMessageInfo
+
+func (m *HelloRequest) GetName() string {
+	if m != nil {
+		return m.Name
+	}
+	return ""
+}
+
+// The response message containing the greetings
+type HelloReply struct {
+	Message              string   `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *HelloReply) Reset()         { *m = HelloReply{} }
+func (m *HelloReply) String() string { return proto.CompactTextString(m) }
+func (*HelloReply) ProtoMessage()    {}
+func (*HelloReply) Descriptor() ([]byte, []int) {
+	return fileDescriptor_17b8c58d586b62f2, []int{1}
+}
+
+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)
+}
+
+var xxx_messageInfo_HelloReply proto.InternalMessageInfo
+
+func (m *HelloReply) GetMessage() string {
+	if m != nil {
+		return m.Message
+	}
+	return ""
+}
+
+func init() {
+	proto.RegisterType((*HelloRequest)(nil), "helloworld.HelloRequest")
+	proto.RegisterType((*HelloReply)(nil), "helloworld.HelloReply")
+}
+
+func init() { proto.RegisterFile("helloworld.proto", fileDescriptor_17b8c58d586b62f2) }
+
+var fileDescriptor_17b8c58d586b62f2 = []byte{
+	// 175 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0xc8, 0x48, 0xcd, 0xc9,
+	0xc9, 0x2f, 0xcf, 0x2f, 0xca, 0x49, 0xd1, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x42, 0x88,
+	0x28, 0x29, 0x71, 0xf1, 0x78, 0x80, 0x78, 0x41, 0xa9, 0x85, 0xa5, 0xa9, 0xc5, 0x25, 0x42, 0x42,
+	0x5c, 0x2c, 0x79, 0x89, 0xb9, 0xa9, 0x12, 0x8c, 0x0a, 0x8c, 0x1a, 0x9c, 0x41, 0x60, 0xb6, 0x92,
+	0x1a, 0x17, 0x17, 0x54, 0x4d, 0x41, 0x4e, 0xa5, 0x90, 0x04, 0x17, 0x7b, 0x6e, 0x6a, 0x71, 0x71,
+	0x62, 0x3a, 0x4c, 0x11, 0x8c, 0x6b, 0xe4, 0xc9, 0xc5, 0xee, 0x5e, 0x94, 0x9a, 0x5a, 0x92, 0x5a,
+	0x24, 0x64, 0xc7, 0xc5, 0x11, 0x9c, 0x58, 0x09, 0xd6, 0x25, 0x24, 0xa1, 0x87, 0xe4, 0x02, 0x64,
+	0xcb, 0xa4, 0xc4, 0xb0, 0xc8, 0x14, 0xe4, 0x54, 0x2a, 0x31, 0x38, 0x19, 0x70, 0x49, 0x67, 0xe6,
+	0xeb, 0xa5, 0x17, 0x15, 0x24, 0xeb, 0xa5, 0x56, 0x24, 0xe6, 0x16, 0xe4, 0xa4, 0x16, 0x23, 0xa9,
+	0x75, 0xe2, 0x07, 0x2b, 0x0e, 0x07, 0xb1, 0x03, 0x40, 0x5e, 0x0a, 0x60, 0x4c, 0x62, 0x03, 0xfb,
+	0xcd, 0x18, 0x10, 0x00, 0x00, 0xff, 0xff, 0x0f, 0xb7, 0xcd, 0xf2, 0xef, 0x00, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// 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
+
+// GreeterClient is the client API for Greeter service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type GreeterClient interface {
+	// Sends a greeting
+	SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error)
+}
+
+type greeterClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewGreeterClient(cc *grpc.ClientConn) GreeterClient {
+	return &greeterClient{cc}
+}
+
+func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {
+	out := new(HelloReply)
+	err := c.cc.Invoke(ctx, "/helloworld.Greeter/SayHello", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// GreeterServer is the server API for Greeter service.
+type GreeterServer interface {
+	// Sends a greeting
+	SayHello(context.Context, *HelloRequest) (*HelloReply, error)
+}
+
+// UnimplementedGreeterServer can be embedded to have forward compatible implementations.
+type UnimplementedGreeterServer struct {
+}
+
+func (*UnimplementedGreeterServer) SayHello(ctx context.Context, req *HelloRequest) (*HelloReply, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SayHello not implemented")
+}
+
+func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
+	s.RegisterService(&_Greeter_serviceDesc, srv)
+}
+
+func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(HelloRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(GreeterServer).SayHello(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/helloworld.Greeter/SayHello",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _Greeter_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "helloworld.Greeter",
+	HandlerType: (*GreeterServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "SayHello",
+			Handler:    _Greeter_SayHello_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "helloworld.proto",
+}