Skip to content
Snippets Groups Projects
Commit a77aebab authored by fangyincheng's avatar fangyincheng
Browse files

Merge remote-tracking branch 'apache/develop' into develop

parents c2ad2814 b1104f91
No related branches found
No related tags found
No related merge requests found
Showing
with 948 additions and 28 deletions
......@@ -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)
......
......@@ -40,7 +40,7 @@ type ConfigCenterConfig struct {
Username string `yaml:"username" json:"username,omitempty"`
Password string `yaml:"password" json:"password,omitempty"`
ConfigFile string `default:"dubbo.properties" yaml:"config_file" json:"config_file,omitempty"`
Namespace string `default:"dubbo.properties" yaml:"namespace" json:"namespace,omitempty"`
Namespace string `default:"dubbo" yaml:"namespace" json:"namespace,omitempty"`
AppConfigFile string `default:"dubbo.properties" yaml:"app_config_file" json:"app_config_file,omitempty"`
AppId string `default:"dubbo" yaml:"app_id" json:"app_id,omitempty"`
TimeoutStr string `yaml:"timeout" json:"timeout,omitempty"`
......
......@@ -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
......
......@@ -89,8 +89,12 @@ func (refconfig *ReferenceConfig) UnmarshalYAML(unmarshal func(interface{}) erro
return nil
}
func (refconfig *ReferenceConfig) Refer() {
url := common.NewURLWithOptions(common.WithPath(refconfig.id), common.WithProtocol(refconfig.Protocol), common.WithParams(refconfig.getUrlMap()))
func (refconfig *ReferenceConfig) Refer(impl interface{}) {
url := common.NewURLWithOptions(common.WithPath(refconfig.id),
common.WithProtocol(refconfig.Protocol),
common.WithParams(refconfig.getUrlMap()),
common.WithParamsValue(constant.BEAN_NAME_KEY, refconfig.id),
)
//1. user specified URL, could be peer-to-peer address, or register center's address.
if refconfig.Url != "" {
......@@ -214,7 +218,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
}
......@@ -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)
......
......@@ -58,7 +58,7 @@ func newApolloConfiguration(url *common.URL) (*apolloConfiguration, error) {
configCluster := url.GetParam(constant.CONFIG_CLUSTER_KEY, "")
appId := url.GetParam(constant.CONFIG_APP_ID_KEY, "")
namespaces := url.GetParam(constant.CONFIG_NAMESPACE_KEY, getProperties(DEFAULT_GROUP))
namespaces := getProperties(url.GetParam(constant.CONFIG_NAMESPACE_KEY, DEFAULT_GROUP))
c.appConf = &agollo.AppConfig{
AppId: appId,
Cluster: configCluster,
......
......@@ -192,7 +192,7 @@ func initMockApollo(t *testing.T) *apolloConfiguration {
Address: "106.12.25.204:8080",
AppId: "testApplication_yang",
Cluster: "dev",
Namespace: "mockDubbog.properties",
Namespace: "mockDubbog",
}}
apollo := initApollo()
apolloUrl := strings.ReplaceAll(apollo.URL, "http", "apollo")
......
......@@ -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
......
......@@ -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))
......
/*
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"
"github.com/apache/dubbo-go/common/constant"
"github.com/apache/dubbo-go/config"
)
type Client struct {
*grpc.ClientConn
invoker reflect.Value
}
func NewClient(url common.URL) *Client {
conn, err := grpc.Dial(url.Location, grpc.WithInsecure(), grpc.WithBlock())
if err != nil {
panic(err)
}
key := url.GetParam(constant.BEAN_NAME_KEY, "")
impl := config.GetConsumerService(key)
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()
}
/*
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()
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(url)
assert.NotNil(t, cli)
}
/*
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)
}
/*
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)
}
}
/*
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()
}
})
}
/*
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()
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(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)
}
/*
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) protocol.Invoker {
invoker := NewGrpcInvoker(url, NewClient(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
}
/*
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)
invoker := proto.Refer(url)
// 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)
}
/*
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"
)
import (
"github.com/apache/dubbo-go/config"
)
func init() {
config.SetConsumerService(&GrpcGreeterImpl{})
}
// 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)
}
/*
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
/*
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.
*/
// 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",
}
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment