diff --git a/README.md b/README.md
index 2035aeed4171c450748e2af55c9fda50a1f7bd7a..5c4de0559d534e03b23d136da5f5f23ff6768d55 100644
--- a/README.md
+++ b/README.md
@@ -16,6 +16,8 @@ Apache License, Version 2.0
## Release note ##
+[v1.4.0-rc1 - Mar 12, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.4.0-rc1)
+
[v1.3.0 - Mar 1, 2020](https://github.com/apache/dubbo-go/releases/tag/v1.3.0)
[v1.2.0 - Nov 15, 2019](https://github.com/apache/dubbo-go/releases/tag/v1.2.0)
@@ -28,7 +30,7 @@ Apache License, Version 2.0
Both extension module and layered project architecture is according to Apache Dubbo (including protocol layer, registry layer, cluster layer, config layer and so on), the advantage of this arch is as following: you can implement these layered interfaces in your own way, override the default implementation of dubbo-go by calling 'extension.SetXXX' of extension, complete your special needs without modifying the source code. At the same time, you are welcome to contribute implementation of useful extension to the community.
-
+
If you wanna know more about dubbo-go, please visit this reference [Project Architeture design](https://github.com/apache/dubbo-go/wiki/dubbo-go-V1.0-design)
diff --git a/README_CN.md b/README_CN.md
index 93c3e37f0a4bc42e071f19dba937a059fcf602ae..5dec68fd6134fc58bf265ea382473d724332ecc6 100644
--- a/README_CN.md
+++ b/README_CN.md
@@ -15,6 +15,8 @@ Apache License, Version 2.0
## 鍙戝竷鏃ュ織 ##
+[v1.4.0-rc1 - 2020骞�3鏈�12鏃(https://github.com/apache/dubbo-go/releases/tag/v1.4.0-rc1)
+
[v1.3.0 - 2020骞�3鏈�1鏃(https://github.com/apache/dubbo-go/releases/tag/v1.3.0)
[v1.2.0 - 2019骞�11鏈�15鏃(https://github.com/apache/dubbo-go/releases/tag/v1.2.0)
@@ -27,7 +29,7 @@ Apache License, Version 2.0
鍩轰簬dubbo鐨別xtension妯″潡鍜屽垎灞傜殑浠ｇ爜璁捐(鍖呮嫭 protocol layer, registry layer, cluster layer, config 绛夌瓑)銆傛垜浠殑鐩爣鏄細浣犲彲浠ュ杩欎簺鍒嗗眰鎺ュ彛杩涜鏂扮殑瀹炵幇锛屽苟閫氳繃璋冪敤 extension 妯″潡鐨勨€� extension.SetXXX 鈥濇柟娉曟潵瑕嗙洊 dubbo-go [鍚� go-for-apache-dubbo ]鐨勯粯璁ゅ疄鐜帮紝浠ュ畬鎴愯嚜宸辩殑鐗规畩闇€姹傝€屾棤闇€淇敼婧愪唬鐮併€傚悓鏃讹紝娆㈣繋浣犱负绀惧尯璐＄尞鏈夌敤鐨勬嫇灞曞疄鐜般€�
-
+
鍏充簬璇︾粏璁捐璇烽槄璇� [code layered design](https://github.com/apache/dubbo-go/wiki/dubbo-go-V1.0-design)
diff --git a/common/rpc_service.go b/common/rpc_service.go
index cb4bc9d96bd1a388600117d03074e09ac1aba3fb..3bb3dfe32250ee24e35b1f594509b477a4eea5c9 100644
--- a/common/rpc_service.go
+++ b/common/rpc_service.go
@@ -71,7 +71,8 @@ var (
// ServiceMap ...
// todo: lowerecas?
ServiceMap = &serviceMap{
- serviceMap: make(map[string]map[string]*Service),
+ serviceMap: make(map[string]map[string]*Service),
+ interfaceMap: make(map[string][]*Service),
}
)
@@ -152,10 +153,12 @@ func (s *Service) Rcvr() reflect.Value {
//////////////////////////
type serviceMap struct {
- mutex sync.RWMutex // protects the serviceMap
- serviceMap map[string]map[string]*Service // protocol -> service name -> service
+ mutex sync.RWMutex // protects the serviceMap
+ serviceMap map[string]map[string]*Service // protocol -> service name -> service
+ interfaceMap map[string][]*Service // interface -> service
}
+// GetService get a service defination by protocol and name
func (sm *serviceMap) GetService(protocol, name string) *Service {
sm.mutex.RLock()
defer sm.mutex.RUnlock()
@@ -168,10 +171,24 @@ func (sm *serviceMap) GetService(protocol, name string) *Service {
return nil
}
-func (sm *serviceMap) Register(protocol string, rcvr RPCService) (string, error) {
+// GetInterface get an interface defination by interface name
+func (sm *serviceMap) GetInterface(interfaceName string) []*Service {
+ sm.mutex.RLock()
+ defer sm.mutex.RUnlock()
+ if s, ok := sm.interfaceMap[interfaceName]; ok {
+ return s
+ }
+ return nil
+}
+
+// Register register a service by @interfaceName and @protocol
+func (sm *serviceMap) Register(interfaceName, protocol string, rcvr RPCService) (string, error) {
if sm.serviceMap[protocol] == nil {
sm.serviceMap[protocol] = make(map[string]*Service)
}
+ if sm.interfaceMap[interfaceName] == nil {
+ sm.interfaceMap[interfaceName] = make([]*Service, 0, 16)
+ }
s := new(Service)
s.rcvrType = reflect.TypeOf(rcvr)
@@ -206,32 +223,65 @@ func (sm *serviceMap) Register(protocol string, rcvr RPCService) (string, error)
}
sm.mutex.Lock()
sm.serviceMap[protocol][s.name] = s
+ sm.interfaceMap[interfaceName] = append(sm.interfaceMap[interfaceName], s)
sm.mutex.Unlock()
return strings.TrimSuffix(methods, ","), nil
}
-func (sm *serviceMap) UnRegister(protocol, serviceId string) error {
+// UnRegister cancel a service by @interfaceName, @protocol and @serviceId
+func (sm *serviceMap) UnRegister(interfaceName, protocol, serviceId string) error {
if protocol == "" || serviceId == "" {
return perrors.New("protocol or serviceName is nil")
}
- sm.mutex.RLock()
- svcs, ok := sm.serviceMap[protocol]
- if !ok {
- sm.mutex.RUnlock()
- return perrors.New("no services for " + protocol)
+
+ var (
+ err error
+ index = -1
+ svcs map[string]*Service
+ svrs []*Service
+ ok bool
+ )
+
+ f := func() error {
+ sm.mutex.RLock()
+ defer sm.mutex.RUnlock()
+ svcs, ok = sm.serviceMap[protocol]
+ if !ok {
+ return perrors.New("no services for " + protocol)
+ }
+ s, ok := svcs[serviceId]
+ if !ok {
+ return perrors.New("no service for " + serviceId)
+ }
+ svrs, ok = sm.interfaceMap[interfaceName]
+ if !ok {
+ return perrors.New("no service for " + interfaceName)
+ }
+ for i, svr := range svrs {
+ if svr == s {
+ index = i
+ }
+ }
+ return nil
}
- _, ok = svcs[serviceId]
- if !ok {
- sm.mutex.RUnlock()
- return perrors.New("no service for " + serviceId)
+
+ if err = f(); err != nil {
+ return err
}
- sm.mutex.RUnlock()
sm.mutex.Lock()
defer sm.mutex.Unlock()
+ sm.interfaceMap[interfaceName] = make([]*Service, 0, len(svrs))
+ for i, _ := range svrs {
+ if i != index {
+ sm.interfaceMap[interfaceName] = append(sm.interfaceMap[interfaceName], svrs[i])
+ }
+ }
delete(svcs, serviceId)
- delete(sm.serviceMap, protocol)
+ if len(sm.serviceMap) == 0 {
+ delete(sm.serviceMap, protocol)
+ }
return nil
}
diff --git a/common/rpc_service_test.go b/common/rpc_service_test.go
index 8c9b9d15cdd4061dbe2f445b5fff7a868e5ae67e..2311205d0ec0c2fd4642a4d8639c0bf871fe1d17 100644
--- a/common/rpc_service_test.go
+++ b/common/rpc_service_test.go
@@ -77,46 +77,48 @@ func TestServiceMap_Register(t *testing.T) {
// lowercase
s0 := &testService{}
// methods, err := ServiceMap.Register("testporotocol", s0)
- _, err := ServiceMap.Register("testporotocol", s0)
+ _, err := ServiceMap.Register("testService", "testporotocol", s0)
assert.EqualError(t, err, "type testService is not exported")
// succ
s := &TestService{}
- methods, err := ServiceMap.Register("testporotocol", s)
+ methods, err := ServiceMap.Register("testService", "testporotocol", s)
assert.NoError(t, err)
assert.Equal(t, "MethodOne,MethodThree,methodTwo", methods)
// repeat
- _, err = ServiceMap.Register("testporotocol", s)
+ _, err = ServiceMap.Register("testService", "testporotocol", s)
assert.EqualError(t, err, "service already defined: com.test.Path")
// no method
s1 := &TestService1{}
- _, err = ServiceMap.Register("testporotocol", s1)
+ _, err = ServiceMap.Register("testService", "testporotocol", s1)
assert.EqualError(t, err, "type com.test.Path1 has no exported methods of suitable type")
ServiceMap = &serviceMap{
- serviceMap: make(map[string]map[string]*Service),
+ serviceMap: make(map[string]map[string]*Service),
+ interfaceMap: make(map[string][]*Service),
}
}
func TestServiceMap_UnRegister(t *testing.T) {
s := &TestService{}
- _, err := ServiceMap.Register("testprotocol", s)
+ _, err := ServiceMap.Register("TestService", "testprotocol", s)
assert.NoError(t, err)
assert.NotNil(t, ServiceMap.GetService("testprotocol", "com.test.Path"))
+ assert.Equal(t, 1, len(ServiceMap.GetInterface("TestService")))
- err = ServiceMap.UnRegister("", "com.test.Path")
+ err = ServiceMap.UnRegister("", "", "com.test.Path")
assert.EqualError(t, err, "protocol or serviceName is nil")
- err = ServiceMap.UnRegister("protocol", "com.test.Path")
+ err = ServiceMap.UnRegister("", "protocol", "com.test.Path")
assert.EqualError(t, err, "no services for protocol")
- err = ServiceMap.UnRegister("testprotocol", "com.test.Path1")
+ err = ServiceMap.UnRegister("", "testprotocol", "com.test.Path1")
assert.EqualError(t, err, "no service for com.test.Path1")
// succ
- err = ServiceMap.UnRegister("testprotocol", "com.test.Path")
+ err = ServiceMap.UnRegister("TestService", "testprotocol", "com.test.Path")
assert.NoError(t, err)
}
diff --git a/config/config_loader.go b/config/config_loader.go
index 814c85b0a23d0ce79cf8c45bb6751467f3f5cac0..34763846060e1685687611432cb1a67cbd7b4834 100644
--- a/config/config_loader.go
+++ b/config/config_loader.go
@@ -200,7 +200,7 @@ func Load() {
svs.Implement(rpcService)
svs.Protocols = providerConfig.Protocols
if err := svs.Export(); err != nil {
- panic(fmt.Sprintf("service %s export failed! ", key))
+ panic(fmt.Sprintf("service %s export failed! err: %#v", key, err))
}
}
}
diff --git a/config/config_loader_test.go b/config/config_loader_test.go
index 3105dc2c09f156a7b09245fdf5a2f18e9d037206..c4cb3efe4b04dcfc7dad997d9535bd6d50289bd5 100644
--- a/config/config_loader_test.go
+++ b/config/config_loader_test.go
@@ -82,7 +82,8 @@ func TestLoad(t *testing.T) {
conServices = map[string]common.RPCService{}
proServices = map[string]common.RPCService{}
- common.ServiceMap.UnRegister("mock", "MockService")
+ err := common.ServiceMap.UnRegister("com.MockService", "mock", "MockService")
+ assert.Nil(t, err)
consumerConfig = nil
providerConfig = nil
}
@@ -110,7 +111,7 @@ func TestLoadWithSingleReg(t *testing.T) {
conServices = map[string]common.RPCService{}
proServices = map[string]common.RPCService{}
- common.ServiceMap.UnRegister("mock", "MockService")
+ common.ServiceMap.UnRegister("com.MockService", "mock", "MockService")
consumerConfig = nil
providerConfig = nil
}
@@ -139,7 +140,7 @@ func TestWithNoRegLoad(t *testing.T) {
conServices = map[string]common.RPCService{}
proServices = map[string]common.RPCService{}
- common.ServiceMap.UnRegister("mock", "MockService")
+ common.ServiceMap.UnRegister("com.MockService", "mock", "MockService")
consumerConfig = nil
providerConfig = nil
}
diff --git a/config/reference_config.go b/config/reference_config.go
index b9f3da13527c4c7de851dac9e5cd8ecf4638758b..3710cbc4bc62a01a014e91bcb978742c4a93c5cb 100644
--- a/config/reference_config.go
+++ b/config/reference_config.go
@@ -188,7 +188,7 @@ func (c *ReferenceConfig) getUrlMap() url.Values {
urlMap.Set(constant.VERSION_KEY, c.Version)
urlMap.Set(constant.GENERIC_KEY, strconv.FormatBool(c.Generic))
urlMap.Set(constant.ROLE_KEY, strconv.Itoa(common.CONSUMER))
- urlMap.Set(constant.CATEGORY_KEY, (common.RoleType(common.CONSUMER)).String())
+
urlMap.Set(constant.RELEASE_KEY, "dubbo-golang-"+constant.Version)
urlMap.Set(constant.SIDE_KEY, (common.RoleType(common.CONSUMER)).Role())
diff --git a/config/service_config.go b/config/service_config.go
index d5d087f4253f04d33e6b534571b019400409bcbb..048bedc6324a6b666b053188361ceb3eb3b72475 100644
--- a/config/service_config.go
+++ b/config/service_config.go
@@ -138,7 +138,7 @@ func (c *ServiceConfig) Export() error {
}
for _, proto := range protocolConfigs {
// registry the service reflect
- methods, err := common.ServiceMap.Register(proto.Name, c.rpcService)
+ methods, err := common.ServiceMap.Register(c.InterfaceName, proto.Name, c.rpcService)
if err != nil {
err := perrors.Errorf("The service %v export the protocol %v error! Error message is %v .", c.InterfaceName, proto.Name, err.Error())
logger.Errorf(err.Error())
@@ -225,7 +225,6 @@ func (c *ServiceConfig) getUrlMap() url.Values {
urlMap.Set(constant.GROUP_KEY, c.Group)
urlMap.Set(constant.VERSION_KEY, c.Version)
urlMap.Set(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))
- urlMap.Set(constant.CATEGORY_KEY, (common.RoleType(common.PROVIDER)).String())
urlMap.Set(constant.RELEASE_KEY, "dubbo-golang-"+constant.Version)
urlMap.Set(constant.SIDE_KEY, (common.RoleType(common.PROVIDER)).Role())
diff --git a/config_center/zookeeper/impl.go b/config_center/zookeeper/impl.go
index 937c56eef0b29f6fee52717f7a43c37ec35c7e95..0a1ce35306dab98363ca475cd5d1b0648e924b90 100644
--- a/config_center/zookeeper/impl.go
+++ b/config_center/zookeeper/impl.go
@@ -76,7 +76,7 @@ func newZookeeperDynamicConfiguration(url *common.URL) (*zookeeperDynamicConfigu
c.cacheListener = NewCacheListener(c.rootPath)
err = c.client.Create(c.rootPath)
- c.listener.ListenServiceEvent(c.rootPath, c.cacheListener)
+ c.listener.ListenServiceEvent(url, c.rootPath, c.cacheListener)
return c, err
}
@@ -102,7 +102,7 @@ func newMockZookeeperDynamicConfiguration(url *common.URL, opts ...zookeeper.Opt
c.cacheListener = NewCacheListener(c.rootPath)
err = c.client.Create(c.rootPath)
- go c.listener.ListenServiceEvent(c.rootPath, c.cacheListener)
+ go c.listener.ListenServiceEvent(url, c.rootPath, c.cacheListener)
return tc, c, err
}
diff --git a/doc/pic/arch/dubbo-go-arch.png b/doc/pic/arch/dubbo-go-arch.png
index 87726d88484c23d6395023bb10e86009d59a1fd7..e5f192715216257929cf4a550a1adf3588ab0e0f 100644
Binary files a/doc/pic/arch/dubbo-go-arch.png and b/doc/pic/arch/dubbo-go-arch.png differ
diff --git a/doc/pic/arch/dubbo-go-ext.png b/doc/pic/arch/dubbo-go-ext.png
new file mode 100644
index 0000000000000000000000000000000000000000..a5285c95570afa13212f7ee6eac5510f20243c3c
Binary files /dev/null and b/doc/pic/arch/dubbo-go-ext.png differ
diff --git a/filter/filter_impl/generic_service_filter_test.go b/filter/filter_impl/generic_service_filter_test.go
index 37c6af7450a75449fce51182684be2f619eda9d8..2a911659f068b53836b87af5caf5773d8ac5f119 100644
--- a/filter/filter_impl/generic_service_filter_test.go
+++ b/filter/filter_impl/generic_service_filter_test.go
@@ -96,7 +96,7 @@ func TestGenericServiceFilter_Invoke(t *testing.T) {
hessian.Object("222")},
}
s := &TestService{}
- _, _ = common.ServiceMap.Register("testprotocol", s)
+ _, _ = common.ServiceMap.Register("TestService", "testprotocol", s)
rpcInvocation := invocation.NewRPCInvocation(methodName, aurguments, nil)
filter := GetGenericServiceFilter()
url, _ := common.NewURL("testprotocol://127.0.0.1:20000/com.test.Path")
diff --git a/protocol/dubbo/client_test.go b/protocol/dubbo/client_test.go
index 1e0a73fac1a6cf6d4d102e5f4f6f1ba60fc4102a..744ffa80d6bc65e8526201b8cd327bb12b43caef 100644
--- a/protocol/dubbo/client_test.go
+++ b/protocol/dubbo/client_test.go
@@ -162,7 +162,7 @@ func InitTest(t *testing.T) (protocol.Protocol, common.URL) {
hessian.RegisterPOJO(&User{})
- methods, err := common.ServiceMap.Register("dubbo", &UserProvider{})
+ methods, err := common.ServiceMap.Register("com.ikurento.user.UserProvider", "dubbo", &UserProvider{})
assert.NoError(t, err)
assert.Equal(t, "GetBigPkg,GetUser,GetUser0,GetUser1,GetUser2,GetUser3,GetUser4,GetUser5,GetUser6", methods)
diff --git a/protocol/dubbo/dubbo_exporter.go b/protocol/dubbo/dubbo_exporter.go
index f4cd0cc1234f71bdcf6ce746f01ff3618d820fc5..1c45c40056f7690ba64838f641fe0b13a1554727 100644
--- a/protocol/dubbo/dubbo_exporter.go
+++ b/protocol/dubbo/dubbo_exporter.go
@@ -43,8 +43,9 @@ func NewDubboExporter(key string, invoker protocol.Invoker, exporterMap *sync.Ma
// Unexport ...
func (de *DubboExporter) Unexport() {
serviceId := de.GetInvoker().GetUrl().GetParam(constant.BEAN_NAME_KEY, "")
+ interfaceName := de.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
de.BaseExporter.Unexport()
- err := common.ServiceMap.UnRegister(DUBBO, serviceId)
+ err := common.ServiceMap.UnRegister(interfaceName, DUBBO, serviceId)
if err != nil {
logger.Errorf("[DubboExporter.Unexport] error: %v", err)
}
diff --git a/protocol/grpc/grpc_exporter.go b/protocol/grpc/grpc_exporter.go
index 1acd2fec391df2c104524792cf40d895cff3faac..0296ad8b5b06b693fd5e0972f18be77a69cd21ed 100644
--- a/protocol/grpc/grpc_exporter.go
+++ b/protocol/grpc/grpc_exporter.go
@@ -43,8 +43,9 @@ func NewGrpcExporter(key string, invoker protocol.Invoker, exporterMap *sync.Map
// Unexport ...
func (gg *GrpcExporter) Unexport() {
serviceId := gg.GetInvoker().GetUrl().GetParam(constant.BEAN_NAME_KEY, "")
+ interfaceName := gg.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
gg.BaseExporter.Unexport()
- err := common.ServiceMap.UnRegister(GRPC, serviceId)
+ err := common.ServiceMap.UnRegister(interfaceName, GRPC, serviceId)
if err != nil {
logger.Errorf("[GrpcExporter.Unexport] error: %v", err)
}
diff --git a/protocol/jsonrpc/http_test.go b/protocol/jsonrpc/http_test.go
index 0cb88b36a8f330059906eb70417b6d4841020c38..f8480bf32e15ad428209eedb72757a299455eb20 100644
--- a/protocol/jsonrpc/http_test.go
+++ b/protocol/jsonrpc/http_test.go
@@ -50,7 +50,7 @@ type (
func TestHTTPClient_Call(t *testing.T) {
- methods, err := common.ServiceMap.Register("jsonrpc", &UserProvider{})
+ methods, err := common.ServiceMap.Register("com.ikurento.user.UserProvider", "jsonrpc", &UserProvider{})
assert.NoError(t, err)
assert.Equal(t, "GetUser,GetUser0,GetUser1,GetUser2,GetUser3,GetUser4", methods)
diff --git a/protocol/jsonrpc/jsonrpc_exporter.go b/protocol/jsonrpc/jsonrpc_exporter.go
index 7f8fd491854f1ab25e63410a22ef5664db92f614..c61cf9adaebe9105a87ece1dcbae4dbe706cb3fc 100644
--- a/protocol/jsonrpc/jsonrpc_exporter.go
+++ b/protocol/jsonrpc/jsonrpc_exporter.go
@@ -43,8 +43,9 @@ func NewJsonrpcExporter(key string, invoker protocol.Invoker, exporterMap *sync.
// Unexport ...
func (je *JsonrpcExporter) Unexport() {
serviceId := je.GetInvoker().GetUrl().GetParam(constant.BEAN_NAME_KEY, "")
+ interfaceName := je.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
je.BaseExporter.Unexport()
- err := common.ServiceMap.UnRegister(JSONRPC, serviceId)
+ err := common.ServiceMap.UnRegister(interfaceName, JSONRPC, serviceId)
if err != nil {
logger.Errorf("[JsonrpcExporter.Unexport] error: %v", err)
}
diff --git a/protocol/jsonrpc/jsonrpc_invoker_test.go b/protocol/jsonrpc/jsonrpc_invoker_test.go
index 9e08eed2b4c61e686073a9039a605c4f73aa08c5..0f14ba11e2dec18bbd4d63e87e8c0fb2727f3755 100644
--- a/protocol/jsonrpc/jsonrpc_invoker_test.go
+++ b/protocol/jsonrpc/jsonrpc_invoker_test.go
@@ -36,7 +36,7 @@ import (
func TestJsonrpcInvoker_Invoke(t *testing.T) {
- methods, err := common.ServiceMap.Register("jsonrpc", &UserProvider{})
+ methods, err := common.ServiceMap.Register("UserProvider", "jsonrpc", &UserProvider{})
assert.NoError(t, err)
assert.Equal(t, "GetUser,GetUser0,GetUser1,GetUser2,GetUser3,GetUser4", methods)
diff --git a/protocol/rest/client/client_impl/resty_client.go b/protocol/rest/client/client_impl/resty_client.go
index aa6c23137dc68492948b85a85555a5340572ac49..b60f50a5a70cde01a051dbb2a4490cbb792e0116 100644
--- a/protocol/rest/client/client_impl/resty_client.go
+++ b/protocol/rest/client/client_impl/resty_client.go
@@ -40,10 +40,12 @@ func init() {
extension.SetRestClient(constant.DEFAULT_REST_CLIENT, NewRestyClient)
}
+// RestyClient a rest client implement by Resty
type RestyClient struct {
client *resty.Client
}
+// NewRestyClient a constructor of RestyClient
func NewRestyClient(restOption *client.RestOptions) client.RestClient {
client := resty.New()
client.SetTransport(
@@ -65,21 +67,21 @@ func NewRestyClient(restOption *client.RestOptions) client.RestClient {
}
}
-func (rc *RestyClient) Do(restRequest *client.RestRequest, res interface{}) error {
- r, err := rc.client.R().
- SetHeader("Content-Type", restRequest.Consumes).
- SetHeader("Accept", restRequest.Produces).
+// Do send request by RestyClient
+func (rc *RestyClient) Do(restRequest *client.RestClientRequest, res interface{}) error {
+ req := rc.client.R()
+ req.Header = restRequest.Header
+ resp, err := req.
SetPathParams(restRequest.PathParams).
SetQueryParams(restRequest.QueryParams).
- SetHeaders(restRequest.Headers).
SetBody(restRequest.Body).
SetResult(res).
Execute(restRequest.Method, "http://"+path.Join(restRequest.Location, restRequest.Path))
if err != nil {
return perrors.WithStack(err)
}
- if r.IsError() {
- return perrors.New(r.String())
+ if resp.IsError() {
+ return perrors.New(resp.String())
}
return nil
}
diff --git a/protocol/rest/client/rest_client.go b/protocol/rest/client/rest_client.go
index 7d020abc81c2bd44473ed25ffec4b2b657e7bcc0..d63c5e0bd0c7e6392eb5b0efc50bb0a585f4192d 100644
--- a/protocol/rest/client/rest_client.go
+++ b/protocol/rest/client/rest_client.go
@@ -18,26 +18,28 @@
package client
import (
+ "net/http"
"time"
)
+// RestOptions
type RestOptions struct {
RequestTimeout time.Duration
ConnectTimeout time.Duration
}
-type RestRequest struct {
+// RestClientRequest
+type RestClientRequest struct {
+ Header http.Header
Location string
Path string
- Produces string
- Consumes string
Method string
PathParams map[string]string
QueryParams map[string]string
Body interface{}
- Headers map[string]string
}
+// RestClient user can implement this client interface to send request
type RestClient interface {
- Do(request *RestRequest, res interface{}) error
+ Do(request *RestClientRequest, res interface{}) error
}
diff --git a/protocol/rest/rest_exporter.go b/protocol/rest/rest_exporter.go
index 470d525ad806687e7a732ce5681eb372eb431a63..1ee208615ea07e3f2850920492ab9e9821e7ffef 100644
--- a/protocol/rest/rest_exporter.go
+++ b/protocol/rest/rest_exporter.go
@@ -40,8 +40,9 @@ func NewRestExporter(key string, invoker protocol.Invoker, exporterMap *sync.Map
func (re *RestExporter) Unexport() {
serviceId := re.GetInvoker().GetUrl().GetParam(constant.BEAN_NAME_KEY, "")
+ interfaceName := re.GetInvoker().GetUrl().GetParam(constant.INTERFACE_KEY, "")
re.BaseExporter.Unexport()
- err := common.ServiceMap.UnRegister(REST, serviceId)
+ err := common.ServiceMap.UnRegister(interfaceName, REST, serviceId)
if err != nil {
logger.Errorf("[RestExporter.Unexport] error: %v", err)
}
diff --git a/protocol/rest/rest_invoker.go b/protocol/rest/rest_invoker.go
index 0c82035ac5eb9a52ab188baa971dbdf1b864e970..121d1217efd3baea0f961b67e243e9a0450aefc2 100644
--- a/protocol/rest/rest_invoker.go
+++ b/protocol/rest/rest_invoker.go
@@ -20,6 +20,7 @@ package rest
import (
"context"
"fmt"
+ "net/http"
)
import (
@@ -56,7 +57,7 @@ func (ri *RestInvoker) Invoke(ctx context.Context, invocation protocol.Invocatio
body interface{}
pathParams map[string]string
queryParams map[string]string
- headers map[string]string
+ header http.Header
err error
)
if methodConfig == nil {
@@ -71,24 +72,21 @@ func (ri *RestInvoker) Invoke(ctx context.Context, invocation protocol.Invocatio
result.Err = err
return &result
}
- if headers, err = restStringMapTransform(methodConfig.HeadersMap, inv.Arguments()); err != nil {
+ if header, err = getRestHttpHeader(methodConfig, inv.Arguments()); err != nil {
result.Err = err
return &result
}
if len(inv.Arguments()) > methodConfig.Body && methodConfig.Body >= 0 {
body = inv.Arguments()[methodConfig.Body]
}
-
- req := &client.RestRequest{
+ req := &client.RestClientRequest{
Location: ri.GetUrl().Location,
- Produces: methodConfig.Produces,
- Consumes: methodConfig.Consumes,
Method: methodConfig.MethodType,
Path: methodConfig.Path,
PathParams: pathParams,
QueryParams: queryParams,
Body: body,
- Headers: headers,
+ Header: header,
}
result.Err = ri.client.Do(req, inv.Reply())
if result.Err == nil {
@@ -107,3 +105,17 @@ func restStringMapTransform(paramsMap map[int]string, args []interface{}) (map[s
}
return resMap, nil
}
+
+func getRestHttpHeader(methodConfig *config.RestMethodConfig, args []interface{}) (http.Header, error) {
+ header := http.Header{}
+ headersMap := methodConfig.HeadersMap
+ header.Set("Content-Type", methodConfig.Consumes)
+ header.Set("Accept", methodConfig.Produces)
+ for k, v := range headersMap {
+ if k >= len(args) || k < 0 {
+ return nil, perrors.Errorf("[Rest Invoke] Index %v is out of bundle", k)
+ }
+ header.Set(v, fmt.Sprint(args[k]))
+ }
+ return header, nil
+}
diff --git a/protocol/rest/rest_invoker_test.go b/protocol/rest/rest_invoker_test.go
index e44c5d9a21026992178bd432676c99bc837c361b..2ea260c58d03c27a691e48b953ce6d64f75040a2 100644
--- a/protocol/rest/rest_invoker_test.go
+++ b/protocol/rest/rest_invoker_test.go
@@ -61,7 +61,7 @@ func TestRestInvoker_Invoke(t *testing.T) {
"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)
- _, err = common.ServiceMap.Register(url.Protocol, &UserProvider{})
+ _, err = common.ServiceMap.Register("UserProvider", url.Protocol, &UserProvider{})
assert.NoError(t, err)
con := config.ProviderConfig{}
config.SetProviderConfig(con)
@@ -206,6 +206,6 @@ func TestRestInvoker_Invoke(t *testing.T) {
assert.Error(t, res.Error(), "test error")
assert.Equal(t, filterNum, 12)
- err = common.ServiceMap.UnRegister(url.Protocol, "com.ikurento.user.UserProvider")
+ err = common.ServiceMap.UnRegister("UserProvider", url.Protocol, "com.ikurento.user.UserProvider")
assert.NoError(t, err)
}
diff --git a/protocol/rest/rest_protocol.go b/protocol/rest/rest_protocol.go
index 47ecb6093b4cfa12a1d3397fa45d59b1e173a93a..e15eeb39d72212eb9f1d0235313eba231d3b0a36 100644
--- a/protocol/rest/rest_protocol.go
+++ b/protocol/rest/rest_protocol.go
@@ -75,7 +75,9 @@ func (rp *RestProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
}
rp.SetExporterMap(serviceKey, exporter)
restServer := rp.getServer(url, restServiceConfig.Server)
- restServer.Deploy(invoker, restServiceConfig.RestMethodConfigsMap)
+ for _, methodConfig := range restServiceConfig.RestMethodConfigsMap {
+ restServer.Deploy(methodConfig, server.GetRouteFunc(invoker, methodConfig))
+ }
return exporter
}
diff --git a/protocol/rest/rest_protocol_test.go b/protocol/rest/rest_protocol_test.go
index 8af73a1839c159fdf58c64d12e039c20bb3221c6..9117148777ca868cb7d2672236e800c836d3de84 100644
--- a/protocol/rest/rest_protocol_test.go
+++ b/protocol/rest/rest_protocol_test.go
@@ -80,7 +80,7 @@ func TestRestProtocol_Export(t *testing.T) {
"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)
- _, err = common.ServiceMap.Register(url.Protocol, &UserProvider{})
+ _, err = common.ServiceMap.Register("UserProvider", url.Protocol, &UserProvider{})
assert.NoError(t, err)
con := config.ProviderConfig{}
config.SetProviderConfig(con)
@@ -128,7 +128,7 @@ func TestRestProtocol_Export(t *testing.T) {
proto.Destroy()
_, ok = proto.(*RestProtocol).serverMap[url.Location]
assert.False(t, ok)
- err = common.ServiceMap.UnRegister(url.Protocol, "com.ikurento.user.UserProvider")
+ err = common.ServiceMap.UnRegister("UserProvider", url.Protocol, "com.ikurento.user.UserProvider")
assert.NoError(t, err)
}
diff --git a/protocol/rest/server/rest_server.go b/protocol/rest/server/rest_server.go
index c10c98a7b677d47c43b64643a69d5b3768a6c663..fbd6fb7ad9dd81f043e4d45ee94a54e12ef89cdd 100644
--- a/protocol/rest/server/rest_server.go
+++ b/protocol/rest/server/rest_server.go
@@ -17,15 +17,306 @@
package server
+import (
+ "context"
+ "errors"
+ "net/http"
+ "reflect"
+ "strconv"
+ "strings"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+)
+
import (
"github.com/apache/dubbo-go/common"
+ "github.com/apache/dubbo-go/common/logger"
"github.com/apache/dubbo-go/protocol"
- "github.com/apache/dubbo-go/protocol/rest/config"
+ "github.com/apache/dubbo-go/protocol/invocation"
+ rest_config "github.com/apache/dubbo-go/protocol/rest/config"
)
+const parseParameterErrorStr = "An error occurred while parsing parameters on the server"
+
+// RestServer user can implement this server interface
type RestServer interface {
+ // Start rest server
Start(url common.URL)
- Deploy(invoker protocol.Invoker, restMethodConfig map[string]*config.RestMethodConfig)
- UnDeploy(restMethodConfig map[string]*config.RestMethodConfig)
+ // Deploy a http api
+ Deploy(restMethodConfig *rest_config.RestMethodConfig, routeFunc func(request RestServerRequest, response RestServerResponse))
+ // UnDeploy a http api
+ UnDeploy(restMethodConfig *rest_config.RestMethodConfig)
+ // Destroy rest server
Destroy()
}
+
+// RestServerRequest interface
+type RestServerRequest interface {
+ // RawRequest get the Ptr of http.Request
+ RawRequest() *http.Request
+ // PathParameter get the path parameter by name
+ PathParameter(name string) string
+ // PathParameters get the map of the path parameters
+ PathParameters() map[string]string
+ // QueryParameter get the query parameter by name
+ QueryParameter(name string) string
+ // QueryParameters get the map of query parameters
+ QueryParameters(name string) []string
+ // BodyParameter get the body parameter of name
+ BodyParameter(name string) (string, error)
+ // HeaderParameter get the header parameter of name
+ HeaderParameter(name string) string
+ // ReadEntity checks the Accept header and reads the content into the entityPointer.
+ ReadEntity(entityPointer interface{}) error
+}
+
+// RestServerResponse interface
+type RestServerResponse interface {
+ http.ResponseWriter
+ // WriteError writes the http status and the error string on the response. err can be nil.
+ // Return an error if writing was not successful.
+ WriteError(httpStatus int, err error) (writeErr error)
+ // WriteEntity marshals the value using the representation denoted by the Accept Header.
+ WriteEntity(value interface{}) error
+}
+
+// GetRouteFunc
+// A route function will be invoked by http server
+func GetRouteFunc(invoker protocol.Invoker, methodConfig *rest_config.RestMethodConfig) func(req RestServerRequest, resp RestServerResponse) {
+ return func(req RestServerRequest, resp RestServerResponse) {
+ var (
+ err error
+ args []interface{}
+ )
+ svc := common.ServiceMap.GetService(invoker.GetUrl().Protocol, strings.TrimPrefix(invoker.GetUrl().Path, "/"))
+ // get method
+ method := svc.Method()[methodConfig.MethodName]
+ argsTypes := method.ArgsType()
+ replyType := method.ReplyType()
+ // two ways to prepare arguments
+ // if method like this 'func1(req []interface{}, rsp *User) error'
+ // we don't have arguments type
+ if (len(argsTypes) == 1 || len(argsTypes) == 2 && replyType == nil) &&
+ argsTypes[0].String() == "[]interface {}" {
+ args, err = getArgsInterfaceFromRequest(req, methodConfig)
+ } else {
+ args, err = getArgsFromRequest(req, argsTypes, methodConfig)
+ }
+ if err != nil {
+ logger.Errorf("[Go Restful] parsing http parameters error:%v", err)
+ err = resp.WriteError(http.StatusInternalServerError, errors.New(parseParameterErrorStr))
+ if err != nil {
+ logger.Errorf("[Go Restful] WriteErrorString error:%v", err)
+ }
+ }
+ result := invoker.Invoke(context.Background(), invocation.NewRPCInvocation(methodConfig.MethodName, args, make(map[string]string)))
+ if result.Error() != nil {
+ err = resp.WriteError(http.StatusInternalServerError, result.Error())
+ if err != nil {
+ logger.Errorf("[Go Restful] WriteError error:%v", err)
+ }
+ return
+ }
+ err = resp.WriteEntity(result.Result())
+ if err != nil {
+ logger.Errorf("[Go Restful] WriteEntity error:%v", err)
+ }
+ }
+}
+
+// getArgsInterfaceFromRequest when service function like GetUser(req []interface{}, rsp *User) error
+// use this method to get arguments
+func getArgsInterfaceFromRequest(req RestServerRequest, methodConfig *rest_config.RestMethodConfig) ([]interface{}, error) {
+ argsMap := make(map[int]interface{}, 8)
+ maxKey := 0
+ for k, v := range methodConfig.PathParamsMap {
+ if maxKey < k {
+ maxKey = k
+ }
+ argsMap[k] = req.PathParameter(v)
+ }
+ for k, v := range methodConfig.QueryParamsMap {
+ if maxKey < k {
+ maxKey = k
+ }
+ params := req.QueryParameters(v)
+ if len(params) == 1 {
+ argsMap[k] = params[0]
+ } else {
+ argsMap[k] = params
+ }
+ }
+ for k, v := range methodConfig.HeadersMap {
+ if maxKey < k {
+ maxKey = k
+ }
+ argsMap[k] = req.HeaderParameter(v)
+ }
+ if methodConfig.Body >= 0 {
+ if maxKey < methodConfig.Body {
+ maxKey = methodConfig.Body
+ }
+ m := make(map[string]interface{})
+ // TODO read as a slice
+ if err := req.ReadEntity(&m); err != nil {
+ return nil, perrors.Errorf("[Go restful] Read body entity as map[string]interface{} error:%v", err)
+ }
+ argsMap[methodConfig.Body] = m
+ }
+ args := make([]interface{}, maxKey+1)
+ for k, v := range argsMap {
+ if k >= 0 {
+ args[k] = v
+ }
+ }
+ return args, nil
+}
+
+// getArgsFromRequest get arguments from server.RestServerRequest
+func getArgsFromRequest(req RestServerRequest, argsTypes []reflect.Type, methodConfig *rest_config.RestMethodConfig) ([]interface{}, error) {
+ argsLength := len(argsTypes)
+ args := make([]interface{}, argsLength)
+ for i, t := range argsTypes {
+ args[i] = reflect.Zero(t).Interface()
+ }
+ if err := assembleArgsFromPathParams(methodConfig, argsLength, argsTypes, req, args); err != nil {
+ return nil, err
+ }
+ if err := assembleArgsFromQueryParams(methodConfig, argsLength, argsTypes, req, args); err != nil {
+ return nil, err
+ }
+ if err := assembleArgsFromBody(methodConfig, argsTypes, req, args); err != nil {
+ return nil, err
+ }
+ if err := assembleArgsFromHeaders(methodConfig, req, argsLength, argsTypes, args); err != nil {
+ return nil, err
+ }
+ return args, nil
+}
+
+// assembleArgsFromHeaders assemble arguments from headers
+func assembleArgsFromHeaders(methodConfig *rest_config.RestMethodConfig, req RestServerRequest, argsLength int, argsTypes []reflect.Type, args []interface{}) error {
+ for k, v := range methodConfig.HeadersMap {
+ param := req.HeaderParameter(v)
+ if k < 0 || k >= argsLength {
+ return perrors.Errorf("[Go restful] Header param parse error, the index %v args of method:%v doesn't exist", k, methodConfig.MethodName)
+ }
+ t := argsTypes[k]
+ if t.Kind() == reflect.Ptr {
+ t = t.Elem()
+ }
+ if t.Kind() == reflect.String {
+ args[k] = param
+ } else {
+ return perrors.Errorf("[Go restful] Header param parse error, the index %v args's type isn't string", k)
+ }
+ }
+ return nil
+}
+
+// assembleArgsFromBody assemble arguments from body
+func assembleArgsFromBody(methodConfig *rest_config.RestMethodConfig, argsTypes []reflect.Type, req RestServerRequest, args []interface{}) error {
+ if methodConfig.Body >= 0 && methodConfig.Body < len(argsTypes) {
+ t := argsTypes[methodConfig.Body]
+ kind := t.Kind()
+ if kind == reflect.Ptr {
+ t = t.Elem()
+ }
+ var ni interface{}
+ if t.String() == "[]interface {}" {
+ ni = make([]map[string]interface{}, 0)
+ } else if t.String() == "interface {}" {
+ ni = make(map[string]interface{})
+ } else {
+ n := reflect.New(t)
+ if n.CanInterface() {
+ ni = n.Interface()
+ }
+ }
+ if err := req.ReadEntity(&ni); err != nil {
+ return perrors.Errorf("[Go restful] Read body entity error, error is %v", perrors.WithStack(err))
+ }
+ args[methodConfig.Body] = ni
+ }
+ return nil
+}
+
+// assembleArgsFromQueryParams assemble arguments from query params
+func assembleArgsFromQueryParams(methodConfig *rest_config.RestMethodConfig, argsLength int, argsTypes []reflect.Type, req RestServerRequest, args []interface{}) error {
+ var (
+ err error
+ param interface{}
+ i64 int64
+ )
+ for k, v := range methodConfig.QueryParamsMap {
+ if k < 0 || k >= argsLength {
+ return perrors.Errorf("[Go restful] Query param parse error, the index %v args of method:%v doesn't exist", k, methodConfig.MethodName)
+ }
+ t := argsTypes[k]
+ kind := t.Kind()
+ if kind == reflect.Ptr {
+ t = t.Elem()
+ }
+ if kind == reflect.Slice {
+ param = req.QueryParameters(v)
+ } else if kind == reflect.String {
+ param = req.QueryParameter(v)
+ } else if kind == reflect.Int {
+ param, err = strconv.Atoi(req.QueryParameter(v))
+ } else if kind == reflect.Int32 {
+ i64, err = strconv.ParseInt(req.QueryParameter(v), 10, 32)
+ if err == nil {
+ param = int32(i64)
+ }
+ } else if kind == reflect.Int64 {
+ param, err = strconv.ParseInt(req.QueryParameter(v), 10, 64)
+ } else {
+ return perrors.Errorf("[Go restful] Query param parse error, the index %v args's type isn't int or string or slice", k)
+ }
+ if err != nil {
+ return perrors.Errorf("[Go restful] Query param parse error, error:%v", perrors.WithStack(err))
+ }
+ args[k] = param
+ }
+ return nil
+}
+
+// assembleArgsFromPathParams assemble arguments from path params
+func assembleArgsFromPathParams(methodConfig *rest_config.RestMethodConfig, argsLength int, argsTypes []reflect.Type, req RestServerRequest, args []interface{}) error {
+ var (
+ err error
+ param interface{}
+ i64 int64
+ )
+ for k, v := range methodConfig.PathParamsMap {
+ if k < 0 || k >= argsLength {
+ return perrors.Errorf("[Go restful] Path param parse error, the index %v args of method:%v doesn't exist", k, methodConfig.MethodName)
+ }
+ t := argsTypes[k]
+ kind := t.Kind()
+ if kind == reflect.Ptr {
+ t = t.Elem()
+ }
+ if kind == reflect.Int {
+ param, err = strconv.Atoi(req.PathParameter(v))
+ } else if kind == reflect.Int32 {
+ i64, err = strconv.ParseInt(req.PathParameter(v), 10, 32)
+ if err == nil {
+ param = int32(i64)
+ }
+ } else if kind == reflect.Int64 {
+ param, err = strconv.ParseInt(req.PathParameter(v), 10, 64)
+ } else if kind == reflect.String {
+ param = req.PathParameter(v)
+ } else {
+ return perrors.Errorf("[Go restful] Path param parse error, the index %v args's type isn't int or string", k)
+ }
+ if err != nil {
+ return perrors.Errorf("[Go restful] Path param parse error, error is %v", perrors.WithStack(err))
+ }
+ args[k] = param
+ }
+ return nil
+}
diff --git a/protocol/rest/server/server_impl/go_restful_server.go b/protocol/rest/server/server_impl/go_restful_server.go
index 69f36a5c80aa51f52dfcfabc5a1bd4003f4cd727..c7d971fcaa5ada0ba02cc436b5ae6705793887ef 100644
--- a/protocol/rest/server/server_impl/go_restful_server.go
+++ b/protocol/rest/server/server_impl/go_restful_server.go
@@ -22,8 +22,6 @@ import (
"fmt"
"net"
"net/http"
- "reflect"
- "strconv"
"strings"
"time"
)
@@ -38,27 +36,29 @@ import (
"github.com/apache/dubbo-go/common/constant"
"github.com/apache/dubbo-go/common/extension"
"github.com/apache/dubbo-go/common/logger"
- "github.com/apache/dubbo-go/protocol"
- "github.com/apache/dubbo-go/protocol/invocation"
"github.com/apache/dubbo-go/protocol/rest/config"
"github.com/apache/dubbo-go/protocol/rest/server"
)
func init() {
- extension.SetRestServer(constant.DEFAULT_REST_SERVER, GetNewGoRestfulServer)
+ extension.SetRestServer(constant.DEFAULT_REST_SERVER, NewGoRestfulServer)
}
var filterSlice []restful.FilterFunction
+// GoRestfulServer a rest server implement by go-restful
type GoRestfulServer struct {
srv *http.Server
container *restful.Container
}
-func NewGoRestfulServer() *GoRestfulServer {
+// NewGoRestfulServer a constructor of GoRestfulServer
+func NewGoRestfulServer() server.RestServer {
return &GoRestfulServer{}
}
+// Start go-restful server
+// It will add all go-restful filters
func (grs *GoRestfulServer) Start(url common.URL) {
grs.container = restful.NewContainer()
for _, filter := range filterSlice {
@@ -80,61 +80,32 @@ func (grs *GoRestfulServer) Start(url common.URL) {
}()
}
-func (grs *GoRestfulServer) Deploy(invoker protocol.Invoker, restMethodConfig map[string]*config.RestMethodConfig) {
- svc := common.ServiceMap.GetService(invoker.GetUrl().Protocol, strings.TrimPrefix(invoker.GetUrl().Path, "/"))
- for methodName, config := range restMethodConfig {
- // get method
- method := svc.Method()[methodName]
- argsTypes := method.ArgsType()
- replyType := method.ReplyType()
- ws := new(restful.WebService)
- ws.Path(config.Path).
- Produces(strings.Split(config.Produces, ",")...).
- Consumes(strings.Split(config.Consumes, ",")...).
- Route(ws.Method(config.MethodType).To(getFunc(methodName, invoker, argsTypes, replyType, config)))
- grs.container.Add(ws)
+// Publish a http api in go-restful server
+// The routeFunc should be invoked when the server receive a request
+func (grs *GoRestfulServer) Deploy(restMethodConfig *config.RestMethodConfig, routeFunc func(request server.RestServerRequest, response server.RestServerResponse)) {
+ ws := &restful.WebService{}
+ rf := func(req *restful.Request, resp *restful.Response) {
+ routeFunc(NewGoRestfulRequestAdapter(req), resp)
}
+ ws.Path(restMethodConfig.Path).
+ Produces(strings.Split(restMethodConfig.Produces, ",")...).
+ Consumes(strings.Split(restMethodConfig.Consumes, ",")...).
+ Route(ws.Method(restMethodConfig.MethodType).To(rf))
+ grs.container.Add(ws)
}
-func getFunc(methodName string, invoker protocol.Invoker, argsTypes []reflect.Type,
- replyType reflect.Type, config *config.RestMethodConfig) func(req *restful.Request, resp *restful.Response) {
- return func(req *restful.Request, resp *restful.Response) {
- var (
- err error
- args []interface{}
- )
- if (len(argsTypes) == 1 || len(argsTypes) == 2 && replyType == nil) &&
- argsTypes[0].String() == "[]interface {}" {
- args = getArgsInterfaceFromRequest(req, config)
- } else {
- args = getArgsFromRequest(req, argsTypes, config)
- }
- result := invoker.Invoke(context.Background(), invocation.NewRPCInvocation(methodName, args, make(map[string]string)))
- if result.Error() != nil {
- err = resp.WriteError(http.StatusInternalServerError, result.Error())
- if err != nil {
- logger.Errorf("[Go Restful] WriteError error:%v", err)
- }
- return
- }
- err = resp.WriteEntity(result.Result())
- if err != nil {
- logger.Error("[Go Restful] WriteEntity error:%v", err)
- }
- }
-}
-func (grs *GoRestfulServer) UnDeploy(restMethodConfig map[string]*config.RestMethodConfig) {
- for _, config := range restMethodConfig {
- ws := new(restful.WebService)
- ws.Path(config.Path)
- err := grs.container.Remove(ws)
- if err != nil {
- logger.Warnf("[Go restful] Remove web service error:%v", err)
- }
+// Delete a http api in go-restful server
+func (grs *GoRestfulServer) UnDeploy(restMethodConfig *config.RestMethodConfig) {
+ ws := new(restful.WebService)
+ ws.Path(restMethodConfig.Path)
+ err := grs.container.Remove(ws)
+ if err != nil {
+ logger.Warnf("[Go restful] Remove web service error:%v", err)
}
}
+// Destroy the go-restful server
func (grs *GoRestfulServer) Destroy() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -144,179 +115,59 @@ func (grs *GoRestfulServer) Destroy() {
logger.Infof("[Go Restful] Server exiting")
}
-func getArgsInterfaceFromRequest(req *restful.Request, config *config.RestMethodConfig) []interface{} {
- argsMap := make(map[int]interface{}, 8)
- maxKey := 0
- for k, v := range config.PathParamsMap {
- if maxKey < k {
- maxKey = k
- }
- argsMap[k] = req.PathParameter(v)
- }
- for k, v := range config.QueryParamsMap {
- if maxKey < k {
- maxKey = k
- }
- params := req.QueryParameters(v)
- if len(params) == 1 {
- argsMap[k] = params[0]
- } else {
- argsMap[k] = params
- }
- }
- for k, v := range config.HeadersMap {
- if maxKey < k {
- maxKey = k
- }
- argsMap[k] = req.HeaderParameter(v)
- }
- if config.Body >= 0 {
- if maxKey < config.Body {
- maxKey = config.Body
- }
- m := make(map[string]interface{})
- // TODO read as a slice
- if err := req.ReadEntity(&m); err != nil {
- logger.Warnf("[Go restful] Read body entity as map[string]interface{} error:%v", perrors.WithStack(err))
- } else {
- argsMap[config.Body] = m
- }
- }
- args := make([]interface{}, maxKey+1)
- for k, v := range argsMap {
- if k >= 0 {
- args[k] = v
- }
- }
- return args
+// AddGoRestfulServerFilter let user add the http server of go-restful
+// addFilter should before config.Load()
+func AddGoRestfulServerFilter(filterFuc restful.FilterFunction) {
+ filterSlice = append(filterSlice, filterFuc)
}
-func getArgsFromRequest(req *restful.Request, argsTypes []reflect.Type, config *config.RestMethodConfig) []interface{} {
- argsLength := len(argsTypes)
- args := make([]interface{}, argsLength)
- for i, t := range argsTypes {
- args[i] = reflect.Zero(t).Interface()
- }
- var (
- err error
- param interface{}
- i64 int64
- )
- for k, v := range config.PathParamsMap {
- if k < 0 || k >= argsLength {
- logger.Errorf("[Go restful] Path param parse error, the args:%v doesn't exist", k)
- continue
- }
- t := argsTypes[k]
- kind := t.Kind()
- if kind == reflect.Ptr {
- t = t.Elem()
- }
- if kind == reflect.Int {
- param, err = strconv.Atoi(req.PathParameter(v))
- } else if kind == reflect.Int32 {
- i64, err = strconv.ParseInt(req.PathParameter(v), 10, 32)
- if err == nil {
- param = int32(i64)
- }
- } else if kind == reflect.Int64 {
- param, err = strconv.ParseInt(req.PathParameter(v), 10, 64)
- } else if kind == reflect.String {
- param = req.PathParameter(v)
- } else {
- logger.Warnf("[Go restful] Path param parse error, the args:%v of type isn't int or string", k)
- continue
- }
- if err != nil {
- logger.Errorf("[Go restful] Path param parse error, error is %v", err)
- continue
- }
- args[k] = param
- }
- for k, v := range config.QueryParamsMap {
- if k < 0 || k >= argsLength {
- logger.Errorf("[Go restful] Query param parse error, the args:%v doesn't exist", k)
- continue
- }
- t := argsTypes[k]
- kind := t.Kind()
- if kind == reflect.Ptr {
- t = t.Elem()
- }
- if kind == reflect.Slice {
- param = req.QueryParameters(v)
- } else if kind == reflect.String {
- param = req.QueryParameter(v)
- } else if kind == reflect.Int {
- param, err = strconv.Atoi(req.QueryParameter(v))
- } else if kind == reflect.Int32 {
- i64, err = strconv.ParseInt(req.QueryParameter(v), 10, 32)
- if err == nil {
- param = int32(i64)
- }
- } else if kind == reflect.Int64 {
- param, err = strconv.ParseInt(req.QueryParameter(v), 10, 64)
- } else {
- logger.Errorf("[Go restful] Query param parse error, the args:%v of type isn't int or string or slice", k)
- continue
- }
- if err != nil {
- logger.Errorf("[Go restful] Query param parse error, error is %v", err)
- continue
- }
- args[k] = param
- }
+// GoRestfulRequestAdapter a adapter struct about RestServerRequest
+type GoRestfulRequestAdapter struct {
+ server.RestServerRequest
+ request *restful.Request
+}
- if config.Body >= 0 && config.Body < len(argsTypes) {
- t := argsTypes[config.Body]
- kind := t.Kind()
- if kind == reflect.Ptr {
- t = t.Elem()
- }
- var ni interface{}
- if t.String() == "[]interface {}" {
- ni = make([]map[string]interface{}, 0)
- } else if t.String() == "interface {}" {
- ni = make(map[string]interface{})
- } else {
- n := reflect.New(t)
- if n.CanInterface() {
- ni = n.Interface()
- }
- }
- if err := req.ReadEntity(&ni); err != nil {
- logger.Errorf("[Go restful] Read body entity error:%v", err)
- } else {
- args[config.Body] = ni
- }
- }
+// NewGoRestfulRequestAdapter a constructor of GoRestfulRequestAdapter
+func NewGoRestfulRequestAdapter(request *restful.Request) *GoRestfulRequestAdapter {
+ return &GoRestfulRequestAdapter{request: request}
+}
- for k, v := range config.HeadersMap {
- param := req.HeaderParameter(v)
- if k < 0 || k >= argsLength {
- logger.Errorf("[Go restful] Header param parse error, the args:%v doesn't exist", k)
- continue
- }
- t := argsTypes[k]
- if t.Kind() == reflect.Ptr {
- t = t.Elem()
- }
- if t.Kind() == reflect.String {
- args[k] = param
- } else {
- logger.Errorf("[Go restful] Header param parse error, the args:%v of type isn't string", k)
- }
- }
+// RawRequest a adapter function of server.RestServerRequest's RawRequest
+func (grra *GoRestfulRequestAdapter) RawRequest() *http.Request {
+ return grra.request.Request
+}
- return args
+// PathParameter a adapter function of server.RestServerRequest's PathParameter
+func (grra *GoRestfulRequestAdapter) PathParameter(name string) string {
+ return grra.request.PathParameter(name)
}
-func GetNewGoRestfulServer() server.RestServer {
- return NewGoRestfulServer()
+// PathParameters a adapter function of server.RestServerRequest's QueryParameter
+func (grra *GoRestfulRequestAdapter) PathParameters() map[string]string {
+ return grra.request.PathParameters()
}
-// Let user addFilter
-// addFilter should before config.Load()
-func AddGoRestfulServerFilter(filterFuc restful.FilterFunction) {
- filterSlice = append(filterSlice, filterFuc)
+// QueryParameter a adapter function of server.RestServerRequest's QueryParameters
+func (grra *GoRestfulRequestAdapter) QueryParameter(name string) string {
+ return grra.request.QueryParameter(name)
+}
+
+// QueryParameters a adapter function of server.RestServerRequest's QueryParameters
+func (grra *GoRestfulRequestAdapter) QueryParameters(name string) []string {
+ return grra.request.QueryParameters(name)
+}
+
+// BodyParameter a adapter function of server.RestServerRequest's BodyParameter
+func (grra *GoRestfulRequestAdapter) BodyParameter(name string) (string, error) {
+ return grra.request.BodyParameter(name)
+}
+
+// HeaderParameter a adapter function of server.RestServerRequest's HeaderParameter
+func (grra *GoRestfulRequestAdapter) HeaderParameter(name string) string {
+ return grra.request.HeaderParameter(name)
+}
+
+// ReadEntity a adapter func of server.RestServerRequest's ReadEntity
+func (grra *GoRestfulRequestAdapter) ReadEntity(entityPointer interface{}) error {
+ return grra.request.ReadEntity(entityPointer)
}
diff --git a/registry/base_registry.go b/registry/base_registry.go
index 04694da25a38899cfc6f56a8092a15547f142e9c..3e1bddf233310871182544b6415c10c8df27e622 100644
--- a/registry/base_registry.go
+++ b/registry/base_registry.go
@@ -121,6 +121,7 @@ func (r *BaseRegistry) Destroy() {
close(r.done)
// wait waitgroup done (wait listeners outside close over)
r.wg.Wait()
+
//close registry client
r.closeRegisters()
}
@@ -178,7 +179,10 @@ func (r *BaseRegistry) RestartCallBack() bool {
}
logger.Infof("success to re-register service :%v", confIf.Key())
}
- r.facadeBasedRegistry.InitListeners()
+
+ if flag {
+ r.facadeBasedRegistry.InitListeners()
+ }
return flag
}
diff --git a/registry/etcdv3/registry.go b/registry/etcdv3/registry.go
index e1c25768119ea7d7122b9aa22a5f881db44bafd9..5d389c36374fe9de5561418bc90d44a7d780fd48 100644
--- a/registry/etcdv3/registry.go
+++ b/registry/etcdv3/registry.go
@@ -164,9 +164,7 @@ func (r *etcdV3Registry) DoSubscribe(svc *common.URL) (registry.Listener, error)
//register the svc to dataListener
r.dataListener.AddInterestedURL(svc)
- for _, v := range strings.Split(svc.GetParam(constant.CATEGORY_KEY, constant.DEFAULT_CATEGORY), ",") {
- go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+v, svc.Service()), r.dataListener)
- }
+ go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, svc.Service()), r.dataListener)
return configListener, nil
}
diff --git a/registry/kubernetes/registry.go b/registry/kubernetes/registry.go
index 7212a83d63b34cc08f193994f004efccf9e21a0c..8a02d0e3e693b58946a97e7b47238e0be4272dcf 100644
--- a/registry/kubernetes/registry.go
+++ b/registry/kubernetes/registry.go
@@ -21,7 +21,6 @@ import (
"fmt"
"os"
"path"
- "strings"
"sync"
"time"
)
@@ -135,9 +134,7 @@ func (r *kubernetesRegistry) DoSubscribe(svc *common.URL) (registry.Listener, er
//register the svc to dataListener
r.dataListener.AddInterestedURL(svc)
- for _, v := range strings.Split(svc.GetParam(constant.CATEGORY_KEY, constant.DEFAULT_CATEGORY), ",") {
- go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+v, svc.Service()), r.dataListener)
- }
+ go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, svc.Service()), r.dataListener)
return configListener, nil
}
diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go
index a7678ba4e2f38cfeb77f202103e03066a7efdbef..b47b9f372da8948d1add97c7b354503e18e23511 100644
--- a/registry/protocol/protocol.go
+++ b/registry/protocol/protocol.go
@@ -45,6 +45,7 @@ import (
var (
regProtocol *registryProtocol
+ once sync.Once
)
type registryProtocol struct {
@@ -346,12 +347,12 @@ func setProviderUrl(regURL *common.URL, providerURL *common.URL) {
regURL.SubURL = providerURL
}
-// GetProtocol ...
+// GetProtocol return the singleton RegistryProtocol
func GetProtocol() protocol.Protocol {
- if regProtocol != nil {
- return regProtocol
- }
- return newRegistryProtocol()
+ once.Do(func() {
+ regProtocol = newRegistryProtocol()
+ })
+ return regProtocol
}
type wrappedInvoker struct {
diff --git a/registry/zookeeper/listener.go b/registry/zookeeper/listener.go
index bef1760e04fd6597721fd19b5d19820f45ed2bf0..c5b2f33c6107e82aa172c818c0d8aca1483248c6 100644
--- a/registry/zookeeper/listener.go
+++ b/registry/zookeeper/listener.go
@@ -35,23 +35,28 @@ import (
zk "github.com/apache/dubbo-go/remoting/zookeeper"
)
-// RegistryDataListener ...
+// RegistryDataListener contains all URL information subscribed by zookeeper registry
type RegistryDataListener struct {
- interestedURL []*common.URL
- listener config_center.ConfigurationListener
+ subscribed map[*common.URL]config_center.ConfigurationListener
+ mutex sync.Mutex
+ closed bool
}
-// NewRegistryDataListener ...
-func NewRegistryDataListener(listener config_center.ConfigurationListener) *RegistryDataListener {
- return &RegistryDataListener{listener: listener}
+// NewRegistryDataListener constructs a new RegistryDataListener
+func NewRegistryDataListener() *RegistryDataListener {
+ return &RegistryDataListener{
+ subscribed: make(map[*common.URL]config_center.ConfigurationListener)}
}
-// AddInterestedURL ...
-func (l *RegistryDataListener) AddInterestedURL(url *common.URL) {
- l.interestedURL = append(l.interestedURL, url)
+// SubscribeURL is used to set a watch listener for url
+func (l *RegistryDataListener) SubscribeURL(url *common.URL, listener config_center.ConfigurationListener) {
+ if l.closed {
+ return
+ }
+ l.subscribed[url] = listener
}
-// DataChange ...
+// DataChange accepts all events sent from the zookeeper server and trigger the corresponding listener for processing
func (l *RegistryDataListener) DataChange(eventType remoting.Event) bool {
// Intercept the last bit
index := strings.Index(eventType.Path, "/providers/")
@@ -65,10 +70,14 @@ func (l *RegistryDataListener) DataChange(eventType remoting.Event) bool {
logger.Errorf("Listen NewURL(r{%s}) = error{%v} eventType.Path={%v}", url, err, eventType.Path)
return false
}
-
- for _, v := range l.interestedURL {
- if serviceURL.URLEqual(*v) {
- l.listener.Process(
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+ if l.closed {
+ return false
+ }
+ for url, listener := range l.subscribed {
+ if serviceURL.URLEqual(*url) {
+ listener.Process(
&config_center.ConfigChangeEvent{
Key: eventType.Path,
Value: serviceURL,
@@ -81,38 +90,48 @@ func (l *RegistryDataListener) DataChange(eventType remoting.Event) bool {
return false
}
-// RegistryConfigurationListener ...
+// Close all RegistryConfigurationListener in subscribed
+func (l *RegistryDataListener) Close() {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+ for _, listener := range l.subscribed {
+ listener.(*RegistryConfigurationListener).Close()
+ }
+}
+
+// RegistryConfigurationListener represent the processor of zookeeper watcher
type RegistryConfigurationListener struct {
client *zk.ZookeeperClient
registry *zkRegistry
events chan *config_center.ConfigChangeEvent
isClosed bool
+ close chan struct{}
closeOnce sync.Once
}
// NewRegistryConfigurationListener for listening the event of zk.
func NewRegistryConfigurationListener(client *zk.ZookeeperClient, reg *zkRegistry) *RegistryConfigurationListener {
reg.WaitGroup().Add(1)
- return &RegistryConfigurationListener{client: client, registry: reg, events: make(chan *config_center.ConfigChangeEvent, 32), isClosed: false}
+ return &RegistryConfigurationListener{client: client, registry: reg, events: make(chan *config_center.ConfigChangeEvent, 32), isClosed: false, close: make(chan struct{}, 1)}
}
-// Process ...
+// Process submit the ConfigChangeEvent to the event chan to notify all observer
func (l *RegistryConfigurationListener) Process(configType *config_center.ConfigChangeEvent) {
l.events <- configType
}
-// Next ...
+// Next will observe the registry state and events chan
func (l *RegistryConfigurationListener) Next() (*registry.ServiceEvent, error) {
for {
select {
case <-l.client.Done():
- logger.Warnf("listener's zk client connection is broken, so zk event listener exit now.")
- return nil, perrors.New("listener stopped")
-
+ logger.Warnf("listener's zk client connection (address {%s}) is broken, so zk event listener exit now.", l.client.ZkAddrs)
+ return nil, perrors.New("zookeeper client stopped")
+ case <-l.close:
+ return nil, perrors.New("listener have been closed")
case <-l.registry.Done():
- logger.Warnf("zk consumer register has quit, so zk event listener exit now.")
- return nil, perrors.New("listener stopped")
-
+ logger.Warnf("zk consumer register has quit, so zk event listener exit now. (registry url {%v}", l.registry.BaseRegistry.URL)
+ return nil, perrors.New("zookeeper registry, (registry url{%v}) stopped")
case e := <-l.events:
logger.Debugf("got zk event %s", e)
if e.ConfigType == remoting.EventTypeDel && !l.valid() {
@@ -127,15 +146,17 @@ func (l *RegistryConfigurationListener) Next() (*registry.ServiceEvent, error) {
}
}
-// Close ...
+// Close RegistryConfigurationListener only once
func (l *RegistryConfigurationListener) Close() {
// ensure that the listener will be closed at most once.
l.closeOnce.Do(func() {
l.isClosed = true
+ l.close <- struct{}{}
l.registry.WaitGroup().Done()
})
}
+// valid return the true if the client conn isn't nil
func (l *RegistryConfigurationListener) valid() bool {
return l.client.ZkConnValid()
}
diff --git a/registry/zookeeper/listener_test.go b/registry/zookeeper/listener_test.go
index 1a76b29a6f64e0329b289ce50218032a25f6f5cd..a0e9147a9e0ee8767efcf78d5e2aa536140f6a8b 100644
--- a/registry/zookeeper/listener_test.go
+++ b/registry/zookeeper/listener_test.go
@@ -32,15 +32,15 @@ import (
)
func Test_DataChange(t *testing.T) {
- listener := NewRegistryDataListener(&MockDataListener{})
+ listener := NewRegistryDataListener()
url, _ := common.NewURL("jsonrpc%3A%2F%2F127.0.0.1%3A20001%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26app.version%3D0.0.1%26application%3DBDTService%26category%3Dproviders%26cluster%3Dfailover%26dubbo%3Ddubbo-provider-golang-1.3.0%26environment%3Ddev%26group%3D%26interface%3Dcom.ikurento.user.UserProvider%26ip%3D10.32.20.124%26loadbalance%3Drandom%26methods.GetUser.loadbalance%3Drandom%26methods.GetUser.retries%3D1%26methods.GetUser.weight%3D0%26module%3Ddubbogo%2Buser-info%2Bserver%26name%3DBDTService%26organization%3Dikurento.com%26owner%3DZX%26pid%3D74500%26retries%3D0%26service.filter%3Decho%26side%3Dprovider%26timestamp%3D1560155407%26version%3D%26warmup%3D100")
- listener.AddInterestedURL(&url)
+ listener.SubscribeURL(&url, &MockConfigurationListener{})
int := listener.DataChange(remoting.Event{Path: "/dubbo/com.ikurento.user.UserProvider/providers/jsonrpc%3A%2F%2F127.0.0.1%3A20001%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26app.version%3D0.0.1%26application%3DBDTService%26category%3Dproviders%26cluster%3Dfailover%26dubbo%3Ddubbo-provider-golang-1.3.0%26environment%3Ddev%26group%3D%26interface%3Dcom.ikurento.user.UserProvider%26ip%3D10.32.20.124%26loadbalance%3Drandom%26methods.GetUser.loadbalance%3Drandom%26methods.GetUser.retries%3D1%26methods.GetUser.weight%3D0%26module%3Ddubbogo%2Buser-info%2Bserver%26name%3DBDTService%26organization%3Dikurento.com%26owner%3DZX%26pid%3D74500%26retries%3D0%26service.filter%3Decho%26side%3Dprovider%26timestamp%3D1560155407%26version%3D%26warmup%3D100"})
assert.Equal(t, true, int)
}
-type MockDataListener struct {
+type MockConfigurationListener struct {
}
-func (*MockDataListener) Process(configType *config_center.ConfigChangeEvent) {
+func (*MockConfigurationListener) Process(configType *config_center.ConfigChangeEvent) {
}
diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go
index e13443d57d7dae9fb5d50b2e1c28f618780fd850..88d5d6221b4bc7136ba4c3e7c95fb53ba35a9a58 100644
--- a/registry/zookeeper/registry.go
+++ b/registry/zookeeper/registry.go
@@ -20,7 +20,6 @@ package zookeeper
import (
"fmt"
"net/url"
- "strings"
"sync"
"time"
)
@@ -54,12 +53,11 @@ func init() {
type zkRegistry struct {
registry.BaseRegistry
- client *zookeeper.ZookeeperClient
- listenerLock sync.Mutex
- listener *zookeeper.ZkEventListener
- dataListener *RegistryDataListener
- configListener *RegistryConfigurationListener
- cltLock sync.Mutex
+ client *zookeeper.ZookeeperClient
+ listenerLock sync.Mutex
+ listener *zookeeper.ZkEventListener
+ dataListener *RegistryDataListener
+ cltLock sync.Mutex
//for provider
zkPath map[string]int // key = protocol://ip:port/interface
}
@@ -78,13 +76,12 @@ func newZkRegistry(url *common.URL) (registry.Registry, error) {
if err != nil {
return nil, err
}
- r.WaitGroup().Add(1) //zk client start successful, then wg +1
go zookeeper.HandleClientRestart(r)
r.listener = zookeeper.NewZkEventListener(r.client)
- r.configListener = NewRegistryConfigurationListener(r.client, r)
- r.dataListener = NewRegistryDataListener(r.configListener)
+
+ r.dataListener = NewRegistryDataListener()
return r, nil
}
@@ -121,8 +118,27 @@ func newMockZkRegistry(url *common.URL, opts ...zookeeper.Option) (*zk.TestClust
func (r *zkRegistry) InitListeners() {
r.listener = zookeeper.NewZkEventListener(r.client)
- r.configListener = NewRegistryConfigurationListener(r.client, r)
- r.dataListener = NewRegistryDataListener(r.configListener)
+ newDataListener := NewRegistryDataListener()
+ // should recover if dataListener isn't nil before
+ if r.dataListener != nil {
+ // close all old listener
+ oldDataListener := r.dataListener
+ oldDataListener.mutex.Lock()
+ defer oldDataListener.mutex.Unlock()
+ recoverd := r.dataListener.subscribed
+ if recoverd != nil && len(recoverd) > 0 {
+ // recover all subscribed url
+ for conf, oldListener := range recoverd {
+ if regConfigListener, ok := oldListener.(*RegistryConfigurationListener); ok {
+ regConfigListener.Close()
+ }
+ newDataListener.SubscribeURL(conf, NewRegistryConfigurationListener(r.client, r))
+ go r.listener.ListenServiceEvent(conf, fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())), newDataListener)
+
+ }
+ }
+ }
+ r.dataListener = newDataListener
}
func (r *zkRegistry) CreatePath(path string) error {
@@ -155,8 +171,8 @@ func (r *zkRegistry) ZkClientLock() *sync.Mutex {
}
func (r *zkRegistry) CloseListener() {
- if r.configListener != nil {
- r.configListener.Close()
+ if r.dataListener != nil {
+ r.dataListener.Close()
}
}
@@ -173,32 +189,49 @@ func (r *zkRegistry) registerTempZookeeperNode(root string, node string) error {
logger.Errorf("zk.Create(root{%s}) = err{%v}", root, perrors.WithStack(err))
return perrors.WithStack(err)
}
+
+ // try to register the node
zkPath, err = r.client.RegisterTemp(root, node)
if err != nil {
- if err == zk.ErrNodeExists {
- logger.Warnf("RegisterTempNode(root{%s}, node{%s}) = error{%v}", root, node, perrors.WithStack(err))
- } else {
- logger.Errorf("RegisterTempNode(root{%s}, node{%s}) = error{%v}", root, node, perrors.WithStack(err))
+ logger.Errorf("Register temp node(root{%s}, node{%s}) = error{%v}", root, node, perrors.WithStack(err))
+ if perrors.Cause(err) == zk.ErrNodeExists {
+ // should delete the old node
+ logger.Info("Register temp node failed, try to delete the old and recreate (root{%s}, node{%s}) , ignore!", root, node)
+ if err = r.client.Delete(zkPath); err == nil {
+ _, err = r.client.RegisterTemp(root, node)
+ }
+ if err != nil {
+ logger.Errorf("Recreate the temp node failed, (root{%s}, node{%s}) = error{%v}", root, node, perrors.WithStack(err))
+ }
}
return perrors.WithMessagef(err, "RegisterTempNode(root{%s}, node{%s})", root, node)
}
- logger.Debugf("create a zookeeper node:%s", zkPath)
+ logger.Debugf("Create a zookeeper node:%s", zkPath)
return nil
}
func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListener, error) {
- var (
- zkListener *RegistryConfigurationListener
- )
- r.listenerLock.Lock()
- if r.configListener.isClosed {
- r.listenerLock.Unlock()
- return nil, perrors.New("configListener already been closed")
+ var zkListener *RegistryConfigurationListener
+ dataListener := r.dataListener
+ dataListener.mutex.Lock()
+ defer dataListener.mutex.Unlock()
+ if r.dataListener.subscribed[conf] != nil {
+
+ zkListener, _ := r.dataListener.subscribed[conf].(*RegistryConfigurationListener)
+ if zkListener != nil {
+ r.listenerLock.Lock()
+ defer r.listenerLock.Unlock()
+ if zkListener.isClosed {
+ return nil, perrors.New("configListener already been closed")
+ } else {
+ return zkListener, nil
+ }
+ }
}
- zkListener = r.configListener
- r.listenerLock.Unlock()
+
+ zkListener = NewRegistryConfigurationListener(r.client, r)
if r.listener == nil {
r.cltLock.Lock()
client := r.client
@@ -216,10 +249,9 @@ func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListen
}
//Interested register to dataconfig.
- r.dataListener.AddInterestedURL(conf)
- for _, v := range strings.Split(conf.GetParam(constant.CATEGORY_KEY, constant.DEFAULT_CATEGORY), ",") {
- go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+v, url.QueryEscape(conf.Service())), r.dataListener)
- }
+ r.dataListener.SubscribeURL(conf, zkListener)
+
+ go r.listener.ListenServiceEvent(conf, fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())), r.dataListener)
return zkListener, nil
}
diff --git a/remoting/zookeeper/client.go b/remoting/zookeeper/client.go
index 7dac6146fa2aab88771ef5b7cb0205736e6478a9..bd1da547766abb12dc742234787262212e3db314 100644
--- a/remoting/zookeeper/client.go
+++ b/remoting/zookeeper/client.go
@@ -118,7 +118,7 @@ func ValidateZookeeperClient(container zkClientFacade, opts ...Option) error {
for _, opt := range opts {
opt(opions)
}
-
+ connected := false
err = nil
lock := container.ZkClientLock()
@@ -143,6 +143,7 @@ func ValidateZookeeperClient(container zkClientFacade, opts ...Option) error {
return perrors.WithMessagef(err, "newZookeeperClient(address:%+v)", url.Location)
}
container.SetZkClient(newClient)
+ connected = true
}
if container.ZkClient().Conn == nil {
@@ -150,10 +151,16 @@ func ValidateZookeeperClient(container zkClientFacade, opts ...Option) error {
container.ZkClient().Conn, event, err = zk.Connect(container.ZkClient().ZkAddrs, container.ZkClient().Timeout)
if err == nil {
container.ZkClient().Wait.Add(1)
+ connected = true
go container.ZkClient().HandleZkEvent(event)
}
}
+ if connected {
+ logger.Info("Connect to zookeeper successfully, name{%s}, zk address{%v}", opions.zkName, url.Location)
+ container.WaitGroup().Add(1) //zk client start successful, then registry wg +1
+ }
+
return perrors.WithMessagef(err, "newZookeeperClient(address:%+v)", url.PrimitiveURL)
}
@@ -386,6 +393,7 @@ func (z *ZookeeperClient) Close() {
z.Conn = nil
z.Unlock()
if conn != nil {
+ logger.Warnf("zkClient Conn{name:%s, zk addr:%s} exit now.", z.name, conn.SessionID())
conn.Close()
}
@@ -470,7 +478,7 @@ func (z *ZookeeperClient) RegisterTemp(basePath string, node string) (string, er
//if err != nil && err != zk.ErrNodeExists {
if err != nil {
logger.Warnf("conn.Create(\"%s\", zk.FlagEphemeral) = error(%v)\n", zkPath, perrors.WithStack(err))
- return "", perrors.WithStack(err)
+ return zkPath, perrors.WithStack(err)
}
logger.Debugf("zkClient{%s} create a temp zookeeper node:%s\n", z.name, tmpPath)
diff --git a/remoting/zookeeper/facade.go b/remoting/zookeeper/facade.go
index 055db4f716a914354d1bada653fbc0a850b615b5..4e3945388ff402f60a02150615a8914f9cba2435 100644
--- a/remoting/zookeeper/facade.go
+++ b/remoting/zookeeper/facade.go
@@ -48,11 +48,11 @@ func HandleClientRestart(r zkClientFacade) {
failTimes int
)
- defer r.WaitGroup().Done()
LOOP:
for {
select {
case <-r.Done():
+ r.WaitGroup().Done() // dec the wg when registry is closed
logger.Warnf("(ZkProviderRegistry)reconnectZkRegistry goroutine exit now...")
break LOOP
// re-register all services
@@ -63,12 +63,14 @@ LOOP:
zkAddress := r.ZkClient().ZkAddrs
r.SetZkClient(nil)
r.ZkClientLock().Unlock()
+ r.WaitGroup().Done() // dec the wg when zk client is closed
// Connect zk until success.
failTimes = 0
for {
select {
case <-r.Done():
+ r.WaitGroup().Done() // dec the wg when registry is closed
logger.Warnf("(ZkProviderRegistry)reconnectZkRegistry goroutine exit now...")
break LOOP
case <-getty.GetTimeWheel().After(timeSecondDuration(failTimes * ConnDelay)): // Prevent crazy reconnection zk.
diff --git a/remoting/zookeeper/facade_test.go b/remoting/zookeeper/facade_test.go
index a41f6cd3230700332519ce1c2d3489bfcc4b6ef0..01d46da6cc1abae90210a323d32ac84bad80249b 100644
--- a/remoting/zookeeper/facade_test.go
+++ b/remoting/zookeeper/facade_test.go
@@ -38,6 +38,16 @@ type mockFacade struct {
done chan struct{}
}
+func newMockFacade(client *ZookeeperClient, url *common.URL) zkClientFacade {
+ mock := &mockFacade{
+ client: client,
+ URL: url,
+ }
+
+ mock.wg.Add(1)
+ return mock
+}
+
func (r *mockFacade) ZkClient() *ZookeeperClient {
return r.client
}
@@ -80,7 +90,7 @@ func Test_Facade(t *testing.T) {
assert.NoError(t, err)
defer ts.Stop()
url, _ := common.NewURL("mock://127.0.0.1")
- mock := &mockFacade{client: z, URL: &url}
+ mock := newMockFacade(z, &url)
go HandleClientRestart(mock)
states := []zk.State{zk.StateConnecting, zk.StateConnected, zk.StateHasSession}
verifyEventStateOrder(t, event, states, "event channel")
diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go
index eaf259f4417201c95172e95d7a87476575e004d5..84877667763ce870e76202844e9dc9dc1c3f008c 100644
--- a/remoting/zookeeper/listener.go
+++ b/remoting/zookeeper/listener.go
@@ -18,6 +18,7 @@
package zookeeper
import (
+ "github.com/apache/dubbo-go/common"
"path"
"strings"
"sync"
@@ -173,7 +174,7 @@ func (l *ZkEventListener) handleZkNodeEvent(zkPath string, children []string, li
}
}
-func (l *ZkEventListener) listenDirEvent(zkPath string, listener remoting.DataListener) {
+func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listener remoting.DataListener) {
defer l.wg.Done()
var (
@@ -224,7 +225,16 @@ func (l *ZkEventListener) listenDirEvent(zkPath string, listener remoting.DataLi
}
failTimes = 0
for _, c := range children {
- // listen l service node
+
+ // Only need to compare Path when subscribing to provider
+ if strings.LastIndex(zkPath, constant.PROVIDER_CATEGORY) != -1 {
+ provider, _ := common.NewURL(c)
+ if provider.Path != conf.Path {
+ continue
+ }
+ }
+
+ //listen l service node
dubboPath := path.Join(zkPath, c)
//Save the path to avoid listen repeatedly
@@ -232,7 +242,7 @@ func (l *ZkEventListener) listenDirEvent(zkPath string, listener remoting.DataLi
_, ok := l.pathMap[dubboPath]
l.pathMapLock.Unlock()
if ok {
- logger.Warnf("@zkPath %s has already been listened.", zkPath)
+ logger.Warnf("@zkPath %s has already been listened.", dubboPath)
continue
}
@@ -263,7 +273,7 @@ func (l *ZkEventListener) listenDirEvent(zkPath string, listener remoting.DataLi
strings.LastIndex(zkPath, constant.CONSUMER_CATEGORY) == -1 {
l.wg.Add(1)
go func(zkPath string, listener remoting.DataListener) {
- l.listenDirEvent(zkPath, listener)
+ l.listenDirEvent(conf, zkPath, listener)
logger.Warnf("listenDirEvent(zkPath{%s}) goroutine exit now", zkPath)
}(dubboPath, listener)
}
@@ -291,11 +301,11 @@ func timeSecondDuration(sec int) time.Duration {
// registry.go:Listen -> listenServiceEvent -> listenDirEvent -> ListenServiceNodeEvent
// |
// --------> ListenServiceNodeEvent
-func (l *ZkEventListener) ListenServiceEvent(zkPath string, listener remoting.DataListener) {
+func (l *ZkEventListener) ListenServiceEvent(conf *common.URL, zkPath string, listener remoting.DataListener) {
logger.Infof("listen dubbo path{%s}", zkPath)
l.wg.Add(1)
go func(zkPath string, listener remoting.DataListener) {
- l.listenDirEvent(zkPath, listener)
+ l.listenDirEvent(conf, zkPath, listener)
logger.Warnf("listenDirEvent(zkPath{%s}) goroutine exit now", zkPath)
}(zkPath, listener)
}
diff --git a/remoting/zookeeper/listener_test.go b/remoting/zookeeper/listener_test.go
index 7301cd52c392b6950b3a49f78e8124eae532b083..ba7d6ba81b6af97dc5ad3788e8399d08cbe5b2bb 100644
--- a/remoting/zookeeper/listener_test.go
+++ b/remoting/zookeeper/listener_test.go
@@ -97,7 +97,7 @@ func TestListener(t *testing.T) {
go client.HandleZkEvent(event)
listener := NewZkEventListener(client)
dataListener := &mockDataListener{client: client, changedData: changedData, wait: &wait}
- listener.ListenServiceEvent("/dubbo", dataListener)
+ listener.ListenServiceEvent(nil, "/dubbo", dataListener)
time.Sleep(1 * time.Second)
_, err := client.Conn.Set("/dubbo/dubbo.properties", []byte(changedData), 1)
assert.NoError(t, err)