Repository: apache/dubbo-go

Issue:
应用级服务发现在分组订阅、泛化下测试
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

-->


**What happened**:

**What you expected to happen**:

**How to reproduce it (as minimally and precisely as possible)**:

**Anything else we need to know?**:


Hints from issue comments:
@1kasa: @jiajia@Alanxtl: @tianhao

Solution patch:
diff --git a/registry/servicediscovery/service_discovery_registry.go b/registry/servicediscovery/service_discovery_registry.go
index 3e5c50ec..2fe06d29 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -154,7 +154,9 @@ func (s *serviceDiscoveryRegistry) UnSubscribe(url *common.URL, listener registr
 	// FIXME ServiceNames.String() is not good
 	serviceNamesKey := services.String()
 	l := s.serviceListeners[serviceNamesKey]
-	l.RemoveListener(url.ServiceKey())
+	if l != nil {
+		l.RemoveListener(url.ServiceKey())
+	}
 	s.stopListen(url)
 	err := s.serviceNameMapping.Remove(url)
 	if err != nil {



Test patch:
diff --git a/registry/servicediscovery/service_discovery_registry_test.go b/registry/servicediscovery/service_discovery_registry_test.go
index 29b0e6f9..c466c042 100644
--- a/registry/servicediscovery/service_discovery_registry_test.go
+++ b/registry/servicediscovery/service_discovery_registry_test.go
@@ -17,199 +17,265 @@
 
 package servicediscovery
 
-//import (
-//	"testing"
-//)
-//
-//import (
-//	gxset "github.com/dubbogo/gost/container/set"
-//	"github.com/dubbogo/gost/hash/page"
-//	"github.com/stretchr/testify/assert"
-//)
-//
-//import (
-//	"dubbo.apache.org/dubbo-go/v3/common"
-//	"dubbo.apache.org/dubbo-go/v3/common/extension"
-//	"dubbo.apache.org/dubbo-go/v3/config"
-//	"dubbo.apache.org/dubbo-go/v3/metadata/mapping"
-//	"dubbo.apache.org/dubbo-go/v3/metadata/service"
-//	"dubbo.apache.org/dubbo-go/v3/registry"
-//)
-//
-//var (
-//	serviceInterface = "org.apache.dubbo.metadata.MetadataService"
-//	group            = "dubbo-provider"
-//	version          = "1.0.0"
-//)
-//
-//func TestServiceDiscoveryRegistry_Register(t *testing.T) {
-//	config.GetApplicationConfig().MetadataType = "mock"
-//	extension.SetLocalMetadataService("mock", func() (service service.MetadataService, err error) {
-//		service = &mockMetadataService{}
-//		return
-//	})
-//
-//	extension.SetServiceDiscovery("mock", func(name string) (discovery registry.ServiceDiscovery, err error) {
-//		return &mockServiceDiscovery{}, nil
-//	})
-//
-//	extension.SetGlobalServiceNameMapping(func() mapping.ServiceNameMapping {
-//		return mapping.NewMockServiceNameMapping()
-//	})
-//
-//	config.GetBaseConfig().ServiceDiscoveries["mock"] = &config.ServiceDiscoveryConfig{
-//		Protocol: "mock",
-//	}
-//	registryURL, _ := common.NewURL("service-discovery://localhost:12345",
-//		common.WithParamsValue("service_discovery", "mock"),
-//		common.WithParamsValue("subscribed-services", "a, b , c,d,e ,"))
-//	url, _ := common.NewURL("dubbo://192.168.0.102:20880/" + serviceInterface +
-//		"?&application=" + group +
-//		"&interface=" + serviceInterface +
-//		"&group=" + group +
-//		"&version=" + version +
-//		"&service_discovery=mock" +
-//		"&methods=getAllServiceKeys,getServiceRestMetadata,getExportedURLs,getAllExportedURLs" +
-//		"&side=provider")
-//	registry, err := newServiceDiscoveryRegistry(registryURL)
-//	assert.Nil(t, err)
-//	assert.NotNil(t, registry)
-//	err = registry.Register(url)
-//	assert.NoError(t, err)
-//}
-//
-//type mockServiceDiscovery struct{}
-//
-//func (m *mockServiceDiscovery) String() string {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) Destroy() error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) Register(registry.ServiceInstance) error {
-//	return nil
-//}
-//
-//func (m *mockServiceDiscovery) Update(registry.ServiceInstance) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) Unregister(registry.ServiceInstance) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetDefaultPageSize() int {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetServices() *gxset.HashSet {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetInstances(string) []registry.ServiceInstance {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetInstancesByPage(string, int, int) gxpage.Pager {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetHealthyInstancesByPage(string, int, int, bool) gxpage.Pager {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) GetRequestInstances([]string, int, int) map[string]gxpage.Pager {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) AddListener(registry.ServiceInstancesChangedListener) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) DispatchEventByServiceName(string) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) DispatchEventForInstances(string, []registry.ServiceInstance) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockServiceDiscovery) DispatchEvent(*registry.ServiceInstancesChangedEvent) error {
-//	panic("implement me")
-//}
-//
-//type mockMetadataService struct{}
-//
-//func (m *mockMetadataService) GetExportedURLs(string, string, string, string) ([]*common.URL, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetMetadataInfo(revision string) (*common.MetadataInfo, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetExportedServiceURLs() []*common.URL {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetMetadataServiceURL() *common.URL {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) SetMetadataServiceURL(url *common.URL) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) Reference() string {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) ServiceName() (string, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) ExportURL(*common.URL) (bool, error) {
-//	return true, nil
-//}
-//
-//func (m *mockMetadataService) UnexportURL(*common.URL) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) SubscribeURL(*common.URL) (bool, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) UnsubscribeURL(*common.URL) error {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) PublishServiceDefinition(*common.URL) error {
-//	return nil
-//}
-//
-//func (m *mockMetadataService) MethodMapper() map[string]string {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetSubscribedURLs() ([]*common.URL, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetServiceDefinition(string, string, string) (string, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) GetServiceDefinitionByServiceKey(string) (string, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) RefreshMetadata(string, string) (bool, error) {
-//	panic("implement me")
-//}
-//
-//func (m *mockMetadataService) Version() (string, error) {
-//	panic("implement me")
-//}
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+)
+
+import (
+	gxset "github.com/dubbogo/gost/container/set"
+	gxpage "github.com/dubbogo/gost/hash/page"
+
+	"github.com/stretchr/testify/assert"
+)
+
+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/global"
+	"dubbo.apache.org/dubbo-go/v3/metadata"
+	"dubbo.apache.org/dubbo-go/v3/metadata/mapping"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+	"dubbo.apache.org/dubbo-go/v3/protocol/base"
+	"dubbo.apache.org/dubbo-go/v3/proxy"
+	"dubbo.apache.org/dubbo-go/v3/registry"
+)
+
+const (
+	testInterface   = "org.apache.dubbo.test.TestService"
+	testGroup       = "test-group"
+	testApp         = "test-app"
+	testRegistryURL = "service-discovery://localhost:12345"
+)
+
+// TestServiceDiscoveryRegistryRegister verifies the registration process.
+func TestServiceDiscoveryRegistryRegister(t *testing.T) {
+	mockSD, mockMapping := setupEnvironment(t)
+	regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+
+	registryURL, err := common.NewURL(testRegistryURL,
+		common.WithParamsValue(constant.RegistryKey, "mock"),
+		common.WithParamsValue(constant.RegistryIdKey, regID))
+	assert.NoError(t, err)
+
+	reg, err := newServiceDiscoveryRegistry(registryURL)
+	assert.NoError(t, err)
+
+	providerURL, _ := common.NewURL("dubbo://127.0.0.1:20880/",
+		common.WithParamsValue(constant.ApplicationKey, testApp),
+		common.WithInterface(testInterface),
+		common.WithParamsValue(constant.SideKey, constant.SideProvider),
+	)
+
+	// Interface Mapping
+	err = reg.Register(providerURL)
+	assert.NoError(t, err)
+	assert.True(t, mockMapping.mapCalled, "ServiceNameMapping.Map should be called")
+
+	// Instance Registration
+	sdReg, ok := reg.(*serviceDiscoveryRegistry)
+	assert.True(t, ok)
+
+	err = sdReg.RegisterService()
+	assert.NoError(t, err)
+
+	assert.True(t, mockSD.registerCalled, "ServiceDiscovery.Register should be called")
+
+	if mockSD.capturedInstance != nil {
+		assert.Equal(t, testApp, mockSD.capturedInstance.GetServiceName())
+		assert.Equal(t, "127.0.0.1", mockSD.capturedInstance.GetHost())
+		assert.Equal(t, 20880, mockSD.capturedInstance.GetPort())
+		assert.Equal(t, "mock", mockSD.capturedInstance.GetMetadata()[constant.MetadataStorageTypePropertyName])
+	}
+}
+
+// TestServiceDiscoveryRegistrySubscribe verifies the subscription flow.
+func TestServiceDiscoveryRegistrySubscribe(t *testing.T) {
+	mockSD, mockMapping := setupEnvironment(t)
+	mockMapping.data[testInterface] = gxset.NewSet(testApp)
+
+	registryURL, _ := common.NewURL(testRegistryURL,
+		common.WithParamsValue(constant.RegistryKey, "mock"))
+
+	reg, err := newServiceDiscoveryRegistry(registryURL)
+	assert.NoError(t, err)
+
+	consumerURL, _ := common.NewURL("dubbo://127.0.0.1:20000/",
+		common.WithInterface(testInterface),
+		common.WithParamsValue(constant.GroupKey, testGroup),
+		common.WithParamsValue(constant.SideKey, constant.SideConsumer),
+	)
+
+	mockSD.wg.Add(1)
+	err = reg.Subscribe(consumerURL, &mockNotifyListener{})
+	assert.NoError(t, err)
+
+	assert.True(t, mockMapping.getCalled)
+	assert.Equal(t, testApp, mockSD.capturedAppName)
+
+	mockSD.wg.Wait()
+	assert.True(t, mockSD.listenerAdded)
+}
+
+// TestServiceDiscoveryRegistryUnSubscribe verifies the unsubscription logic.
+func TestServiceDiscoveryRegistryUnSubscribe(t *testing.T) {
+	mockSD, mockMapping := setupEnvironment(t)
+	mockMapping.data[testInterface] = gxset.NewSet(testApp)
+
+	registryURL, _ := common.NewURL(testRegistryURL,
+		common.WithParamsValue(constant.RegistryKey, "mock"))
+
+	reg, err := newServiceDiscoveryRegistry(registryURL)
+	assert.NoError(t, err)
+
+	consumerURL, _ := common.NewURL("dubbo://127.0.0.1:20000/",
+		common.WithInterface(testInterface),
+		common.WithParamsValue(constant.SideKey, constant.SideConsumer),
+	)
+
+	mockSD.wg.Add(1)
+	_ = reg.Subscribe(consumerURL, &mockNotifyListener{})
+	mockSD.wg.Wait()
+
+	err = reg.UnSubscribe(consumerURL, &mockNotifyListener{})
+	assert.NoError(t, err)
+	assert.True(t, mockMapping.removeCalled)
+}
+
+// setupEnvironment initializes the test environment.
+func setupEnvironment(t *testing.T) (*mockServiceDiscovery, *mockServiceNameMapping) {
+	appConfig := global.DefaultApplicationConfig()
+	appConfig.Name = testApp
+
+	mockSD := &mockServiceDiscovery{}
+	mockMapping := &mockServiceNameMapping{data: make(map[string]*gxset.HashSet)}
+
+	extension.SetProtocol("dubbo", func() protocol.Protocol { return &mockProtocol{} })
+	extension.SetProxyFactory("default", func(options ...proxy.Option) proxy.ProxyFactory { return &mockProxyFactory{} })
+	extension.SetRegistry("service-discovery", newServiceDiscoveryRegistry)
+	extension.SetServiceDiscovery("mock", func(url *common.URL) (registry.ServiceDiscovery, error) {
+		return mockSD, nil
+	})
+	extension.SetGlobalServiceNameMapping(func() mapping.ServiceNameMapping {
+		return mockMapping
+	})
+
+	opts := metadata.NewOptions(metadata.WithMetadataType("mock"))
+	_ = opts.Init()
+	return mockSD, mockMapping
+}
+
+type mockServiceDiscovery struct {
+	wg               sync.WaitGroup
+	registerCalled   bool
+	listenerAdded    bool
+	capturedAppName  string
+	capturedInstance registry.ServiceInstance
+}
+
+func (m *mockServiceDiscovery) String() string { return "mock" }
+func (m *mockServiceDiscovery) Destroy() error { return nil }
+func (m *mockServiceDiscovery) Register(inst registry.ServiceInstance) error {
+	m.registerCalled = true
+	m.capturedInstance = inst
+	return nil
+}
+func (m *mockServiceDiscovery) Update(inst registry.ServiceInstance) error     { return nil }
+func (m *mockServiceDiscovery) Unregister(inst registry.ServiceInstance) error { return nil }
+func (m *mockServiceDiscovery) GetDefaultPageSize() int                        { return 10 }
+func (m *mockServiceDiscovery) GetServices() *gxset.HashSet                    { return gxset.NewSet("mock-service") }
+func (m *mockServiceDiscovery) GetInstances(name string) []registry.ServiceInstance {
+	m.capturedAppName = name
+	return []registry.ServiceInstance{}
+}
+func (m *mockServiceDiscovery) GetInstancesByPage(string, int, int) gxpage.Pager { return nil }
+func (m *mockServiceDiscovery) GetHealthyInstancesByPage(string, int, int, bool) gxpage.Pager {
+	return nil
+}
+func (m *mockServiceDiscovery) GetRequestInstances([]string, int, int) map[string]gxpage.Pager {
+	return nil
+}
+func (m *mockServiceDiscovery) AddListener(registry.ServiceInstancesChangedListener) error {
+	defer m.wg.Done()
+	m.listenerAdded = true
+	return nil
+}
+
+type mockServiceNameMapping struct {
+	data          map[string]*gxset.HashSet
+	mapCalled     bool
+	getCalled     bool
+	removeCalled  bool
+	capturedGroup string
+}
+
+func (m *mockServiceNameMapping) Map(url *common.URL) error {
+	m.mapCalled = true
+	serviceInterface := url.GetParam(constant.InterfaceKey, "")
+	appName := url.GetParam(constant.ApplicationKey, "")
+	m.data[serviceInterface] = gxset.NewSet(appName)
+	return nil
+}
+func (m *mockServiceNameMapping) Get(url *common.URL, _ mapping.MappingListener) (*gxset.HashSet, error) {
+	m.getCalled = true
+	m.capturedGroup = url.GetParam(constant.GroupKey, "")
+	serviceInterface := url.GetParam(constant.InterfaceKey, "")
+	if s, ok := m.data[serviceInterface]; ok {
+		return s, nil
+	}
+	return gxset.NewSet(), nil
+}
+func (m *mockServiceNameMapping) Remove(url *common.URL) error { m.removeCalled = true; return nil }
+
+type mockNotifyListener struct{}
+
+func (m *mockNotifyListener) Notify(*registry.ServiceEvent) {
+	// for mocking
+}
+func (m *mockNotifyListener) NotifyAll([]*registry.ServiceEvent, func()) {
+	// for mocking
+}
+
+type mockProtocol struct{}
+
+func (m *mockProtocol) Export(invoker protocol.Invoker) protocol.Exporter { return &mockExporter{} }
+func (m *mockProtocol) Refer(url *common.URL) protocol.Invoker            { return &mockInvoker{} }
+func (m *mockProtocol) Destroy() {
+	// for mocking
+}
+
+type mockExporter struct{}
+
+func (m *mockExporter) UnExport() {
+	// for mocking
+}
+
+func (m *mockExporter) Unexport() {
+	// for mocking
+}
+func (m *mockExporter) GetInvoker() protocol.Invoker { return &mockInvoker{} }
+
+type mockProxyFactory struct{}
+
+func (m *mockProxyFactory) GetProxy(invoker base.Invoker, url *common.URL) *proxy.Proxy {
+	return &proxy.Proxy{}
+}
+
+func (m *mockProxyFactory) GetAsyncProxy(invoker base.Invoker, callBack any, url *common.URL) *proxy.Proxy {
+	return &proxy.Proxy{}
+}
+
+func (m *mockProxyFactory) GetInvoker(url *common.URL) protocol.Invoker { return &mockInvoker{} }
+
+type mockInvoker struct{}
+
+func (m *mockInvoker) GetURL() *common.URL { return nil }
+func (m *mockInvoker) IsAvailable() bool   { return true }
+func (m *mockInvoker) Destroy() {
+	// for mocking
+}
+func (m *mockInvoker) Invoke(context.Context, protocol.Invocation) protocol.Result { return nil }