diff --git a/common/constant/key.go b/common/constant/key.go
index 8150b7033f4ccf046297e3e749559d112dc4850d..cd23dd0f1ad3e000fe54c251d9563bcf12ba86c7 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -159,6 +159,8 @@ const (
NACOS_PROTOCOL_KEY = "protocol"
NACOS_PATH_KEY = "path"
NACOS_NAMESPACE_ID = "namespaceId"
+ NACOS_PASSWORD = "password"
+ NACOS_USERNAME = "username"
)
const (
diff --git a/common/rpc_service.go b/common/rpc_service.go
index 200e61b398a663ae677a9a18e4c1ef841ac1b538..9ef2b956aa955f4fc79c6f75bd060ccfee2d02ca 100644
--- a/common/rpc_service.go
+++ b/common/rpc_service.go
@@ -277,7 +277,7 @@ func (sm *serviceMap) UnRegister(interfaceName, protocol, serviceId string) erro
}
}
delete(svcs, serviceId)
- if len(sm.serviceMap) == 0 {
+ if len(sm.serviceMap[protocol]) == 0 {
delete(sm.serviceMap, protocol)
}
diff --git a/config_center/nacos/client.go b/config_center/nacos/client.go
index 3b432819f43327888ade3da5303e445d6a2ef0fe..6fe5c4d7df28a7693c732543140ed74f959dc77e 100644
--- a/config_center/nacos/client.go
+++ b/config_center/nacos/client.go
@@ -33,6 +33,7 @@ import (
)
import (
+ "github.com/apache/dubbo-go/common"
"github.com/apache/dubbo-go/common/constant"
"github.com/apache/dubbo-go/common/logger"
)
@@ -89,20 +90,17 @@ func ValidateNacosClient(container nacosClientFacade, opts ...option) error {
}
url := container.GetUrl()
- logDir = url.GetParam(constant.CONFIG_LOG_DIR_KEY, logDir)
-
+ timeout, err := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT))
+ if err != nil {
+ logger.Errorf("invalid timeout config %+v,got err %+v",
+ url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT), err)
+ return perrors.WithMessagef(err, "newNacosClient(address:%+v)", url.Location)
+ }
+ nacosAddresses := strings.Split(url.Location, ",")
if container.NacosClient() == nil {
- //in dubbo ,every registry only connect one node ,so this is []string{r.Address}
- timeout, err := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT))
- if err != nil {
- logger.Errorf("timeout config %v is invalid ,err is %v",
- url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT), err.Error())
- return perrors.WithMessagef(err, "newNacosClient(address:%+v)", url.Location)
- }
- nacosAddresses := strings.Split(url.Location, ",")
- newClient, err := newNacosClient(os.nacosName, nacosAddresses, timeout)
+ newClient, err := newNacosClient(os.nacosName, nacosAddresses, timeout, url)
if err != nil {
- logger.Warnf("newNacosClient(name{%s}, nacos address{%v}, timeout{%d}) = error{%v}",
+ logger.Errorf("newNacosClient(name{%s}, nacos address{%v}, timeout{%d}) = error{%v}",
os.nacosName, url.Location, timeout.String(), err)
return perrors.WithMessagef(err, "newNacosClient(address:%+v)", url.Location)
}
@@ -110,41 +108,19 @@ func ValidateNacosClient(container nacosClientFacade, opts ...option) error {
}
if container.NacosClient().Client() == nil {
- svrConfList := []nacosconst.ServerConfig{}
- for _, nacosAddr := range container.NacosClient().NacosAddrs {
- split := strings.Split(nacosAddr, ":")
- port, err := strconv.ParseUint(split[1], 10, 64)
- if err != nil {
- logger.Warnf("nacos addr port parse error ,error message is %v", err)
- continue
- }
- svrconf := nacosconst.ServerConfig{
- IpAddr: split[0],
- Port: port,
- }
- svrConfList = append(svrConfList, svrconf)
- }
-
- client, err := clients.CreateConfigClient(map[string]interface{}{
- "serverConfigs": svrConfList,
- "clientConfig": nacosconst.ClientConfig{
- TimeoutMs: uint64(int32(container.NacosClient().Timeout / time.Millisecond)),
- ListenInterval: 10000,
- NotLoadCacheAtStart: true,
- LogDir: logDir,
- },
- })
-
- container.NacosClient().SetClient(&client)
+ configClient, err := initNacosConfigClient(nacosAddresses, timeout, url)
if err != nil {
- logger.Errorf("nacos create config client error:%v", err)
+ logger.Errorf("initNacosConfigClient(addr:%+v,timeout:%v,url:%v) = err %+v",
+ nacosAddresses, timeout.String(), url, err)
+ return perrors.WithMessagef(err, "newNacosClient(address:%+v)", url.Location)
}
+ container.NacosClient().SetClient(&configClient)
}
return perrors.WithMessagef(nil, "newNacosClient(address:%+v)", url.PrimitiveURL)
}
-func newNacosClient(name string, nacosAddrs []string, timeout time.Duration) (*NacosClient, error) {
+func newNacosClient(name string, nacosAddrs []string, timeout time.Duration, url common.URL) (*NacosClient, error) {
var (
err error
n *NacosClient
@@ -160,12 +136,24 @@ func newNacosClient(name string, nacosAddrs []string, timeout time.Duration) (*N
},
}
- svrConfList := make([]nacosconst.ServerConfig, 0, len(n.NacosAddrs))
- for _, nacosAddr := range n.NacosAddrs {
+ configClient, err := initNacosConfigClient(nacosAddrs, timeout, url)
+ if err != nil {
+ logger.Errorf("initNacosConfigClient(addr:%+v,timeout:%v,url:%v) = err %+v",
+ nacosAddrs, timeout.String(), url, err)
+ return n, perrors.WithMessagef(err, "newNacosClient(address:%+v)", url.Location)
+ }
+ n.SetClient(&configClient)
+
+ return n, nil
+}
+
+func initNacosConfigClient(nacosAddrs []string, timeout time.Duration, url common.URL) (config_client.IConfigClient, error) {
+ svrConfList := []nacosconst.ServerConfig{}
+ for _, nacosAddr := range nacosAddrs {
split := strings.Split(nacosAddr, ":")
port, err := strconv.ParseUint(split[1], 10, 64)
if err != nil {
- logger.Warnf("convert port , source:%s , error:%v ", split[1], err)
+ logger.Errorf("strconv.ParseUint(nacos addr port:%+v) = error %+v", split[1], err)
continue
}
svrconf := nacosconst.ServerConfig{
@@ -174,21 +162,21 @@ func newNacosClient(name string, nacosAddrs []string, timeout time.Duration) (*N
}
svrConfList = append(svrConfList, svrconf)
}
- client, err := clients.CreateConfigClient(map[string]interface{}{
+
+ return clients.CreateConfigClient(map[string]interface{}{
"serverConfigs": svrConfList,
"clientConfig": nacosconst.ClientConfig{
- TimeoutMs: uint64(timeout / time.Millisecond),
- ListenInterval: 20000,
+ TimeoutMs: uint64(int32(timeout / time.Millisecond)),
+ ListenInterval: uint64(int32(timeout / time.Millisecond)),
NotLoadCacheAtStart: true,
- LogDir: logDir,
+ LogDir: url.GetParam(constant.NACOS_LOG_DIR_KEY, logDir),
+ CacheDir: url.GetParam(constant.NACOS_CACHE_DIR_KEY, ""),
+ Endpoint: url.GetParam(constant.NACOS_ENDPOINT, ""),
+ Username: url.GetParam(constant.NACOS_USERNAME, ""),
+ Password: url.GetParam(constant.NACOS_PASSWORD, ""),
+ NamespaceId: url.GetParam(constant.NACOS_NAMESPACE_ID, ""),
},
})
- n.SetClient(&client)
- if err != nil {
- return nil, perrors.WithMessagef(err, "nacos clients.CreateConfigClient(nacosAddrs:%+v)", nacosAddrs)
- }
-
- return n, nil
}
// Done Get nacos client exit signal
@@ -233,5 +221,4 @@ func (n *NacosClient) Close() {
n.stop()
n.SetClient(nil)
- logger.Warnf("nacosClient{name:%s, nacos addr:%s} exit now.", n.name, n.NacosAddrs)
}
diff --git a/config_center/nacos/client_test.go b/config_center/nacos/client_test.go
index 0cc36eb7c1648349047fea795e8bd8fef4bed72d..01319f362b956a0a15be8d5d4dde2a8b2be57c89 100644
--- a/config_center/nacos/client_test.go
+++ b/config_center/nacos/client_test.go
@@ -56,7 +56,7 @@ func TestNewNacosClient(t *testing.T) {
func TestSetNacosClient(t *testing.T) {
server := mockCommonNacosServer()
- nacosURL := server.Listener.Addr().String()
+ nacosURL := "registry://" + server.Listener.Addr().String()
registryUrl, _ := common.NewURL(nacosURL)
c := &nacosDynamicConfiguration{
url: ®istryUrl,
diff --git a/go.sum b/go.sum
index 5f35190e4cd1678e871b2a445e23f7289be329cd..eaeae0a22622a65dcf95b9b65c36289035aa3afd 100644
--- a/go.sum
+++ b/go.sum
@@ -37,6 +37,8 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/aliyun/alibaba-cloud-sdk-go v0.0.0-20190802083043-4cd0c391755e/go.mod h1:myCDvQSzCW+wB1WAlocEru4wMGJxy+vlxHdhegi1CDQ=
github.com/aliyun/alibaba-cloud-sdk-go v1.61.18 h1:zOVTBdCKFd9JbCKz9/nt+FovbjPFmb7mUnp8nH9fQBA=
+github.com/aliyun/alibaba-cloud-sdk-go v1.61.18 h1:zOVTBdCKFd9JbCKz9/nt+FovbjPFmb7mUnp8nH9fQBA=
+github.com/aliyun/alibaba-cloud-sdk-go v1.61.18/go.mod h1:v8ESoHo4SyHmuB4b1tJqDHxfTGEciD+yhvOU/5s1Rfk=
github.com/aliyun/alibaba-cloud-sdk-go v1.61.18/go.mod h1:v8ESoHo4SyHmuB4b1tJqDHxfTGEciD+yhvOU/5s1Rfk=
github.com/aliyun/aliyun-oss-go-sdk v0.0.0-20190307165228-86c17b95fcd5/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8=
github.com/apache/dubbo-go-hessian2 v1.5.0 h1:fzulDG5G7nX0ccgKdiN9XipJ7tZ4WXKgmk4stdlDS6s=
@@ -398,6 +400,8 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/nacos-group/nacos-sdk-go v0.3.2 h1:q+ukmIImL6u0zBtbceMZl2frgeAc45QT6cIrTZZz50c=
+github.com/nacos-group/nacos-sdk-go v0.3.2/go.mod h1:4TdsN7eZnnVCDlOlBa61b0gsRnvNJI74m9+2+OKZkcw=
github.com/nacos-group/nacos-sdk-go v0.3.3-0.20200617023039-50c7537d6a5f h1:gid5/0AkHvINWK69Fgbidb3BVIXqlf1YEm7wO0NVPsw=
github.com/nacos-group/nacos-sdk-go v0.3.3-0.20200617023039-50c7537d6a5f/go.mod h1:fti1GlX/EB6RDKvzK/P7Vuibqj0JMPJHQwrcTU1tLXk=
github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2 h1:BQ1HW7hr4IVovMwWg0E0PYcyW8CzqDcVmaew9cujU4s=
@@ -470,6 +474,8 @@ github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735 h1:7YvPJVmEeFHR1T
github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc=
github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b h1:gQZ0qzfKHQIybLANtM3mBXNUtOfsCFXeTsnBqCsx1KM=
+github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b h1:gQZ0qzfKHQIybLANtM3mBXNUtOfsCFXeTsnBqCsx1KM=
+github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
diff --git a/remoting/nacos/builder.go b/remoting/nacos/builder.go
index 4319627c6da6a5a19874f188d9eba6b032645ff1..8a247e267daa02d748c9a5e47ced698f617dfe9a 100644
--- a/remoting/nacos/builder.go
+++ b/remoting/nacos/builder.go
@@ -64,10 +64,7 @@ func getNacosConfig(url *common.URL) (map[string]interface{}, error) {
if err != nil {
return nil, perrors.WithMessagef(err, "split [%s] ", addr)
}
- port, err := strconv.Atoi(portStr)
- if err != nil {
- return configMap, perrors.WithMessage(err, "the port string is invalid. "+portStr)
- }
+ port, _ := strconv.Atoi(portStr)
serverConfigs = append(serverConfigs, nacosConstant.ServerConfig{
IpAddr: ip,
Port: uint64(port),
@@ -75,21 +72,22 @@ func getNacosConfig(url *common.URL) (map[string]interface{}, error) {
}
configMap["serverConfigs"] = serverConfigs
+ var clientConfig nacosConstant.ClientConfig
timeout, err := time.ParseDuration(url.GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT))
if err != nil {
return nil, err
}
-
- timeoutMs := uint64(timeout.Nanoseconds() / constant.MsToNanoRate)
-
- configMap["clientConfig"] = nacosConstant.ClientConfig{
- TimeoutMs: timeoutMs,
- ListenInterval: 2 * timeoutMs,
- CacheDir: url.GetParam(constant.NACOS_CACHE_DIR_KEY, ""),
- LogDir: url.GetParam(constant.NACOS_LOG_DIR_KEY, ""),
- Endpoint: url.GetParam(constant.NACOS_ENDPOINT, ""),
- NotLoadCacheAtStart: true,
- }
+ clientConfig.TimeoutMs = uint64(timeout.Seconds() * 1000)
+ clientConfig.ListenInterval = 2 * clientConfig.TimeoutMs
+ clientConfig.CacheDir = url.GetParam(constant.NACOS_CACHE_DIR_KEY, "")
+ clientConfig.LogDir = url.GetParam(constant.NACOS_LOG_DIR_KEY, "")
+ clientConfig.Endpoint = url.GetParam(constant.NACOS_ENDPOINT, "")
+ clientConfig.NamespaceId = url.GetParam(constant.NACOS_NAMESPACE_ID, "")
+ clientConfig.Username = url.GetParam(constant.NACOS_USERNAME, "")
+ clientConfig.Password = url.GetParam(constant.NACOS_PASSWORD, "")
+ clientConfig.NamespaceId = url.GetParam(constant.NACOS_NAMESPACE_ID, "")
+ clientConfig.NotLoadCacheAtStart = true
+ configMap["clientConfig"] = clientConfig
return configMap, nil
}
@@ -122,7 +120,9 @@ func NewNacosClient(rc *config.RemoteConfig) (naming_client.INamingClient, error
clientConfig.ListenInterval = 2 * clientConfig.TimeoutMs
clientConfig.CacheDir = rc.GetParam(constant.NACOS_CACHE_DIR_KEY, "")
clientConfig.LogDir = rc.GetParam(constant.NACOS_LOG_DIR_KEY, "")
- clientConfig.Endpoint = rc.GetParam(constant.NACOS_ENDPOINT, "")
+ clientConfig.Endpoint = rc.Address
+ clientConfig.Username = rc.Username
+ clientConfig.Password = rc.Password
clientConfig.NotLoadCacheAtStart = true
configMap["clientConfig"] = clientConfig
diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go
index cde5d6c497cfa0c93a8b88e340d3a489ebcc483a..9a4874db24696d90e4fcc7d9d987f5888f1be599 100644
--- a/remoting/zookeeper/listener.go
+++ b/remoting/zookeeper/listener.go
@@ -89,13 +89,21 @@ func (l *ZkEventListener) listenServiceNodeEvent(zkPath string, listener ...remo
case zk.EventNodeDataChanged:
logger.Warnf("zk.ExistW(key{%s}) = event{EventNodeDataChanged}", zkPath)
if len(listener) > 0 {
- content, _, _ := l.client.Conn.Get(zkEvent.Path)
+ content, _, err := l.client.Conn.Get(zkEvent.Path)
+ if err != nil {
+ logger.Warnf("zk.Conn.Get{key:%s} = error{%v}", zkPath, err)
+ return false
+ }
listener[0].DataChange(remoting.Event{Path: zkEvent.Path, Action: remoting.EventTypeUpdate, Content: string(content)})
}
case zk.EventNodeCreated:
logger.Warnf("zk.ExistW(key{%s}) = event{EventNodeCreated}", zkPath)
if len(listener) > 0 {
- content, _, _ := l.client.Conn.Get(zkEvent.Path)
+ content, _, err := l.client.Conn.Get(zkEvent.Path)
+ if err != nil {
+ logger.Warnf("zk.Conn.Get{key:%s} = error{%v}", zkPath, err)
+ return false
+ }
listener[0].DataChange(remoting.Event{Path: zkEvent.Path, Action: remoting.EventTypeAdd, Content: string(content)})
}
case zk.EventNotWatching:
@@ -239,15 +247,15 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listen
// 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 {
+ if provider.ServiceKey() != conf.ServiceKey() {
continue
}
}
- //listen l service node
+ // listen l service node
dubboPath := path.Join(zkPath, c)
- //Save the path to avoid listen repeatedly
+ // Save the path to avoid listen repeatedly
l.pathMapLock.Lock()
_, ok := l.pathMap[dubboPath]
l.pathMapLock.Unlock()
@@ -259,7 +267,7 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listen
l.pathMapLock.Lock()
l.pathMap[dubboPath] = struct{}{}
l.pathMapLock.Unlock()
- //When Zk disconnected, the Conn will be set to nil, so here need check the value of Conn
+ // When Zk disconnected, the Conn will be set to nil, so here need check the value of Conn
l.client.RLock()
if l.client.Conn == nil {
l.client.RUnlock()
@@ -283,8 +291,8 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listen
logger.Warnf("listenSelf(zk path{%s}) goroutine exit now", zkPath)
}(dubboPath, listener)
- //listen sub path recursive
- //if zkPath is end of "providers/ & consumers/" we do not listen children dir
+ // listen sub path recursive
+ // if zkPath is end of "providers/ & consumers/" we do not listen children dir
if strings.LastIndex(zkPath, constant.PROVIDER_CATEGORY) == -1 &&
strings.LastIndex(zkPath, constant.CONSUMER_CATEGORY) == -1 {
l.wg.Add(1)