diff --git a/common/constant/env.go b/common/constant/env.go
index f0e4fabe02cb09f63173bc9fe0cbf83885bdefb6..759cb0be0a2bd36a2a345a360c541b7d56813d70 100644
--- a/common/constant/env.go
+++ b/common/constant/env.go
@@ -18,7 +18,7 @@
package constant
const (
- CONF_CONSUMER_FILE_PATH = "CONF_CONSUMER_FILE_PATH"
- CONF_PROVIDER_FILE_PATH = "CONF_PROVIDER_FILE_PATH"
- APP_LOG_CONF_FILE string = "APP_LOG_CONF_FILE"
+ CONF_CONSUMER_FILE_PATH = "CONF_CONSUMER_FILE_PATH"
+ CONF_PROVIDER_FILE_PATH = "CONF_PROVIDER_FILE_PATH"
+ APP_LOG_CONF_FILE = "APP_LOG_CONF_FILE"
)
diff --git a/config/config_loader.go b/config/config_loader.go
index 0b48761fd63b713b66f7b1b92f62b682dda5c927..b737d3f233700f596469cfd678aa7ae7f9a82b85 100644
--- a/config/config_loader.go
+++ b/config/config_loader.go
@@ -54,6 +54,7 @@ func init() {
providerConfig = nil
}
}
+
func checkRegistries(registries map[string]*RegistryConfig, singleRegistry *RegistryConfig) {
if len(registries) == 0 && singleRegistry != nil {
registries[constant.DEFAULT_KEY] = singleRegistry
diff --git a/registry/consul/registry.go b/registry/consul/registry.go
index 1fd3e54e96c446f59dd31ab30575eef5b455b72f..73bf3975bc7c73f4a7748f46280ffb1aa5525ca8 100644
--- a/registry/consul/registry.go
+++ b/registry/consul/registry.go
@@ -111,31 +111,21 @@ func (r *consulRegistry) unregister(url common.URL) error {
return r.client.Agent().ServiceDeregister(buildId(url))
}
-func (r *consulRegistry) subscribe(url *common.URL) (registry.Listener, error) {
- var (
- listener registry.Listener
- err error
- )
-
+func (r *consulRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
role, _ := strconv.Atoi(r.URL.GetParam(constant.ROLE_KEY, ""))
if role == common.CONSUMER {
- listener, err = r.getListener(*url)
- if err != nil {
- return nil, err
- }
+ r.subscribe(url, notifyListener)
}
- return listener, nil
}
-//subscibe from registry
-func (r *consulRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
+func (r *consulRegistry) subscribe(url *common.URL, notifyListener registry.NotifyListener) {
for {
if !r.IsAvailable() {
logger.Warnf("event listener game over.")
return
}
- listener, err := r.subscribe(url)
+ listener, err := r.getListener(*url)
if err != nil {
if !r.IsAvailable() {
logger.Warnf("event listener game over.")
@@ -155,9 +145,7 @@ func (r *consulRegistry) Subscribe(url *common.URL, notifyListener registry.Noti
logger.Infof("update begin, service event: %v", serviceEvent.String())
notifyListener.Notify(serviceEvent)
}
-
}
-
}
}
diff --git a/registry/consul/registry_test.go b/registry/consul/registry_test.go
index ff8c2e23163e85b128436bde001455f316b28dc6..bb6842cd8fb67dd2cc70b1a7530fbb94f618a9b0 100644
--- a/registry/consul/registry_test.go
+++ b/registry/consul/registry_test.go
@@ -51,7 +51,7 @@ func (suite *consulRegistryTestSuite) testUnregister() {
func (suite *consulRegistryTestSuite) testSubscribe() {
consumerUrl := newConsumerUrl(consumerHost, consumerPort, service, protocol)
suite.consumerUrl = consumerUrl
- listener, err := suite.consumerRegistry.subscribe(&consumerUrl)
+ listener, err := suite.consumerRegistry.getListener(consumerUrl)
assert.NoError(suite.t, err)
suite.listener = listener
}
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index 54f0acd8433f8eb1bbefb36403248056f70177de..b6794e2ebf212ec024d73335ddb19441afa5b96e 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -40,10 +40,6 @@ import (
"github.com/apache/dubbo-go/registry"
)
-const (
- RegistryConnDelay = 3
-)
-
type Options struct {
serviceTTL time.Duration
}
@@ -87,7 +83,7 @@ func NewRegistryDirectory(url *common.URL, registry registry.Registry, opts ...O
return dir, nil
}
-//subscibe from registry
+//subscribe from registry
func (dir *registryDirectory) Subscribe(url *common.URL) {
dir.consumerConfigurationListener.addNotifyListener(dir)
dir.referenceConfigurationListener = newReferenceConfigurationListener(dir, url)
@@ -245,6 +241,7 @@ func (dir *registryDirectory) Destroy() {
dir.cacheInvokers = []protocol.Invoker{}
})
}
+
func (dir *registryDirectory) overrideUrl(targetUrl *common.URL) {
doOverrideUrl(dir.configurators, targetUrl)
doOverrideUrl(dir.consumerConfigurationListener.Configurators(), targetUrl)
@@ -293,9 +290,11 @@ func newConsumerConfigurationListener(dir *registryDirectory) *consumerConfigura
)
return listener
}
+
func (l *consumerConfigurationListener) addNotifyListener(listener registry.NotifyListener) {
l.listeners = append(l.listeners, listener)
}
+
func (l *consumerConfigurationListener) Process(event *config_center.ConfigChangeEvent) {
l.BaseConfigurationListener.Process(event)
l.directory.refreshInvokers(nil)
diff --git a/registry/etcdv3/registry.go b/registry/etcdv3/registry.go
index 8bb1ff430eb8e674666266af5169bba4b4e22d30..bf097f0b59573bc8d811015105065d6e2e1eb386 100644
--- a/registry/etcdv3/registry.go
+++ b/registry/etcdv3/registry.go
@@ -330,7 +330,7 @@ func (r *etcdV3Registry) subscribe(svc *common.URL) (registry.Listener, error) {
return configListener, nil
}
-//subscibe from registry
+//subscribe from registry
func (r *etcdV3Registry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
for {
if !r.IsAvailable() {
diff --git a/registry/nacos/registry.go b/registry/nacos/registry.go
index f1a78264ce431745181ca6f633eda642cf90a31e..229c6f8f0cea862b5ac676fe9d0d3f34cc6ed1b4 100644
--- a/registry/nacos/registry.go
+++ b/registry/nacos/registry.go
@@ -172,17 +172,17 @@ func (nr *nacosRegistry) subscribe(conf *common.URL) (registry.Listener, error)
return NewNacosListener(*conf, nr.namingClient)
}
-//subscibe from registry
-func (r *nacosRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
+//subscribe from registry
+func (nr *nacosRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
for {
- if !r.IsAvailable() {
+ if !nr.IsAvailable() {
logger.Warnf("event listener game over.")
return
}
- listener, err := r.subscribe(url)
+ listener, err := nr.subscribe(url)
if err != nil {
- if !r.IsAvailable() {
+ if !nr.IsAvailable() {
logger.Warnf("event listener game over.")
return
}
@@ -205,6 +205,7 @@ func (r *nacosRegistry) Subscribe(url *common.URL, notifyListener registry.Notif
}
}
+
func (nr *nacosRegistry) GetUrl() common.URL {
return *nr.URL
}
diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go
index e14541dd04691bf89825b1d77d79932ad54f7720..972a4b6c5e0af7e53e9aad6df33209d15f71a587 100644
--- a/registry/zookeeper/registry.go
+++ b/registry/zookeeper/registry.go
@@ -147,6 +147,7 @@ func newMockZkRegistry(url *common.URL, opts ...zookeeper.Option) (*zk.TestClust
return c, r, nil
}
+
func (r *zkRegistry) ZkClient() *zookeeper.ZookeeperClient {
return r.client
}
@@ -399,7 +400,7 @@ func (r *zkRegistry) subscribe(conf *common.URL) (registry.Listener, error) {
return r.getListener(conf)
}
-//subscibe from registry
+//subscribe from registry
func (r *zkRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) {
for {
if !r.IsAvailable() {
@@ -432,6 +433,7 @@ func (r *zkRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyLi
}
}
+
func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListener, error) {
var (
zkListener *RegistryConfigurationListener