Ftr: Application-level Registry Model

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/default.go

@@ -74,3 +74,12 @@ const (
 const (
 	COMMA_SPLIT_PATTERN = "\\s*[,]+\\s*"
 )
+
+const (
+	SIMPLE_METADATA_SERVICE_NAME = "MetadataService"
+	DEFAULT_REVIESION            = "N/A"
+)
+
+const (
+	SERVICE_DISCOVERY_DEFAULT_GROUP = "DEFAULT_GROUP"
+)

common/constant/key.go

@@ -22,10 +22,12 @@ const (
 )
 
 const (
+	PORT_KEY               = "port"
 	GROUP_KEY              = "group"
 	VERSION_KEY            = "version"
 	INTERFACE_KEY          = "interface"
 	PATH_KEY               = "path"
+	PROTOCOL_KEY           = "protocol"
 	SERVICE_KEY            = "service"
 	METHODS_KEY            = "methods"
 	TIMEOUT_KEY            = "timeout"
@@ -40,6 +42,7 @@ const (
 	TOKEN_KEY              = "token"
 	LOCAL_ADDR             = "local-addr"
 	REMOTE_ADDR            = "remote-addr"
+	PATH_SEPARATOR         = "/"
 	DUBBO_KEY              = "dubbo"
 	RELEASE_KEY            = "release"
 	ANYHOST_KEY            = "anyhost"
@@ -77,6 +80,11 @@ 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"
+	CYCLE_REPORT_KEY                       = "cycle.report"
 )
 
 const (
@@ -226,7 +234,6 @@ const (
 	KEY_SEPARATOR      = ":"
 	DEFAULT_PATH_TAG   = "metadata"
 	KEY_REVISON_PREFIX = "revision"
-	PATH_SEPARATOR     = "/"
 
 	// metadata service
 	METADATA_SERVICE_NAME = "org.apache.dubbo.metadata.MetadataService"
@@ -255,7 +262,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"
+	SUBSCRIBED_SERVICES_REVISION_PROPERTY_NAME = "dubbo.subscribed-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"
+
+	// SERVICE_DISCOVERY_KEY indicate which service discovery instance will be used
+	SERVICE_DISCOVERY_KEY = "service_discovery"
+	LANGUAGE_KEY          = "language"
+	GO_LANG               = "golang"
 )

common/extension/event_dispatcher.go

@@ -0,0 +1,70 @@
+/*
+ * 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 (
+	"sync"
+
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/common/observer"
+)
+
+var (
+	globalEventDispatcher observer.EventDispatcher
+	initEventListeners    []func() observer.EventListener
+	initEventOnce         sync.Once
+)
+
+var (
+	dispatchers = make(map[string]func() observer.EventDispatcher, 8)
+)
+
+// SetEventDispatcher by name
+func SetEventDispatcher(name string, v func() observer.EventDispatcher) {
+	dispatchers[name] = v
+}
+
+// SetAndInitGlobalDispatcher
+func SetAndInitGlobalDispatcher(name string) {
+	if len(name) == 0 {
+		name = "direct"
+	}
+	if globalEventDispatcher != nil {
+		logger.Warnf("EventDispatcher already init. It will be replaced")
+	}
+	if dp, ok := dispatchers[name]; !ok || dp == nil {
+		panic("EventDispatcher for " + name + " is not existing, make sure you have import the package.")
+	}
+	globalEventDispatcher = dispatchers[name]()
+}
+
+// GetGlobalDispatcher will init all listener and then return dispatcher
+func GetGlobalDispatcher() observer.EventDispatcher {
+	initEventOnce.Do(func() {
+		// we should delay to add the listeners to avoid some listeners left
+		for _, l := range initEventListeners {
+			globalEventDispatcher.AddEventListener(l())
+		}
+	})
+	return globalEventDispatcher
+}
+
+// AddEventListener it will be added in global event dispatcher
+func AddEventListener(creator func() observer.EventListener) {
+	initEventListeners = append(initEventListeners, creator)
+}

common/extension/metadata_report_factory.go

@@ -18,20 +18,20 @@
 package extension
 
 import (
-	"github.com/apache/dubbo-go/metadata"
+	"github.com/apache/dubbo-go/metadata/report/factory"
 )
 
 var (
-	metaDataReportFactories = make(map[string]func() metadata.MetadataReportFactory, 8)
+	metaDataReportFactories = make(map[string]func() factory.MetadataReportFactory, 8)
 )
 
 // SetMetadataReportFactory sets the MetadataReportFactory with @name
-func SetMetadataReportFactory(name string, v func() metadata.MetadataReportFactory) {
+func SetMetadataReportFactory(name string, v func() factory.MetadataReportFactory) {
 	metaDataReportFactories[name] = v
 }
 
 // GetMetadataReportFactory finds the MetadataReportFactory with @name
-func GetMetadataReportFactory(name string) metadata.MetadataReportFactory {
+func GetMetadataReportFactory(name string) factory.MetadataReportFactory {
 	if metaDataReportFactories[name] == nil {
 		panic("metadata report for " + name + " is not existing, make sure you have import the package.")
 	}

common/extension/metadata_service.go

@@ -0,0 +1,42 @@
+/*
+ * 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 (
+	"fmt"
+
+	"github.com/apache/dubbo-go/metadata/service"
+)
+
+var (
+	// there will be two types: local or remote
+	metadataServiceInsMap = make(map[string]func() (service.MetadataService, error), 2)
+)
+
+func SetMetadataService(msType string, creator func() (service.MetadataService, error)) {
+	metadataServiceInsMap[msType] = creator
+}
+
+func GetMetadataService(msType string) (service.MetadataService, error) {
+	if creator, ok := metadataServiceInsMap[msType]; ok {
+		return creator()
+	}
+	panic(fmt.Sprintf("could not find the metadata service creator for metadataType: %s, please check whether you have imported relative packages, \n"+
+		"local - github.com/apache/dubbo-go/metadata/service/inmemory, \n"+
+		"remote - github.com/apache/dubbo-go/metadata/service/remote", msType))
+}

common/extension/metadata_service_proxy_factory.go

@@ -0,0 +1,41 @@
+/*
+ * 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 (
+	"fmt"
+
+	"github.com/apache/dubbo-go/metadata/service"
+)
+
+var (
+	metadataServiceProxyFactoryMap = make(map[string]func() service.MetadataServiceProxyFactory)
+)
+
+func SetMetadataServiceProxyFactory(name string, creator func() service.MetadataServiceProxyFactory) {
+	metadataServiceProxyFactoryMap[name] = creator
+}
+
+func GetMetadataServiceProxyFactory(name string) service.MetadataServiceProxyFactory {
+	if f, ok := metadataServiceProxyFactoryMap[name]; ok {
+		return f()
+	}
+	panic(fmt.Sprintf("could not find the metadata service factory creator for name: %s, please check whether you have imported relative packages, \n"+
+		"local - github.com/apache/dubbo-go/metadata/service/inmemory, \n"+
+		"remote - github.com/apache/dubbo-go/metadata/service/remote", name))
+}

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_customizer.go

@@ -0,0 +1,49 @@
+/*
+ * 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 (
+	"sort"
+
+	"github.com/apache/dubbo-go/registry"
+)
+
+var (
+	customizers = make([]registry.ServiceInstanceCustomizer, 0, 8)
+)
+
+// AddCustomizers will put the customizer into slices and then sort them;
+// this method will be invoked several time, so we sort them here.
+func AddCustomizers(cus registry.ServiceInstanceCustomizer) {
+	customizers = append(customizers, cus)
+	sort.Stable(customizerSlice(customizers))
+}
+
+// GetCustomizers will return the sorted customizer
+func GetCustomizers() []registry.ServiceInstanceCustomizer {
+	return customizers
+}
+
+type customizerSlice []registry.ServiceInstanceCustomizer
+
+func (c customizerSlice) Len() int {
+	return len(c)
+}
+
+func (c customizerSlice) Swap(i, j int)      { c[i], c[j] = c[j], c[i] }
+func (c customizerSlice) Less(i, j int) bool { return c[i].GetPriority() < c[j].GetPriority() }