(WIP) ServiceDiscoveryRegistry implementation

cluster/directory/base_directory.go

@@ -22,7 +22,6 @@ import (
 )
 
 import (
-	"github.com/dubbogo/gost/container/set"
 	"go.uber.org/atomic"
 )
 
@@ -35,8 +34,6 @@ import (
 	"github.com/apache/dubbo-go/common/logger"
 )
 
-var routerURLSet = gxset.NewSet()
-
 // BaseDirectory Abstract implementation of Directory: Invoker list returned from this Directory's list method have been filtered by Routers
 type BaseDirectory struct {
 	url       *common.URL
@@ -120,14 +117,3 @@ func (dir *BaseDirectory) Destroy(doDestroy func()) {
 func (dir *BaseDirectory) IsAvailable() bool {
 	return !dir.destroyed.Load()
 }
-
-// GetRouterURLSet Return router URL
-func GetRouterURLSet() *gxset.HashSet {
-	return routerURLSet
-}
-
-// AddRouterURLSet Add router URL
-// Router URL will init in config/config_loader.go
-func AddRouterURLSet(url *common.URL) {
-	routerURLSet.Add(url)
-}

common/constant/key.go

@@ -79,6 +79,7 @@ const (
 	EXECUTE_REJECTED_EXECUTION_HANDLER_KEY = "execute.limit.rejected.handler"
 	PROVIDER_SHUTDOWN_FILTER               = "pshutdown"
 	CONSUMER_SHUTDOWN_FILTER               = "cshutdown"
+	PID_KEY                                = "pid"
 	SYNC_REPORT_KEY                        = "sync.report"
 	RETRY_PERIOD_KEY                       = "retry.period"
 	RETRY_TIMES_KEY                        = "retry.times"
@@ -257,7 +258,21 @@ const (
 )
 
 // service discovery
-
 const (
-	NACOS_GROUP = "nacos.group"
+	SUBSCRIBED_SERVICE_NAMES_KEY              = "subscribed-services"
+	PROVIDER_BY                               = "provided-by"
+	EXPORTED_SERVICES_REVISION_PROPERTY_NAME  = "dubbo.exported-services.revision"
+	SERVICE_INSTANCE_SELECTOR                 = "service-instance-selector"
+	METADATA_STORAGE_TYPE_PROPERTY_NAME       = "dubbo.metadata.storage-type"
+	DEFAULT_METADATA_STORAGE_TYPE             = "local"
+	SERVICE_INSTANCE_ENDPOINTS                = "dubbo.endpoints"
+	METADATA_SERVICE_PREFIX                   = "dubbo.metadata-service."
+	METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME = METADATA_SERVICE_PREFIX + "url-params"
+	METADATA_SERVICE_URLS_PROPERTY_NAME       = METADATA_SERVICE_PREFIX + "urls"
+
+	// used by URL
+	// SERVICE_NAME_MAPPING_KEY indicate that which service name mapping instance will be used
+	SERVICE_NAME_MAPPING_KEY = "name_mapping"
+	// SERVICE_DISCOVERY_KEY indicate which service discovery instance will be used
+	SERVICE_DISCOVERY_KEY = "service_discovery"
 )

common/extension/service_discovery.go

@@ -21,25 +21,28 @@ import (
 	perrors "github.com/pkg/errors"
 )
 import (
-	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/registry"
 )
 
 var (
-	discoveryCreatorMap = make(map[string]func(url *common.URL) (registry.ServiceDiscovery, error), 4)
+	discoveryCreatorMap = make(map[string]func(name string) (registry.ServiceDiscovery, error), 4)
 )
 
 // SetServiceDiscovery will store the creator and name
-func SetServiceDiscovery(name string, creator func(url *common.URL) (registry.ServiceDiscovery, error)) {
-	discoveryCreatorMap[name] = creator
+// protocol indicate the implementation, like nacos
+// the name like nacos-1...
+func SetServiceDiscovery(protocol string, creator func(name string) (registry.ServiceDiscovery, error)) {
+	discoveryCreatorMap[protocol] = creator
 }
 
 // GetServiceDiscovery will return the registry.ServiceDiscovery
+// protocol indicate the implementation, like nacos
+// the name like nacos-1...
 // if not found, or initialize instance failed, it will return error.
-func GetServiceDiscovery(name string, url *common.URL) (registry.ServiceDiscovery, error) {
-	creator, ok := discoveryCreatorMap[name]
+func GetServiceDiscovery(protocol string, name string) (registry.ServiceDiscovery, error) {
+	creator, ok := discoveryCreatorMap[protocol]
 	if !ok {
 		return nil, perrors.New("Could not find the service discovery with name: " + name)
 	}
-	return creator(url)
+	return creator(name)
 }

common/extension/service_instance_selector_factory.go

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package extension
+
+import (
+	perrors "github.com/pkg/errors"
+)
+
+import (
+	"github.com/apache/dubbo-go/registry/servicediscovery/instance"
+)
+
+var (
+	serviceInstanceSelectorMappings = make(map[string]func() instance.ServiceInstanceSelector)
+)
+
+func SetServiceInstanceSelector(name string, f func() instance.ServiceInstanceSelector) {
+	serviceInstanceSelectorMappings[name] = f
+}
+
+func GetServiceInstanceSelector(name string) (instance.ServiceInstanceSelector, error) {
+	serviceInstanceSelector, ok := serviceInstanceSelectorMappings[name]
+	if !ok {
+		return nil, perrors.New("Could not find service instance selector with" +
+			"name:" + name)
+	}
+	return serviceInstanceSelector(), nil
+}

common/extension/service_name_mapping.go

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package extension
+
+import "github.com/apache/dubbo-go/metadata/mapping"
+
+var (
+	nameMappings = make(map[string]func() mapping.ServiceNameMapping)
+)
+
+func SetServiceNameMapping(name string, creator func() mapping.ServiceNameMapping) {
+	nameMappings[name] = creator
+}
+
+func GetServiceNameMapping(name string) mapping.ServiceNameMapping {
+	creator, ok := nameMappings[name]
+	if !ok {
+		panic("Can not find the target service name mapping: " + name)
+	}
+	return creator()
+}

common/observer/event_listener.go

@@ -42,7 +42,6 @@ type ConditionalEventListener interface {
 	Accept(e Event) bool
 }
 
-// TODO (implement ConditionalEventListener)
-type ServiceInstancesChangedListener struct {
-	ServiceName string
+type ChangedNotify interface {
+	Notify(e Event)
 }

config/base_config.go

@@ -42,7 +42,12 @@ type multiConfiger interface {
 
 // BaseConfig is the common configuration for provider and consumer
 type BaseConfig struct {
-	ConfigCenterConfig  *ConfigCenterConfig `yaml:"config_center" json:"config_center,omitempty"`
+	ConfigCenterConfig *ConfigCenterConfig                `yaml:"config_center" json:"config_center,omitempty"`
+	Remotes            map[string]*RemoteConfig           `yaml:"remote" json:"remote,omitempty"`
+	ServiceDiscoveries map[string]*ServiceDiscoveryConfig `yaml:"service_discovery" json:"service_discovery,omitempty"`
+	// application config
+	ApplicationConfig *ApplicationConfig `yaml:"application" json:"application,omitempty" property:"application"`
+
 	configCenterUrl     *common.URL
 	prefix              string
 	fatherConfig        interface{}
@@ -51,6 +56,17 @@ type BaseConfig struct {
 	fileStream          *bytes.Buffer
 }
 
+func (c *BaseConfig) GetServiceDiscoveries(name string) (config *ServiceDiscoveryConfig, ok bool) {
+	config, ok = c.ServiceDiscoveries[name]
+	return
+}
+
+// GetRemoteConfig will return the remote's config with the name if found
+func (c *BaseConfig) GetRemoteConfig(name string) (config *RemoteConfig, ok bool) {
+	config, ok = c.Remotes[name]
+	return
+}
+
 // startConfigCenter will start the config center.
 // it will prepare the environment
 func (c *BaseConfig) startConfigCenter() error {
@@ -63,7 +79,7 @@ func (c *BaseConfig) startConfigCenter() error {
 	if c.prepareEnvironment() != nil {
 		return perrors.WithMessagef(err, "start config center error!")
 	}
-	//c.fresh()
+	// c.fresh()
 	return err
 }
 
@@ -101,14 +117,14 @@ func (c *BaseConfig) prepareEnvironment() error {
 			return perrors.WithStack(err)
 		}
 	}
-	//global config file
+	// global config file
 	mapContent, err := dynamicConfig.Parser().Parse(content)
 	if err != nil {
 		return perrors.WithStack(err)
 	}
 	config.GetEnvInstance().UpdateExternalConfigMap(mapContent)
 
-	//appGroup config file
+	// appGroup config file
 	if len(appContent) != 0 {
 		appMapConent, err := dynamicConfig.Parser().Parse(appContent)
 		if err != nil {
@@ -264,7 +280,7 @@ func setFieldValue(val reflect.Value, id reflect.Value, config *config.InmemoryC
 				if f.Kind() == reflect.Map {
 
 					if f.Type().Elem().Kind() == reflect.Ptr {
-						//initiate config
+						// initiate config
 						s := reflect.New(f.Type().Elem().Elem())
 						prefix := s.MethodByName("Prefix").Call(nil)[0].String()
 						for _, pfx := range strings.Split(prefix, "|") {
@@ -279,7 +295,7 @@ func setFieldValue(val reflect.Value, id reflect.Value, config *config.InmemoryC
 
 					}
 
-					//iter := f.MapRange()
+					// iter := f.MapRange()
 
 					for _, k := range f.MapKeys() {
 						v := f.MapIndex(k)
@@ -314,7 +330,7 @@ func (c *BaseConfig) fresh() {
 }
 
 func (c *BaseConfig) freshInternalConfig(config *config.InmemoryConfiguration) {
-	//reflect to init struct
+	// reflect to init struct
 	tp := reflect.ValueOf(c.fatherConfig).Elem().Type()
 	initializeStruct(tp, reflect.ValueOf(c.fatherConfig).Elem())
 

config/base_config_test.go

@@ -45,13 +45,15 @@ func Test_refresh(t *testing.T) {
 
 	father := &ConsumerConfig{
 		Check: &[]bool{true}[0],
-		ApplicationConfig: &ApplicationConfig{
-			Organization: "dubbo_org",
-			Name:         "dubbo",
-			Module:       "module",
-			Version:      "2.6.0",
-			Owner:        "dubbo",
-			Environment:  "test"},
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Organization: "dubbo_org",
+				Name:         "dubbo",
+				Module:       "module",
+				Version:      "2.6.0",
+				Owner:        "dubbo",
+				Environment:  "test"},
+		},
 		Registries: map[string]*RegistryConfig{
 			"shanghai_reg2": {
 				Protocol:   "mock",
@@ -139,13 +141,15 @@ func Test_appExternal_refresh(t *testing.T) {
 	config.GetEnvInstance().UpdateExternalConfigMap(mockMap)
 	father := &ConsumerConfig{
 		Check: &[]bool{true}[0],
-		ApplicationConfig: &ApplicationConfig{
-			Organization: "dubbo_org",
-			Name:         "dubbo",
-			Module:       "module",
-			Version:      "2.6.0",
-			Owner:        "dubbo",
-			Environment:  "test"},
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Organization: "dubbo_org",
+				Name:         "dubbo",
+				Module:       "module",
+				Version:      "2.6.0",
+				Owner:        "dubbo",
+				Environment:  "test"},
+		},
 		Registries: map[string]*RegistryConfig{
 			"shanghai_reg2": {
 				Protocol:   "mock",
@@ -225,13 +229,15 @@ func Test_appExternalWithoutId_refresh(t *testing.T) {
 	config.GetEnvInstance().UpdateExternalConfigMap(mockMap)
 	father := &ConsumerConfig{
 		Check: &[]bool{true}[0],
-		ApplicationConfig: &ApplicationConfig{
-			Organization: "dubbo_org",
-			Name:         "dubbo",
-			Module:       "module",
-			Version:      "2.6.0",
-			Owner:        "dubbo",
-			Environment:  "test"},
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Organization: "dubbo_org",
+				Name:         "dubbo",
+				Module:       "module",
+				Version:      "2.6.0",
+				Owner:        "dubbo",
+				Environment:  "test"},
+		},
 		Registries: map[string]*RegistryConfig{
 			"shanghai_reg2": {
 				Protocol:   "mock",
@@ -310,13 +316,15 @@ func Test_refresh_singleRegistry(t *testing.T) {
 
 	father := &ConsumerConfig{
 		Check: &[]bool{true}[0],
-		ApplicationConfig: &ApplicationConfig{
-			Organization: "dubbo_org",
-			Name:         "dubbo",
-			Module:       "module",
-			Version:      "2.6.0",
-			Owner:        "dubbo",
-			Environment:  "test"},
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Organization: "dubbo_org",
+				Name:         "dubbo",
+				Module:       "module",
+				Version:      "2.6.0",
+				Owner:        "dubbo",
+				Environment:  "test"},
+		},
 		Registries: map[string]*RegistryConfig{},
 		Registry:   &RegistryConfig{},
 		References: map[string]*ReferenceConfig{
@@ -373,13 +381,15 @@ func Test_refreshProvider(t *testing.T) {
 	config.GetEnvInstance().UpdateExternalConfigMap(mockMap)
 
 	father := &ProviderConfig{
-		ApplicationConfig: &ApplicationConfig{
-			Organization: "dubbo_org",
-			Name:         "dubbo",
-			Module:       "module",
-			Version:      "2.6.0",
-			Owner:        "dubbo",
-			Environment:  "test"},
+		BaseConfig: BaseConfig{
+			ApplicationConfig: &ApplicationConfig{
+				Organization: "dubbo_org",
+				Name:         "dubbo",
+				Module:       "module",
+				Version:      "2.6.0",
+				Owner:        "dubbo",
+				Environment:  "test"},
+		},
 		Registries: map[string]*RegistryConfig{
 			"shanghai_reg2": {
 				Protocol:   "mock",
@@ -472,7 +482,7 @@ func Test_initializeStruct(t *testing.T) {
 	reflect.ValueOf(consumerConfig).Elem().Set(v.Elem())
 
 	assert.Condition(t, func() (success bool) {
-		return consumerConfig.ApplicationConfig != nil
+		return consumerConfig.Registry != nil
 	})
 	assert.Condition(t, func() (success bool) {
 		return consumerConfig.Registries != nil