This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new e17b91ce1 [Enhancement] remove legacy config package (#3372)
e17b91ce1 is described below
commit e17b91ce10b2b1d70a208be054aee90ad7814281
Author: hurukawa <[email protected]>
AuthorDate: Wed Jun 10 09:14:06 2026 +0800
[Enhancement] remove legacy config package (#3372)
* pr1: remove
config.GetRootConfig()、config.GetProviderConfig()、config.GetProviderService()、config.GetShutDown()
---
common/config/utils.go | 72 ++++++++++++
common/config/utils_test.go | 60 ++++++++++
config/service_config.go | 2 +
config/service_config_test.go | 40 +++++++
registry/directory/directory.go | 107 ++++++++---------
registry/directory/directory_test.go | 175 ++++++++++++++++++++++++++++
registry/protocol/protocol.go | 188 ++++++++++++++++--------------
registry/protocol/protocol_test.go | 218 ++++++++++++++++++++++++++++++++++-
8 files changed, 710 insertions(+), 152 deletions(-)
diff --git a/common/config/utils.go b/common/config/utils.go
index 2f5a68e5b..cec9f98c0 100644
--- a/common/config/utils.go
+++ b/common/config/utils.go
@@ -30,7 +30,9 @@ import (
)
import (
+ "dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
)
var validate *validator.Validate
@@ -114,3 +116,73 @@ func removeMinus(strArr []string) string {
func IsValid(addr string) bool {
return addr != "" && addr != constant.NotAvailable
}
+
+// EnsureApplicationAttribute resolves application config from the primary URL,
+// fallback URLs, URL params, then the default application name.
+func EnsureApplicationAttribute(url *common.URL, fallbackURLs ...*common.URL)
*global.ApplicationConfig {
+ urls := make([]*common.URL, 0, len(fallbackURLs)+1)
+ urls = append(urls, url)
+ urls = append(urls, fallbackURLs...)
+
+ for _, candidate := range urls {
+ if application, ok := applicationFromAttribute(candidate); ok {
+ setApplicationAttribute(url, application)
+ return application
+ }
+ }
+
+ for _, candidate := range urls {
+ if application := applicationFromParam(candidate); application
!= nil {
+ setApplicationAttribute(url, application)
+ return application
+ }
+ }
+
+ application := &global.ApplicationConfig{Name: constant.DefaultDubboApp}
+ setApplicationAttribute(url, application)
+ return application
+}
+
+func setApplicationAttribute(url *common.URL, application
*global.ApplicationConfig) {
+ if url != nil {
+ url.SetAttribute(constant.ApplicationKey, application)
+ }
+}
+
+func applicationFromAttribute(url *common.URL) (*global.ApplicationConfig,
bool) {
+ if url == nil {
+ return nil, false
+ }
+
+ applicationRaw, ok := url.GetAttribute(constant.ApplicationKey)
+ if !ok {
+ return nil, false
+ }
+
+ switch application := applicationRaw.(type) {
+ case *global.ApplicationConfig:
+ if application != nil && application.Name != "" {
+ return application, true
+ }
+ case global.ApplicationConfig:
+ if application.Name != "" {
+ applicationCopy := application
+ return &applicationCopy, true
+ }
+ case string:
+ if application != "" {
+ return &global.ApplicationConfig{Name: application},
true
+ }
+ }
+ return nil, false
+}
+
+func applicationFromParam(url *common.URL) *global.ApplicationConfig {
+ if url == nil {
+ return nil
+ }
+ if applicationName := url.GetParam(constant.ApplicationKey, "");
applicationName != "" {
+ return &global.ApplicationConfig{Name: applicationName}
+ }
+ return nil
+}
diff --git a/common/config/utils_test.go b/common/config/utils_test.go
index ca287e1d1..ace61bcaa 100644
--- a/common/config/utils_test.go
+++ b/common/config/utils_test.go
@@ -27,7 +27,9 @@ import (
)
import (
+ "dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
)
func TestTranslateIds(t *testing.T) {
@@ -148,3 +150,61 @@ func TestIsValid(t *testing.T) {
assert.False(t, IsValid(""))
assert.False(t, IsValid(constant.NotAvailable))
}
+
+func TestEnsureApplicationAttribute(t *testing.T) {
+ t.Run("primary attribute has highest priority", func(t *testing.T) {
+ primary := common.NewURLWithOptions(
+ common.WithAttribute(constant.ApplicationKey,
"primary-app"),
+ common.WithParamsValue(constant.ApplicationKey,
"primary-param"),
+ )
+ fallback := common.NewURLWithOptions(
+ common.WithAttribute(constant.ApplicationKey,
"fallback-app"),
+ )
+
+ application := EnsureApplicationAttribute(primary, fallback)
+
+ require.NotNil(t, application)
+ assert.Equal(t, "primary-app", application.Name)
+ applicationRaw, ok :=
primary.GetAttribute(constant.ApplicationKey)
+ require.True(t, ok)
+ assert.Equal(t, application, applicationRaw)
+ })
+
+ t.Run("fallback attribute wins over primary param", func(t *testing.T) {
+ primary := common.NewURLWithOptions(
+ common.WithParamsValue(constant.ApplicationKey,
"primary-param"),
+ )
+ fallback := common.NewURLWithOptions(
+ common.WithAttribute(constant.ApplicationKey,
global.ApplicationConfig{Name: "fallback-app"}),
+ )
+
+ application := EnsureApplicationAttribute(primary, fallback)
+
+ require.NotNil(t, application)
+ assert.Equal(t, "fallback-app", application.Name)
+ applicationRaw, ok :=
primary.GetAttribute(constant.ApplicationKey)
+ require.True(t, ok)
+ assert.Equal(t, application, applicationRaw)
+ })
+
+ t.Run("fallback param is used before default", func(t *testing.T) {
+ primary := common.NewURLWithOptions()
+ fallback := common.NewURLWithOptions(
+ common.WithParamsValue(constant.ApplicationKey,
"fallback-param"),
+ )
+
+ application := EnsureApplicationAttribute(primary, fallback)
+
+ require.NotNil(t, application)
+ assert.Equal(t, "fallback-param", application.Name)
+ })
+
+ t.Run("default application is used when no source exists", func(t
*testing.T) {
+ primary := common.NewURLWithOptions()
+
+ application := EnsureApplicationAttribute(primary)
+
+ require.NotNil(t, application)
+ assert.Equal(t, constant.DefaultDubboApp, application.Name)
+ })
+}
diff --git a/config/service_config.go b/config/service_config.go
index e17c46f56..cf7e5fb1a 100644
--- a/config/service_config.go
+++ b/config/service_config.go
@@ -284,6 +284,8 @@ func (s *ServiceConfig) Export() error {
common.WithParamsValue(constant.ApplicationTagKey,
s.rc.Application.Tag),
//common.WithParamsValue(constant.SslEnabledKey,
strconv.FormatBool(config.GetSslEnabled())),
common.WithMethods(strings.Split(methods, ",")),
+ common.WithAttribute(constant.RpcServiceKey,
s.rpcService),
+ common.WithAttribute(constant.ProviderConfigKey,
s.rc.Provider),
common.WithToken(s.Token),
common.WithParamsValue(constant.MetadataTypeKey,
s.metadataType),
// fix https://github.com/apache/dubbo-go/issues/2176
diff --git a/config/service_config_test.go b/config/service_config_test.go
index 8624a51ea..36f5f2bab 100644
--- a/config/service_config_test.go
+++ b/config/service_config_test.go
@@ -34,6 +34,8 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/protocol/protocolwrapper"
_ "dubbo.apache.org/dubbo-go/v3/proxy/proxy_factory"
)
@@ -189,3 +191,41 @@ func
TestServiceConfigExportDoesNotWarnOnNonVariadicRPCMethods(t *testing.T) {
assert.NotContains(t, warn, "variadic RPC method")
}
}
+
+func TestServiceConfigExportCarriesProviderAttributes(t *testing.T) {
+ extension.SetProtocol(protocolwrapper.FILTER,
protocolwrapper.NewMockProtocolFilter)
+
+ const interfaceName = "org.apache.dubbo.ProviderAttributeService"
+ serviceConfig := newEmptyServiceConfig()
+ serviceConfig.id = "ProviderAttributeService"
+ serviceConfig.Interface = interfaceName
+ serviceConfig.NotRegister = true
+ serviceConfig.ProtocolIDs = []string{"dubbo"}
+ serviceConfig.rpcService = &HelloService{}
+
+ rc := newEmptyRootConfig()
+ rc.Protocols["dubbo"] = NewProtocolConfigBuilder().
+ SetName("dubbo").
+ SetPort("20000").
+ Build()
+
+ err := serviceConfig.Init(rc)
+ require.NoError(t, err)
+ err = serviceConfig.Export()
+ require.NoError(t, err)
+ t.Cleanup(serviceConfig.Unexport)
+ t.Cleanup(func() {
+ _ = common.ServiceMap.UnRegister(interfaceName, "dubbo",
common.ServiceKey(interfaceName, "", ""))
+ })
+
+ urls := serviceConfig.GetExportedUrls()
+ require.Len(t, urls, 1)
+
+ providerRaw, ok := urls[0].GetAttribute(constant.ProviderConfigKey)
+ require.True(t, ok)
+ assert.Same(t, rc.Provider, providerRaw)
+
+ rpcServiceRaw, ok := urls[0].GetAttribute(constant.RpcServiceKey)
+ require.True(t, ok)
+ assert.Same(t, serviceConfig.rpcService, rpcServiceRaw)
+}
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index dc1821c41..f5dbdf835 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -37,9 +37,9 @@ import (
"dubbo.apache.org/dubbo-go/v3/cluster/directory/static"
"dubbo.apache.org/dubbo-go/v3/cluster/router/chain"
"dubbo.apache.org/dubbo-go/v3/common"
+ commonConfig "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
- "dubbo.apache.org/dubbo-go/v3/config"
"dubbo.apache.org/dubbo-go/v3/config_center"
_ "dubbo.apache.org/dubbo-go/v3/config_center/configurator"
"dubbo.apache.org/dubbo-go/v3/global"
@@ -97,60 +97,57 @@ var defaultClosingTombstoneTTL = func() time.Duration {
return 30 * time.Second
}()
-// NewRegistryDirectory will create a new RegistryDirectory
-func NewRegistryDirectory(url *common.URL, registry registry.Registry)
(directory.Directory, error) {
- if url.SubURL == nil {
- return nil, perrors.Errorf("url is invalid, suburl can not be
nil")
+func ensureRegistriesAttribute(url *common.URL) {
+ if registries, ok := registriesFromAttribute(url); ok {
+ url.SetAttribute(constant.RegistriesConfigKey, registries)
+ return
}
- logger.Debugf("[Registry][Directory] new RegistryDirectory for
service=%s", url.Key())
+ if registries, ok := registriesFromAttribute(url.SubURL); ok {
+ url.SetAttribute(constant.RegistriesConfigKey, registries)
+ return
+ }
+ url.SetAttribute(constant.RegistriesConfigKey,
map[string]*global.RegistryConfig{
+ constant.DefaultKey: global.DefaultRegistryConfig(),
+ })
+}
- // TODO: Temporary compatibility with old APIs, can be removed later
+func registriesFromAttribute(url *common.URL)
(map[string]*global.RegistryConfig, bool) {
+ if url == nil {
+ return nil, false
+ }
- // set application if not exist
- if _, ok := url.GetAttribute(constant.ApplicationKey); !ok {
- application := config.GetRootConfig().Application
- if application == nil {
- defaultAppConfig := global.DefaultApplicationConfig()
- url.SetAttribute(constant.ApplicationKey,
defaultAppConfig)
- } else {
- url.SetAttribute(constant.ApplicationKey, application)
- }
+ registriesRaw, ok := url.GetAttribute(constant.RegistriesConfigKey)
+ if !ok {
+ return nil, false
}
- // set registry if not exist
- if _, ok := url.GetAttribute(constant.RegistriesConfigKey); !ok {
- configRegistries := config.GetRootConfig().Registries
- if configRegistries == nil {
- defaultRegistryConfig := global.DefaultRegistryConfig()
- url.SetAttribute(constant.RegistriesConfigKey,
map[string]*global.RegistryConfig{
- constant.DefaultKey: defaultRegistryConfig,
- })
- } else {
- // convert config.RegistryConfig to
global.RegistryConfig
- globalRegistries :=
make(map[string]*global.RegistryConfig, len(configRegistries))
- for key, configRegistry := range configRegistries {
- globalRegistry := &global.RegistryConfig{
- Protocol:
configRegistry.Protocol,
- Timeout:
configRegistry.Timeout,
- Group: configRegistry.Group,
- Namespace:
configRegistry.Namespace,
- TTL: configRegistry.TTL,
- Address:
configRegistry.Address,
- Username:
configRegistry.Username,
- Password:
configRegistry.Password,
- Simplified:
configRegistry.Simplified,
- Preferred:
configRegistry.Preferred,
- Zone: configRegistry.Zone,
- Weight:
configRegistry.Weight,
- Params:
configRegistry.Params,
- RegistryType:
configRegistry.RegistryType,
- UseAsMetaReport:
configRegistry.UseAsMetaReport,
- UseAsConfigCenter:
configRegistry.UseAsConfigCenter,
- }
- globalRegistries[key] = globalRegistry
+
+ switch registries := registriesRaw.(type) {
+ case map[string]*global.RegistryConfig:
+ if registries != nil {
+ return registries, true
+ }
+ case map[string]global.RegistryConfig:
+ if registries != nil {
+ converted := make(map[string]*global.RegistryConfig,
len(registries))
+ for key, registryConfig := range registries {
+ registryConfigCopy := registryConfig
+ converted[key] = ®istryConfigCopy
}
- url.SetAttribute(constant.RegistriesConfigKey,
globalRegistries)
+ return converted, true
}
}
+ return nil, false
+}
+
+// NewRegistryDirectory will create a new RegistryDirectory
+func NewRegistryDirectory(url *common.URL, registry registry.Registry)
(directory.Directory, error) {
+ if url.SubURL == nil {
+ return nil, perrors.Errorf("url is invalid, suburl can not be
nil")
+ }
+ logger.Debugf("[Registry][Directory] new RegistryDirectory for
service=%s", url.Key())
+
+ commonConfig.EnsureApplicationAttribute(url, url.SubURL)
+ ensureRegistriesAttribute(url)
dir := &RegistryDirectory{
Directory: base.NewDirectory(url),
@@ -894,25 +891,13 @@ type consumerConfigurationListener struct {
func newConsumerConfigurationListener(dir *RegistryDirectory, url *common.URL)
*consumerConfigurationListener {
listener := &consumerConfigurationListener{directory: dir}
-
- // TODO: Temporary compatibility with old APIs, can be removed later
- application := config.GetRootConfig().Application
+ application := commonConfig.EnsureApplicationAttribute(url, url.SubURL)
listener.InitWith(
application.Name+constant.ConfiguratorSuffix,
listener,
extension.GetDefaultConfiguratorFunc(),
)
- if ApplicationConfRaw, ok := url.GetAttribute(constant.ApplicationKey);
ok {
- if ApplicationConfig, ok :=
ApplicationConfRaw.(*global.ApplicationConfig); ok {
- listener.InitWith(
-
ApplicationConfig.Name+constant.ConfiguratorSuffix,
- listener,
- extension.GetDefaultConfiguratorFunc(),
- )
- }
- }
-
return listener
}
diff --git a/registry/directory/directory_test.go
b/registry/directory/directory_test.go
index 89abbe5ce..443e0e170 100644
--- a/registry/directory/directory_test.go
+++ b/registry/directory/directory_test.go
@@ -26,6 +26,8 @@ import (
)
import (
+ gxset "github.com/dubbogo/gost/container/set"
+
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
@@ -36,8 +38,11 @@ import (
"dubbo.apache.org/dubbo-go/v3/cluster/cluster"
_ "dubbo.apache.org/dubbo-go/v3/cluster/router/tag"
"dubbo.apache.org/dubbo-go/v3/common"
+ common_cfg "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/config_center"
+ configparser "dubbo.apache.org/dubbo-go/v3/config_center/parser"
"dubbo.apache.org/dubbo-go/v3/global"
protocolbase "dubbo.apache.org/dubbo-go/v3/protocol/base"
"dubbo.apache.org/dubbo-go/v3/protocol/invocation"
@@ -61,6 +66,100 @@ func TestSubscribe_InvalidUrl(t *testing.T) {
require.Error(t, err)
}
+func TestNewRegistryDirectoryResolvesApplicationAttribute(t *testing.T) {
+ tests := []struct {
+ name string
+ configure func(registryURL *common.URL, subURL *common.URL)
+ wantName string
+ }{
+ {
+ name: "sub URL attribute",
+ configure: func(_ *common.URL, subURL *common.URL) {
+ subURL.SetAttribute(constant.ApplicationKey,
&global.ApplicationConfig{Name: "sub-attr-app"})
+ },
+ wantName: "sub-attr-app",
+ },
+ {
+ name: "registry URL param",
+ configure: func(registryURL *common.URL, _ *common.URL)
{
+ registryURL.SetParam(constant.ApplicationKey,
"registry-param-app")
+ },
+ wantName: "registry-param-app",
+ },
+ {
+ name: "sub URL param",
+ configure: func(_ *common.URL, subURL *common.URL) {
+ subURL.SetParam(constant.ApplicationKey,
"sub-param-app")
+ },
+ wantName: "sub-param-app",
+ },
+ {
+ name: "default",
+ configure: func(_ *common.URL, _ *common.URL) {},
+ wantName: constant.DefaultDubboApp,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ registryURL, subURL :=
newRegistryDirectoryAttributeTestURL(t)
+ tt.configure(registryURL, subURL)
+
+ newRegistryDirectoryForAttributeTest(t, registryURL)
+
+ applicationRaw, ok :=
registryURL.GetAttribute(constant.ApplicationKey)
+ require.True(t, ok)
+ application, ok :=
applicationRaw.(*global.ApplicationConfig)
+ require.True(t, ok)
+ assert.Equal(t, tt.wantName, application.Name)
+ })
+ }
+}
+
+func TestNewRegistryDirectoryCopiesRegistriesAttributeFromSubURL(t *testing.T)
{
+ registryURL, subURL := newRegistryDirectoryAttributeTestURL(t)
+ registries := map[string]*global.RegistryConfig{
+ "mock-registry": {Protocol: "mock", Address: "127.0.0.1:1111"},
+ }
+ subURL.SetAttribute(constant.RegistriesConfigKey, registries)
+
+ newRegistryDirectoryForAttributeTest(t, registryURL)
+
+ registriesRaw, ok :=
registryURL.GetAttribute(constant.RegistriesConfigKey)
+ require.True(t, ok)
+ assert.Equal(t, registries, registriesRaw)
+}
+
+func TestNewRegistryDirectoryUsesDefaultRegistriesAttribute(t *testing.T) {
+ registryURL, _ := newRegistryDirectoryAttributeTestURL(t)
+
+ newRegistryDirectoryForAttributeTest(t, registryURL)
+
+ registriesRaw, ok :=
registryURL.GetAttribute(constant.RegistriesConfigKey)
+ require.True(t, ok)
+ registries, ok := registriesRaw.(map[string]*global.RegistryConfig)
+ require.True(t, ok)
+ require.Contains(t, registries, constant.DefaultKey)
+ assert.Equal(t, global.DefaultRegistryConfig(),
registries[constant.DefaultKey])
+}
+
+func TestNewRegistryDirectoryConsumerListenerUsesResolvedApplicationName(t
*testing.T) {
+ env := common_cfg.GetEnvInstance()
+ previousDynamicConfiguration := env.GetDynamicConfiguration()
+ dynamicConfiguration := newRecordingDynamicConfiguration()
+ env.SetDynamicConfiguration(dynamicConfiguration)
+ t.Cleanup(func() {
+ env.SetDynamicConfiguration(previousDynamicConfiguration)
+ })
+
+ registryURL, subURL := newRegistryDirectoryAttributeTestURL(t)
+ subURL.SetParam(constant.ApplicationKey, "consumer-listener-app")
+
+ newRegistryDirectoryForAttributeTest(t, registryURL)
+
+ assert.Contains(t, dynamicConfiguration.keys,
"consumer-listener-app"+constant.ConfiguratorSuffix)
+}
+
func Test_Destroy(t *testing.T) {
registryDirectory, _ := normalRegistryDir()
@@ -387,6 +486,34 @@ func mustURL(t *testing.T, rawURL string) *common.URL {
return u
}
+func newRegistryDirectoryAttributeTestURL(t *testing.T) (*common.URL,
*common.URL) {
+ t.Helper()
+
+ registryURL, err := common.NewURL("mock://127.0.0.1:1111")
+ require.NoError(t, err)
+ subURL, err := common.NewURL(
+ "dubbo://127.0.0.1:20000/org.apache.dubbo-go.mockService",
+ common.WithParamsValue(constant.ClusterKey, "mock"),
+ common.WithParamsValue(constant.GroupKey, "group"),
+ common.WithParamsValue(constant.VersionKey, "1.0.0"),
+ )
+ require.NoError(t, err)
+ registryURL.SubURL = subURL
+ return registryURL, subURL
+}
+
+func newRegistryDirectoryForAttributeTest(t *testing.T, registryURL
*common.URL) *RegistryDirectory {
+ t.Helper()
+
+ mockRegistry, err := registry.NewMockRegistry(&common.URL{})
+ require.NoError(t, err)
+ dir, err := NewRegistryDirectory(registryURL, mockRegistry)
+ require.NoError(t, err)
+ registryDirectory := dir.(*RegistryDirectory)
+ t.Cleanup(registryDirectory.Destroy)
+ return registryDirectory
+}
+
func findInvokerURLByPort(t *testing.T, dir *RegistryDirectory, port string)
*common.URL {
t.Helper()
for _, invoker := range dir.snapshotCacheInvokers() {
@@ -680,3 +807,51 @@ func newRegistryDirectoryForConcurrencyTest(t *testing.T)
*RegistryDirectory {
require.NoError(t, err)
return dir.(*RegistryDirectory)
}
+
+type recordingDynamicConfiguration struct {
+ parser configparser.ConfigurationParser
+ keys []string
+}
+
+func newRecordingDynamicConfiguration() *recordingDynamicConfiguration {
+ return &recordingDynamicConfiguration{parser:
&configparser.DefaultConfigurationParser{}}
+}
+
+func (c *recordingDynamicConfiguration) Parser()
configparser.ConfigurationParser {
+ return c.parser
+}
+
+func (c *recordingDynamicConfiguration) SetParser(parser
configparser.ConfigurationParser) {
+ c.parser = parser
+}
+
+func (c *recordingDynamicConfiguration) AddListener(key string, _
config_center.ConfigurationListener, _ ...config_center.Option) {
+ c.keys = append(c.keys, key)
+}
+
+func (c *recordingDynamicConfiguration) RemoveListener(_ string, _
config_center.ConfigurationListener, _ ...config_center.Option) {
+}
+
+func (c *recordingDynamicConfiguration) GetProperties(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) GetRule(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) GetInternalProperty(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) PublishConfig(_, _, _ string) error {
+ return nil
+}
+
+func (c *recordingDynamicConfiguration) RemoveConfig(_, _ string) error {
+ return nil
+}
+
+func (c *recordingDynamicConfiguration) GetConfigKeysByGroup(_ string)
(*gxset.HashSet, error) {
+ return gxset.NewSet(), nil
+}
diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go
index c16923f62..ccd9491fb 100644
--- a/registry/protocol/protocol.go
+++ b/registry/protocol/protocol.go
@@ -33,9 +33,9 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common"
+ commonConfig "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
- "dubbo.apache.org/dubbo-go/v3/config"
"dubbo.apache.org/dubbo-go/v3/config_center"
_ "dubbo.apache.org/dubbo-go/v3/config_center/configurator"
"dubbo.apache.org/dubbo-go/v3/global"
@@ -129,6 +129,48 @@ func (proto *registryProtocol)
initConfigurationListeners(url *common.URL) {
proto.providerConfigurationListener =
newProviderConfigurationListener(proto.overrideListeners, url)
}
+// ensureShutdownAttributes resolves shutdown config by priority:
+// url attribute, fallbackURL attribute, then global default.
+// It writes the resolved config to url and, when missing, fallbackURL.
+func ensureShutdownAttributes(url *common.URL, fallbackURL *common.URL)
*global.ShutdownConfig {
+ if shutdownConfig, ok := shutdownFromAttribute(url); ok {
+ url.SetAttribute(constant.ShutdownConfigPrefix, shutdownConfig)
+ ensureFallbackShutdownAttribute(fallbackURL, shutdownConfig)
+ return shutdownConfig
+ }
+ if shutdownConfig, ok := shutdownFromAttribute(fallbackURL); ok {
+ url.SetAttribute(constant.ShutdownConfigPrefix, shutdownConfig)
+ return shutdownConfig
+ }
+
+ shutdownConfig := global.DefaultShutdownConfig()
+ url.SetAttribute(constant.ShutdownConfigPrefix, shutdownConfig)
+ ensureFallbackShutdownAttribute(fallbackURL, shutdownConfig)
+ return shutdownConfig
+}
+
+func ensureFallbackShutdownAttribute(url *common.URL, shutdownConfig
*global.ShutdownConfig) {
+ if url == nil {
+ return
+ }
+ if _, ok := url.GetAttribute(constant.ShutdownConfigPrefix); !ok {
+ url.SetAttribute(constant.ShutdownConfigPrefix, shutdownConfig)
+ }
+}
+
+func shutdownFromAttribute(url *common.URL) (*global.ShutdownConfig, bool) {
+ if url == nil {
+ return nil, false
+ }
+
+ shutdownRaw, ok := url.GetAttribute(constant.ShutdownConfigPrefix)
+ if !ok {
+ return nil, false
+ }
+ shutdownConfig, ok := shutdownRaw.(*global.ShutdownConfig)
+ return shutdownConfig, ok && shutdownConfig != nil
+}
+
// GetRegistries returns all underlying registry instances.
func (proto *registryProtocol) GetRegistries() []registry.Registry {
var rs []registry.Registry
@@ -188,25 +230,8 @@ func (proto *registryProtocol) Export(originInvoker
base.Invoker) base.Exporter
registryUrl := getRegistryUrl(originInvoker)
providerUrl := getProviderUrl(originInvoker)
- // Copy ShutdownConfig from providerUrl to registryUrl if registryUrl
doesn't have it
- // (server layer sets it in ivkURL, which becomes providerUrl here)
- if _, ok := registryUrl.GetAttribute(constant.ShutdownConfigPrefix);
!ok {
- if config.GetShutDown() == nil {
- // Fallback to default if config package doesn't have
one
- registryUrl.SetAttribute(constant.ShutdownConfigPrefix,
global.DefaultShutdownConfig())
- }
- }
-
- // Copy ApplicationKey from providerUrl to registryUrl if registryUrl
doesn't have it
- // ApplicationKey is passed as URL parameter (application name string)
- // (server layer sets it in ivkURL, which becomes providerUrl here)
- if _, ok := registryUrl.GetAttribute(constant.ApplicationKey); !ok {
- // Fallback to config package for old API compatibility
- if config.GetRootConfig().Application == nil {
- // Use default application name
- registryUrl.SetAttribute(constant.ApplicationKey,
global.DefaultApplicationConfig())
- }
- }
+ ensureShutdownAttributes(registryUrl, providerUrl)
+ commonConfig.EnsureApplicationAttribute(registryUrl, providerUrl)
proto.once.Do(func() {
proto.initConfigurationListeners(providerUrl)
@@ -296,49 +321,41 @@ func registerServiceMap(invoker base.Invoker) error {
// such as
dubbo://:20000/org.apache.dubbo.UserProvider?bean.name=UserProvider&cluster=failfast...
id := providerUrl.GetParam(constant.BeanNameKey, "")
- // TODO: Temporary compatibility with old APIs, can be removed later
-
- providerConfig := config.GetProviderConfig()
+ providerConfRaw, ok :=
providerUrl.GetAttribute(constant.ProviderConfigKey)
+ if !ok {
+ return perrors.Errorf("reExport can not get provider config
from url attribute %s", constant.ProviderConfigKey)
+ }
+ providerConf, ok := providerConfRaw.(*global.ProviderConfig)
+ if !ok || providerConf == nil {
+ return perrors.Errorf("reExport got illegal provider config
from url attribute %s", constant.ProviderConfigKey)
+ }
- if providerConfig != nil {
- if serviceConfig := providerConfig.Services[id]; serviceConfig
!= nil {
- rpcService := config.GetProviderService(id)
- if rpcService == nil {
- return perrors.New("reExport can not get
RPCService")
- }
+ serviceConf := providerConf.Services[id]
+ if serviceConf == nil {
+ return perrors.Errorf("reExport can not get service config %q
from provider config", id)
+ }
- _, err :=
common.ServiceMap.Register(serviceConfig.Interface,
- serviceConfig.ProtocolIDs[0],
serviceConfig.Group,
- serviceConfig.Version, rpcService)
- if err != nil {
- s := "reExport can not re register ServiceMap.
Error message is " + err.Error()
- return perrors.New(s)
- }
- return nil
- }
+ rpcService, ok := providerUrl.GetAttribute(constant.RpcServiceKey)
+ if !ok || rpcService == nil {
+ return perrors.Errorf("reExport can not get RPCService from url
attribute %s", constant.RpcServiceKey)
}
- if providerConfRaw, ok :=
providerUrl.GetAttribute(constant.ProviderConfigKey); ok {
- if providerConf, ok :=
providerConfRaw.(*global.ProviderConfig); ok {
- if serviceConf, ok := providerConf.Services[id]; ok {
- if serviceConf == nil {
- return perrors.New("reExport can not
get RPCService")
- }
- if rpcService, ok :=
providerUrl.GetAttribute(constant.RpcServiceKey); ok {
- _, err :=
common.ServiceMap.Register(serviceConf.Interface,
- serviceConf.ProtocolIDs[0],
serviceConf.Group,
- serviceConf.Version, rpcService)
- if err != nil {
- s := "reExport can not re
register ServiceMap. Error message is " + err.Error()
- return perrors.New(s)
- }
- return nil
- }
- }
- }
+ protocol := providerUrl.Protocol
+ if len(serviceConf.ProtocolIDs) > 0 && serviceConf.ProtocolIDs[0] != ""
{
+ protocol = serviceConf.ProtocolIDs[0]
+ }
+ if protocol == "" {
+ return perrors.New("reExport can not get protocol")
}
- return perrors.New("reExport can not get serviceConfig of config")
+ _, err := common.ServiceMap.Register(serviceConf.Interface,
+ protocol, serviceConf.Group,
+ serviceConf.Version, rpcService)
+ if err != nil {
+ s := "reExport can not re register ServiceMap. Error message is
" + err.Error()
+ return perrors.New(s)
+ }
+ return nil
}
type overrideSubscribeListener struct {
@@ -488,24 +505,10 @@ func (proto *registryProtocol) Destroy() {
// close all protocol server after consumerUpdateWait +
stepTimeout(max time wait during
// waitAndAcceptNewRequests procedure)
go func() {
- if configShutdown := config.GetShutDown();
configShutdown != nil {
- <-time.After(configShutdown.GetStepTimeout() +
configShutdown.GetConsumerUpdateWaitTime())
- exporter.UnExport()
- proto.bounds.Delete(key)
- return
- }
-
- if shutdownConfRaw, ok :=
exporter.registerUrl.GetAttribute(constant.ShutdownConfigPrefix); ok {
- if shutdownConfig, ok :=
shutdownConfRaw.(*global.ShutdownConfig); ok {
- stepTimeout, _ :=
time.ParseDuration(shutdownConfig.StepTimeout)
- consumerUpdateWaitTime, _ :=
time.ParseDuration(shutdownConfig.ConsumerUpdateWaitTime)
- <-time.After(stepTimeout +
consumerUpdateWaitTime)
- exporter.UnExport()
- proto.bounds.Delete(key)
- return
- }
+ wait := destroyWaitDuration(exporter.registerUrl)
+ if wait > 0 {
+ <-time.After(wait)
}
-
exporter.UnExport()
proto.bounds.Delete(key)
}()
@@ -527,6 +530,29 @@ func (proto *registryProtocol) Destroy() {
}
+func destroyWaitDuration(url *common.URL) time.Duration {
+ if url == nil {
+ return 0
+ }
+ shutdownConfRaw, ok := url.GetAttribute(constant.ShutdownConfigPrefix)
+ if !ok {
+ return 0
+ }
+ shutdownConfig, ok := shutdownConfRaw.(*global.ShutdownConfig)
+ if !ok || shutdownConfig == nil {
+ return 0
+ }
+ stepTimeout, err := time.ParseDuration(shutdownConfig.StepTimeout)
+ if err != nil {
+ stepTimeout = 0
+ }
+ consumerUpdateWaitTime, err :=
time.ParseDuration(shutdownConfig.ConsumerUpdateWaitTime)
+ if err != nil {
+ consumerUpdateWaitTime = 0
+ }
+ return stepTimeout + consumerUpdateWaitTime
+}
+
// UnregisterRegistries only unregisters exported services from registries
during graceful shutdown.
// Protocol servers keep running until the later destroy phase.
func (proto *registryProtocol) UnregisterRegistries() {
@@ -648,25 +674,13 @@ type providerConfigurationListener struct {
func newProviderConfigurationListener(overrideListeners *sync.Map, url
*common.URL) *providerConfigurationListener {
listener := &providerConfigurationListener{}
listener.overrideListeners = overrideListeners
-
- // TODO: Temporary compatibility with old APIs, can be removed later
- application := config.GetRootConfig().Application
+ application := commonConfig.EnsureApplicationAttribute(url)
listener.InitWith(
application.Name+constant.ConfiguratorSuffix,
listener,
extension.GetDefaultConfiguratorFunc(),
)
- if ApplicationConfRaw, ok := url.GetAttribute(constant.ApplicationKey);
ok {
- if ApplicationConfig, ok :=
ApplicationConfRaw.(*global.ApplicationConfig); ok {
- listener.InitWith(
-
ApplicationConfig.Name+constant.ConfiguratorSuffix,
- listener,
- extension.GetDefaultConfiguratorFunc(),
- )
- }
- }
-
return listener
}
diff --git a/registry/protocol/protocol_test.go
b/registry/protocol/protocol_test.go
index bef134d97..6da1720cd 100644
--- a/registry/protocol/protocol_test.go
+++ b/registry/protocol/protocol_test.go
@@ -38,6 +38,7 @@ import (
"dubbo.apache.org/dubbo-go/v3/common/extension"
"dubbo.apache.org/dubbo-go/v3/config_center"
"dubbo.apache.org/dubbo-go/v3/config_center/configurator"
+ configparser "dubbo.apache.org/dubbo-go/v3/config_center/parser"
"dubbo.apache.org/dubbo-go/v3/global"
"dubbo.apache.org/dubbo-go/v3/protocol/base"
"dubbo.apache.org/dubbo-go/v3/protocol/protocolwrapper"
@@ -168,7 +169,7 @@ func exporterNormal(t *testing.T, regProtocol
*registryProtocol) *common.URL {
common.WithParamsValue(constant.VersionKey, "1.0.0"),
common.WithParamsValue(constant.BeanNameKey,
"org.apache.dubbo-go.mockService"),
common.WithAttribute(constant.ApplicationKey,
applicationConfig),
- common.WithAttribute(constant.ProviderConfigPrefix,
providerConfig),
+ common.WithAttribute(constant.ProviderConfigKey,
providerConfig),
common.WithAttribute(constant.RpcServiceKey, mockRPCService),
)
@@ -186,6 +187,83 @@ func TestExporter(t *testing.T) {
exporterNormal(t, regProtocol)
}
+func TestExportCopiesProviderAttributesToRegistryURL(t *testing.T) {
+ extension.SetRegistry("mock", registry.NewMockRegistry)
+ extension.SetProtocol(protocolwrapper.FILTER,
protocolwrapper.NewMockProtocolFilter)
+
+ regProtocol := newRegistryProtocol()
+ shutdownConfig := &global.ShutdownConfig{
+ StepTimeout: "1ms",
+ ConsumerUpdateWaitTime: "1ms",
+ }
+ applicationConfig := &global.ApplicationConfig{Name:
"provider-application"}
+
+ registryURL, err := common.NewURL("mock://127.0.0.1:1111")
+ require.NoError(t, err)
+ providerURL, err := common.NewURL(
+ "dubbo://127.0.0.1:20000/org.apache.dubbo-go.mockService",
+ common.WithAttribute(constant.ShutdownConfigPrefix,
shutdownConfig),
+ common.WithAttribute(constant.ApplicationKey,
applicationConfig),
+ )
+ require.NoError(t, err)
+ registryURL.SubURL = providerURL
+
+ exporter := regProtocol.Export(base.NewBaseInvoker(registryURL))
+ require.NotNil(t, exporter)
+ t.Cleanup(regProtocol.Destroy)
+
+ shutdownRaw, ok :=
registryURL.GetAttribute(constant.ShutdownConfigPrefix)
+ require.True(t, ok)
+ assert.Same(t, shutdownConfig, shutdownRaw)
+ applicationRaw, ok := registryURL.GetAttribute(constant.ApplicationKey)
+ require.True(t, ok)
+ assert.Same(t, applicationConfig, applicationRaw)
+
+ wrapper := exporter.(*exporterChangeableWrapper)
+ registeredShutdownRaw, ok :=
wrapper.registerUrl.GetAttribute(constant.ShutdownConfigPrefix)
+ require.True(t, ok)
+ assert.Same(t, shutdownConfig, registeredShutdownRaw)
+}
+
+func TestRegisterServiceMapUsesURLAttributesAndProviderProtocolFallback(t
*testing.T) {
+ const (
+ interfaceName = "org.apache.dubbo-go.attributeOnlyService"
+ protocolName = "attribute-protocol"
+ serviceID = "attribute-service"
+ )
+
+ serviceConfig := &global.ServiceConfig{
+ Interface: interfaceName,
+ Group: "group",
+ Version: "1.0.0",
+ }
+ serviceKey := common.ServiceKey(interfaceName, serviceConfig.Group,
serviceConfig.Version)
+ _ = common.ServiceMap.UnRegister(interfaceName, protocolName,
serviceKey)
+ t.Cleanup(func() {
+ _ = common.ServiceMap.UnRegister(interfaceName, protocolName,
serviceKey)
+ })
+
+ providerConfig := &global.ProviderConfig{
+ Services: map[string]*global.ServiceConfig{
+ serviceID: serviceConfig,
+ },
+ }
+ registryURL, err := common.NewURL("mock://127.0.0.1:1111")
+ require.NoError(t, err)
+ providerURL, err := common.NewURL(
+ protocolName+"://127.0.0.1:20000/"+interfaceName,
+ common.WithParamsValue(constant.BeanNameKey, serviceID),
+ common.WithAttribute(constant.ProviderConfigKey,
providerConfig),
+ common.WithAttribute(constant.RpcServiceKey, &MockRPCService{}),
+ )
+ require.NoError(t, err)
+ registryURL.SubURL = providerURL
+
+ err = registerServiceMap(base.NewBaseInvoker(registryURL))
+ require.NoError(t, err)
+ assert.NotNil(t, common.ServiceMap.GetService(protocolName,
interfaceName, serviceConfig.Group, serviceConfig.Version))
+}
+
func TestMultiRegAndMultiProtoExporter(t *testing.T) {
regProtocol := newRegistryProtocol()
exporterNormal(t, regProtocol)
@@ -256,7 +334,7 @@ func TestOneRegAndProtoExporter(t *testing.T) {
common.WithParamsValue(constant.VersionKey, "1.0.0"),
common.WithParamsValue(constant.BeanNameKey,
"org.apache.dubbo-go.mockService"),
common.WithAttribute(constant.ApplicationKey,
applicationConfig),
- common.WithAttribute(constant.ProviderConfigPrefix,
providerConfig),
+ common.WithAttribute(constant.ProviderConfigKey,
providerConfig),
common.WithAttribute(constant.RpcServiceKey, mockRPCService),
)
@@ -321,6 +399,77 @@ func TestDestroyUnsubscribesOverrideListener(t *testing.T)
{
assert.Equal(t, 0,
registry.CountSyncMapEntries(regProtocol.serviceConfigurationListeners))
}
+func TestDestroyUsesRegisteredURLShutdownAttribute(t *testing.T) {
+ extension.SetRegistry("destroy-recording", registry.NewMockRegistry)
+
+ regProtocol := newRegistryProtocol()
+ regProtocol.overrideListeners = &sync.Map{}
+ regProtocol.serviceConfigurationListeners = &sync.Map{}
+
+ registryURL, err := common.NewURL("destroy-recording://127.0.0.1:1111")
+ require.NoError(t, err)
+ providerURL, err :=
common.NewURL("dubbo://127.0.0.1:20000/org.apache.dubbo-go.mockService")
+ require.NoError(t, err)
+ registryURL.SubURL = providerURL
+ originInvoker := base.NewBaseInvoker(registryURL)
+
+ unexported := make(chan time.Time, 1)
+ exporter := newExporterChangeableWrapper(originInvoker,
&recordingExporter{
+ invoker: base.NewBaseInvoker(providerURL),
+ unexported: unexported,
+ })
+ registerURL := providerURL.Clone()
+ registerURL.SetAttribute(constant.ShutdownConfigPrefix,
&global.ShutdownConfig{
+ StepTimeout: "30ms",
+ ConsumerUpdateWaitTime: "30ms",
+ })
+ exporter.SetRegisterUrl(registerURL)
+ exporter.SetSubscribeUrl(getSubscribedOverrideUrl(providerURL))
+ regProtocol.bounds.Store(getCacheKey(originInvoker), exporter)
+
+ regProtocol.Destroy()
+
+ select {
+ case <-unexported:
+ require.Fail(t, "exporter unexported before shutdown wait
elapsed")
+ case <-time.After(20 * time.Millisecond):
+ }
+
+ select {
+ case <-unexported:
+ case <-time.After(time.Second):
+ require.Fail(t, "exporter was not unexported")
+ }
+ assert.Eventually(t, func() bool {
+ return registry.CountSyncMapEntries(regProtocol.bounds) == 0
+ }, time.Second, 10*time.Millisecond)
+}
+
+func TestNewProviderConfigurationListenerUsesResolvedApplicationName(t
*testing.T) {
+ env := common_cfg.GetEnvInstance()
+ previousDynamicConfiguration := env.GetDynamicConfiguration()
+ dynamicConfiguration := newRecordingDynamicConfiguration()
+ env.SetDynamicConfiguration(dynamicConfiguration)
+ t.Cleanup(func() {
+ env.SetDynamicConfiguration(previousDynamicConfiguration)
+ })
+
+ providerURL, err := common.NewURL(
+ "dubbo://127.0.0.1:20000/org.apache.dubbo-go.mockService",
+ common.WithParamsValue(constant.ApplicationKey,
"provider-listener-app"),
+ )
+ require.NoError(t, err)
+
+ newProviderConfigurationListener(&sync.Map{}, providerURL)
+
+ assert.Contains(t, dynamicConfiguration.keys,
"provider-listener-app"+constant.ConfiguratorSuffix)
+ applicationRaw, ok := providerURL.GetAttribute(constant.ApplicationKey)
+ require.True(t, ok)
+ application, ok := applicationRaw.(*global.ApplicationConfig)
+ require.True(t, ok)
+ assert.Equal(t, "provider-listener-app", application.Name)
+}
+
func TestReExportReplacesConfigurationListeners(t *testing.T) {
extension.SetDefaultConfigurator(configurator.NewMockConfigurator)
@@ -412,7 +561,7 @@ func TestExportWithServiceConfig(t *testing.T) {
extension.SetDefaultConfigurator(configurator.NewMockConfigurator)
ccUrl, _ := common.NewURL("mock://127.0.0.1:1111")
dc, _ :=
(&config_center.MockDynamicConfigurationFactory{}).GetDynamicConfiguration(ccUrl)
- // Use common/config (not dubbo.apache.org/dubbo-go/v3/config)
+ // Use the common config environment, not the legacy config package.
common_cfg.GetEnvInstance().SetDynamicConfiguration(dc)
regProtocol := newRegistryProtocol()
@@ -437,7 +586,7 @@ func TestExportWithApplicationConfig(t *testing.T) {
extension.SetDefaultConfigurator(configurator.NewMockConfigurator)
ccUrl, _ := common.NewURL("mock://127.0.0.1:1111")
dc, _ :=
(&config_center.MockDynamicConfigurationFactory{}).GetDynamicConfiguration(ccUrl)
- // Use common/config (not dubbo.apache.org/dubbo-go/v3/config)
+ // Use the common config environment, not the legacy config package.
common_cfg.GetEnvInstance().SetDynamicConfiguration(dc)
regProtocol := newRegistryProtocol()
@@ -487,6 +636,19 @@ func (r *unsubscribeRecordingRegistry) UnSubscribe(url
*common.URL, notifyListen
return nil
}
+type recordingExporter struct {
+ invoker base.Invoker
+ unexported chan<- time.Time
+}
+
+func (e *recordingExporter) GetInvoker() base.Invoker {
+ return e.invoker
+}
+
+func (e *recordingExporter) UnExport() {
+ e.unexported <- time.Now()
+}
+
func newRegistryProtocolWithSubscribedExporter(
t *testing.T,
) (*registryProtocol, *unsubscribeRecordingRegistry, base.Invoker,
*exporterChangeableWrapper, *overrideSubscribeListener) {
@@ -544,3 +706,51 @@ func (m *MockRPCService) MockMethod(arg1, arg2 string)
error {
func (m *MockRPCService) Reference() string {
return "org.apache.dubbo-go.mockService"
}
+
+type recordingDynamicConfiguration struct {
+ parser configparser.ConfigurationParser
+ keys []string
+}
+
+func newRecordingDynamicConfiguration() *recordingDynamicConfiguration {
+ return &recordingDynamicConfiguration{parser:
&configparser.DefaultConfigurationParser{}}
+}
+
+func (c *recordingDynamicConfiguration) Parser()
configparser.ConfigurationParser {
+ return c.parser
+}
+
+func (c *recordingDynamicConfiguration) SetParser(parser
configparser.ConfigurationParser) {
+ c.parser = parser
+}
+
+func (c *recordingDynamicConfiguration) AddListener(key string, _
config_center.ConfigurationListener, _ ...config_center.Option) {
+ c.keys = append(c.keys, key)
+}
+
+func (c *recordingDynamicConfiguration) RemoveListener(_ string, _
config_center.ConfigurationListener, _ ...config_center.Option) {
+}
+
+func (c *recordingDynamicConfiguration) GetProperties(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) GetRule(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) GetInternalProperty(_ string, _
...config_center.Option) (string, error) {
+ return "", nil
+}
+
+func (c *recordingDynamicConfiguration) PublishConfig(_, _, _ string) error {
+ return nil
+}
+
+func (c *recordingDynamicConfiguration) RemoveConfig(_, _ string) error {
+ return nil
+}
+
+func (c *recordingDynamicConfiguration) GetConfigKeysByGroup(_ string)
(*gxset.HashSet, error) {
+ return gxset.NewSet(), nil
+}
