From d68c44e689eb15c4617d13362fe01e28ec23f069 Mon Sep 17 00:00:00 2001
From: fangyincheng <fangyincheng@sina.com>
Date: Mon, 29 Apr 2019 15:04:10 +0800
Subject: [PATCH] Tst&Fix: add dubbo protocol test and fix bug
---
config/rpc_service.go | 26 +++++-----
go.mod | 9 +---
go.sum | 8 ---
protocol/dubbo/client.go | 19 +++++---
protocol/dubbo/dubbo_exporter.go | 6 ++-
protocol/dubbo/dubbo_invoker.go | 2 +-
protocol/dubbo/dubbo_protocol.go | 6 ++-
protocol/dubbo/dubbo_protocol_test.go | 70 +++++++++++++++++++++++++++
protocol/dubbo/listener.go | 8 ++-
protocol/dubbo/readwriter.go | 2 +-
protocol/invoker.go | 4 +-
protocol/jsonrpc/jsonrpc_exporter.go | 6 ++-
protocol/jsonrpc/jsonrpc_invoker.go | 2 +-
protocol/jsonrpc/jsonrpc_protocol.go | 6 ++-
protocol/jsonrpc/server.go | 2 +-
protocol/protocol.go | 33 ++++++++-----
16 files changed, 153 insertions(+), 56 deletions(-)
create mode 100644 protocol/dubbo/dubbo_protocol_test.go
diff --git a/config/rpc_service.go b/config/rpc_service.go
index 3d1e0482f..3a627177e 100644
--- a/config/rpc_service.go
+++ b/config/rpc_service.go
@@ -26,7 +26,7 @@ var (
typeOfError = reflect.TypeOf((*error)(nil)).Elem()
ServiceMap = &serviceMap{
- serviceMap: make(map[string]*Service),
+ serviceMap: make(map[string]map[string]*Service),
}
)
@@ -87,24 +87,28 @@ func (s *Service) Rcvr() reflect.Value {
//////////////////////////
type serviceMap struct {
- mutex sync.RWMutex // protects the serviceMap
- serviceMap map[string]*Service // service name -> service
+ mutex sync.RWMutex // protects the serviceMap
+ serviceMap map[string]map[string]*Service // protocol -> service name -> service
}
-func (sm *serviceMap) GetService(name string) *Service {
+func (sm *serviceMap) GetService(protocol, name string) *Service {
sm.mutex.RLock()
defer sm.mutex.RUnlock()
- if s, ok := sm.serviceMap[name]; ok {
- return s
+ if s, ok := sm.serviceMap[protocol]; ok {
+ if srv, ok := s[name]; ok {
+ return srv
+ }
+ return nil
}
return nil
}
-func (sm *serviceMap) Register(rcvr RPCService) (string, error) {
+// todo:Register is called by 'ServiceConfig'
+func (sm *serviceMap) Register(protocol string, rcvr RPCService) (string, error) {
sm.mutex.Lock()
defer sm.mutex.Unlock()
- if sm.serviceMap == nil {
- sm.serviceMap = make(map[string]*Service)
+ if sm.serviceMap[protocol] == nil {
+ sm.serviceMap[protocol] = make(map[string]*Service)
}
s := new(Service)
@@ -123,7 +127,7 @@ func (sm *serviceMap) Register(rcvr RPCService) (string, error) {
}
sname = rcvr.Service()
- if server := sm.GetService(sname); server == nil {
+ if server := sm.GetService(protocol, sname); server != nil {
return "", jerrors.New("service already defined: " + sname)
}
s.name = sname
@@ -138,7 +142,7 @@ func (sm *serviceMap) Register(rcvr RPCService) (string, error) {
log.Error(s)
return "", jerrors.New(s)
}
- sm.serviceMap[s.name] = s
+ sm.serviceMap[protocol][s.name] = s
return strings.TrimSuffix(methods, ","), nil
}
diff --git a/go.mod b/go.mod
index 5d8c0aec0..aa565e80b 100644
--- a/go.mod
+++ b/go.mod
@@ -4,15 +4,10 @@ require (
github.com/AlexStocks/getty v0.0.0-20190331201845-1ca64ac5a589
github.com/AlexStocks/goext v0.3.2
github.com/AlexStocks/log4go v1.0.2
- github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect
- github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect
github.com/dubbogo/hessian2 v0.0.0-20190410112310-f093e4436e31
github.com/juju/errors v0.0.0-20190207033735-e65537c515d7
- github.com/juju/utils v0.0.0-20180820210520-bf9cc5bdd62d
- github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223 // indirect
- github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
github.com/samuel/go-zookeeper v0.0.0-20180130194729-c4fab1ac1bec
- github.com/tevino/abool v0.0.0-20170917061928-9b9efcf221b5 // indirect
- gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect
+ github.com/stretchr/testify v1.3.0
+ github.com/tevino/abool v0.0.0-20170917061928-9b9efcf221b5
gopkg.in/yaml.v2 v2.2.2
)
diff --git a/go.sum b/go.sum
index b12dffcd7..583504cf1 100644
--- a/go.sum
+++ b/go.sum
@@ -8,10 +8,6 @@ github.com/AlexStocks/goext v0.3.2/go.mod h1:3M5j9Pjge4CdkNg2WIjRLUeoPedJHHKwkkg
github.com/AlexStocks/log4go v1.0.2 h1:1K5WM8KjSUECaoXUl8FSF05KGeCJDfBrhKUBsxwUvhk=
github.com/AlexStocks/log4go v1.0.2/go.mod h1:6kCCRo/orDo8mh5CEDOeuSSM674wBQ8M6E0K8dVOIz4=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
-github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY=
-github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
github.com/cheekybits/is v0.0.0-20150225183255-68e9c0620927/go.mod h1:h/aW8ynjgkuj+NQRlZcDbAbM1ORAbXjXX77sX7T289U=
@@ -86,8 +82,6 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0j
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223 h1:F9x/1yl3T2AeKLr2AMdilSD8+f9bvMnNN8VS5iDtovc=
-github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd h1:22rhYEzttbrnKjgYh5pifnDluXHHcJ3uSOi2l8Nw+9A=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd/go.mod h1:JrOIxq3vDxvtuEI7Kmm2yqkuBfuT9DMLFMnCyYHLaKM=
github.com/name5566/leaf v0.0.0-20181103040206-1364c176dfbd/go.mod h1:JrOIxq3vDxvtuEI7Kmm2yqkuBfuT9DMLFMnCyYHLaKM=
@@ -159,8 +153,6 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
-gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
-gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/protocol/dubbo/client.go b/protocol/dubbo/client.go
index c1c63f807..2f1eed2ea 100644
--- a/protocol/dubbo/client.go
+++ b/protocol/dubbo/client.go
@@ -1,9 +1,7 @@
package dubbo
import (
- "gopkg.in/yaml.v2"
"io/ioutil"
- "math/rand"
"os"
"strings"
"sync"
@@ -16,6 +14,7 @@ import (
log "github.com/AlexStocks/log4go"
"github.com/dubbogo/hessian2"
jerrors "github.com/juju/errors"
+ "gopkg.in/yaml.v2"
)
import (
@@ -29,34 +28,40 @@ var (
errClientClosed = jerrors.New("client closed")
errClientReadTimeout = jerrors.New("client read timeout")
- conf *ClientConfig
+ clientConf *ClientConfig
)
const CONF_CLIENT_FILE_PATH = "CONF_CLIENT_FILE_PATH"
func init() {
- rand.Seed(time.Now().UnixNano())
- conf = &ClientConfig{}
// load clientconfig from *.yml
path := os.Getenv(CONF_CLIENT_FILE_PATH)
if path == "" {
log.Info("CONF_CLIENT_FILE_PATH is null")
+ return
}
file, err := ioutil.ReadFile(path)
if err != nil {
log.Error(jerrors.Trace(err))
+ return
}
+ conf := &ClientConfig{}
err = yaml.Unmarshal(file, conf)
if err != nil {
log.Error(jerrors.Trace(err))
+ return
}
if err := conf.CheckValidity(); err != nil {
log.Error("ClientConfig check failed: ", jerrors.Trace(err))
+ return
}
+
+ clientConf = conf
+
}
type CallOptions struct {
@@ -127,9 +132,9 @@ func NewClient() *Client {
c := &Client{
pendingResponses: make(map[SequenceType]*PendingResponse),
- conf: *conf,
+ conf: *clientConf,
}
- c.pool = newGettyRPCClientConnPool(c, conf.PoolSize, time.Duration(int(time.Second)*conf.PoolTTL))
+ c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
return c
}
diff --git a/protocol/dubbo/dubbo_exporter.go b/protocol/dubbo/dubbo_exporter.go
index 274fdd301..111b1782f 100644
--- a/protocol/dubbo/dubbo_exporter.go
+++ b/protocol/dubbo/dubbo_exporter.go
@@ -1,5 +1,9 @@
package dubbo
+import (
+ "sync"
+)
+
import (
"github.com/dubbo/dubbo-go/protocol"
)
@@ -8,7 +12,7 @@ type DubboExporter struct {
protocol.BaseExporter
}
-func NewDubboExporter(key string, invoker protocol.Invoker, exporterMap map[string]protocol.Exporter) *DubboExporter {
+func NewDubboExporter(key string, invoker protocol.Invoker, exporterMap *sync.Map) *DubboExporter {
return &DubboExporter{
BaseExporter: *protocol.NewBaseExporter(key, invoker, exporterMap),
}
diff --git a/protocol/dubbo/dubbo_invoker.go b/protocol/dubbo/dubbo_invoker.go
index 9d737877c..c5882675b 100644
--- a/protocol/dubbo/dubbo_invoker.go
+++ b/protocol/dubbo/dubbo_invoker.go
@@ -26,7 +26,7 @@ type DubboInvoker struct {
func NewDubboInvoker(url config.IURL, client *Client) *DubboInvoker {
return &DubboInvoker{
- BaseInvoker: protocol.NewBaseInvoker(url),
+ BaseInvoker: *protocol.NewBaseInvoker(url),
client: client,
}
}
diff --git a/protocol/dubbo/dubbo_protocol.go b/protocol/dubbo/dubbo_protocol.go
index 271d3727d..eb6807b69 100644
--- a/protocol/dubbo/dubbo_protocol.go
+++ b/protocol/dubbo/dubbo_protocol.go
@@ -62,7 +62,11 @@ func (dp *DubboProtocol) Destroy() {
}
func (dp *DubboProtocol) openServer(url config.URL) {
- srv := NewServer(dp.ExporterMap()[url.Key()])
+ exporter, ok := dp.ExporterMap().Load(url.Key())
+ if !ok {
+ panic("[DubboProtocol]" + url.Key() + "is not existing")
+ }
+ srv := NewServer(exporter.(protocol.Exporter))
dp.serverMap[url.Location] = srv
srv.Start(url)
}
diff --git a/protocol/dubbo/dubbo_protocol_test.go b/protocol/dubbo/dubbo_protocol_test.go
new file mode 100644
index 000000000..9a7649ee4
--- /dev/null
+++ b/protocol/dubbo/dubbo_protocol_test.go
@@ -0,0 +1,70 @@
+package dubbo
+
+import (
+ "context"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "github.com/dubbo/dubbo-go/config"
+ "github.com/dubbo/dubbo-go/protocol"
+)
+
+func TestDubboProtocol_Export(t *testing.T) {
+ // Export
+ proto := GetProtocol()
+ url, err := config.NewURL(context.Background(), "dubbo://192.168.56.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
+ "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
+ "side=provider&timeout=3000×tamp=1556509797245")
+ assert.NoError(t, err)
+ srvConf = &ServerConfig{}
+ exporter := proto.Export(protocol.NewBaseInvoker(url))
+
+ // make sure url
+ eq := exporter.GetInvoker().GetUrl().URLEqual(url)
+ assert.True(t, eq)
+
+ // make sure exporterMap after 'Unexport'
+ _, ok := proto.(*DubboProtocol).ExporterMap().Load(url.Key())
+ assert.True(t, ok)
+ exporter.Unexport()
+ _, ok = proto.(*DubboProtocol).ExporterMap().Load(url.Key())
+ assert.False(t, ok)
+
+ // make sure serverMap after 'Destroy'
+ _, ok = proto.(*DubboProtocol).serverMap[url.Location]
+ assert.True(t, ok)
+ proto.Destroy()
+ _, ok = proto.(*DubboProtocol).serverMap[url.Location]
+ assert.False(t, ok)
+}
+
+func TestDubboProtocol_Refer(t *testing.T) {
+ // Refer
+ proto := GetProtocol()
+ url, err := config.NewURL(context.Background(), "dubbo://192.168.56.1:20000/com.ikurento.user.UserProvider?anyhost=true&"+
+ "application=BDTService&category=providers&default.timeout=10000&dubbo=dubbo-provider-golang-1.0.0&"+
+ "environment=dev&interface=com.ikurento.user.UserProvider&ip=192.168.56.1&methods=GetUser%2C&"+
+ "module=dubbogo+user-info+server&org=ikurento.com&owner=ZX&pid=1447&revision=0.0.1&"+
+ "side=provider&timeout=3000×tamp=1556509797245")
+ assert.NoError(t, err)
+ clientConf = &ClientConfig{}
+ invoker := proto.Refer(url)
+
+ // make sure url
+ eq := invoker.GetUrl().URLEqual(url)
+ assert.True(t, eq)
+
+ // make sure invokers after 'Destroy'
+ invokersLen := len(proto.(*DubboProtocol).Invokers())
+ assert.Equal(t, 1, invokersLen)
+ proto.Destroy()
+ invokersLen = len(proto.(*DubboProtocol).Invokers())
+ assert.Equal(t, 0, invokersLen)
+}
diff --git a/protocol/dubbo/listener.go b/protocol/dubbo/listener.go
index 38a12e275..5835b3295 100644
--- a/protocol/dubbo/listener.go
+++ b/protocol/dubbo/listener.go
@@ -261,6 +261,12 @@ func (h *RpcServerHandler) callService(req *DubboPackage, ctx context.Context) {
}
svc := svcIf.(*config.Service)
method := svc.Method()[req.Service.Method]
+ if method == nil {
+ log.Error("method not found!")
+ req.Header.ResponseStatus = hessian.Response_SERVICE_NOT_FOUND
+ req.Body = nil
+ return
+ }
// prepare argv
var argv reflect.Value
@@ -280,7 +286,7 @@ func (h *RpcServerHandler) callService(req *DubboPackage, ctx context.Context) {
// prepare replyv
replyv := reflect.New(method.ReplyType().Elem())
var returnValues []reflect.Value
- if method.CtxType == nil {
+ if method.CtxType() == nil {
returnValues = method.Method().Func.Call([]reflect.Value{svc.Rcvr(), reflect.ValueOf(argvTmp), reflect.ValueOf(replyv.Interface())})
} else {
if contextv := reflect.ValueOf(ctx); contextv.IsValid() {
diff --git a/protocol/dubbo/readwriter.go b/protocol/dubbo/readwriter.go
index be720e009..2c870da83 100644
--- a/protocol/dubbo/readwriter.go
+++ b/protocol/dubbo/readwriter.go
@@ -107,7 +107,7 @@ func (p *RpcServerPackageHandler) Read(ss getty.Session, data []byte) (interface
"dubboVersion": dubboVersion,
"argsTypes": argsTypes,
"args": args,
- "service": config.ServiceMap.GetService(pkg.Service.Target),
+ "service": config.ServiceMap.GetService(DUBBO, pkg.Service.Target),
"attachments": attachments,
}
}
diff --git a/protocol/invoker.go b/protocol/invoker.go
index 922362864..03a39efa2 100644
--- a/protocol/invoker.go
+++ b/protocol/invoker.go
@@ -25,8 +25,8 @@ type BaseInvoker struct {
destroyed bool
}
-func NewBaseInvoker(url config.IURL) BaseInvoker {
- return BaseInvoker{
+func NewBaseInvoker(url config.IURL) *BaseInvoker {
+ return &BaseInvoker{
url: url,
available: true,
destroyed: false,
diff --git a/protocol/jsonrpc/jsonrpc_exporter.go b/protocol/jsonrpc/jsonrpc_exporter.go
index dd0918281..3c6c16cbb 100644
--- a/protocol/jsonrpc/jsonrpc_exporter.go
+++ b/protocol/jsonrpc/jsonrpc_exporter.go
@@ -1,5 +1,9 @@
package jsonrpc
+import (
+ "sync"
+)
+
import (
"github.com/dubbo/dubbo-go/protocol"
)
@@ -8,7 +12,7 @@ type JsonrpcExporter struct {
protocol.BaseExporter
}
-func NewJsonrpcExporter(key string, invoker protocol.Invoker, exporterMap map[string]protocol.Exporter) *JsonrpcExporter {
+func NewJsonrpcExporter(key string, invoker protocol.Invoker, exporterMap *sync.Map) *JsonrpcExporter {
return &JsonrpcExporter{
BaseExporter: *protocol.NewBaseExporter(key, invoker, exporterMap),
}
diff --git a/protocol/jsonrpc/jsonrpc_invoker.go b/protocol/jsonrpc/jsonrpc_invoker.go
index 38b929c02..36952d3e3 100644
--- a/protocol/jsonrpc/jsonrpc_invoker.go
+++ b/protocol/jsonrpc/jsonrpc_invoker.go
@@ -22,7 +22,7 @@ type JsonrpcInvoker struct {
func NewJsonrpcInvoker(url config.IURL, client *HTTPClient) *JsonrpcInvoker {
return &JsonrpcInvoker{
- BaseInvoker: protocol.NewBaseInvoker(url),
+ BaseInvoker: *protocol.NewBaseInvoker(url),
client: client,
}
}
diff --git a/protocol/jsonrpc/jsonrpc_protocol.go b/protocol/jsonrpc/jsonrpc_protocol.go
index cb6cd5258..343e03ef3 100644
--- a/protocol/jsonrpc/jsonrpc_protocol.go
+++ b/protocol/jsonrpc/jsonrpc_protocol.go
@@ -67,7 +67,11 @@ func (jp *JsonrpcProtocol) Destroy() {
}
func (jp *JsonrpcProtocol) openServer(url config.URL) {
- srv := NewServer(jp.ExporterMap()[url.Key()])
+ exporter, ok := jp.ExporterMap().Load(url.Key())
+ if !ok {
+ panic("[JsonrpcProtocol]" + url.Key() + "is not existing")
+ }
+ srv := NewServer(exporter.(protocol.Exporter))
jp.serverMap[url.Location] = srv
srv.Start(url)
}
diff --git a/protocol/jsonrpc/server.go b/protocol/jsonrpc/server.go
index 01950d3ab..465617b92 100644
--- a/protocol/jsonrpc/server.go
+++ b/protocol/jsonrpc/server.go
@@ -272,7 +272,7 @@ func serveRequest(ctx context.Context,
}
// get method
- svc := config.ServiceMap.GetService(serviceName)
+ svc := config.ServiceMap.GetService(JSONRPC, serviceName)
if svc == nil {
codec.ReadBody(nil)
return jerrors.New("cannot find svc " + serviceName)
diff --git a/protocol/protocol.go b/protocol/protocol.go
index a490ba4cb..601611515 100644
--- a/protocol/protocol.go
+++ b/protocol/protocol.go
@@ -1,8 +1,15 @@
package protocol
+import (
+ "sync"
+)
+
+import (
+ log "github.com/AlexStocks/log4go"
+)
+
import (
"github.com/dubbo/dubbo-go/config"
- "github.com/prometheus/common/log"
)
// Extension - Protocol
@@ -23,21 +30,21 @@ type Exporter interface {
/////////////////////////////
type BaseProtocol struct {
- exporterMap map[string]Exporter
+ exporterMap *sync.Map
invokers []Invoker
}
func NewBaseProtocol() BaseProtocol {
return BaseProtocol{
- exporterMap: make(map[string]Exporter),
+ exporterMap: new(sync.Map),
}
}
func (bp *BaseProtocol) SetExporterMap(key string, exporter Exporter) {
- bp.exporterMap[key] = exporter
+ bp.exporterMap.Store(key, exporter)
}
-func (bp *BaseProtocol) ExporterMap() map[string]Exporter {
+func (bp *BaseProtocol) ExporterMap() *sync.Map {
return bp.exporterMap
}
@@ -57,6 +64,7 @@ func (bp *BaseProtocol) Refer(url config.IURL) Invoker {
return nil
}
+//Destroy will destroy all invoker and exporter, so it only is called once.
func (bp *BaseProtocol) Destroy() {
// destroy invokers
for _, invoker := range bp.invokers {
@@ -67,13 +75,14 @@ func (bp *BaseProtocol) Destroy() {
bp.invokers = []Invoker{}
// unexport exporters
- for key, exporter := range bp.ExporterMap() {
+ bp.exporterMap.Range(func(key, exporter interface{}) bool {
if exporter != nil {
- exporter.Unexport()
+ exporter.(Exporter).Unexport()
} else {
- delete(bp.exporterMap, key)
+ bp.exporterMap.Delete(key)
}
- }
+ return true
+ })
}
/////////////////////////////
@@ -83,10 +92,10 @@ func (bp *BaseProtocol) Destroy() {
type BaseExporter struct {
key string
invoker Invoker
- exporterMap map[string]Exporter
+ exporterMap *sync.Map
}
-func NewBaseExporter(key string, invoker Invoker, exporterMap map[string]Exporter) *BaseExporter {
+func NewBaseExporter(key string, invoker Invoker, exporterMap *sync.Map) *BaseExporter {
return &BaseExporter{
key: key,
invoker: invoker,
@@ -102,5 +111,5 @@ func (de *BaseExporter) GetInvoker() Invoker {
func (de *BaseExporter) Unexport() {
log.Info("Exporter unexport.")
de.invoker.Destroy()
- delete(de.exporterMap, de.key)
+ de.exporterMap.Delete(de.key)
}
--
GitLab