Ftr: config center apollo

common/constant/key.go

@@ -100,6 +100,9 @@ const (
 
 const (
 	CONFIG_NAMESPACE_KEY  = "config.namespace"
+	CONFIG_GROUP_KEY      = "config.group"
+	CONFIG_CLUSTER_KEY    = "config.cluster"
+	CONFIG_CHECK_KEY      = "config.check"
 	CONFIG_TIMEOUT_KET    = "config.timeout"
 	CONFIG_VERSION_KEY    = "configVersion"
 	COMPATIBLE_CONFIG_KEY = "compatible_config"

config/base_config.go

@@ -47,7 +47,7 @@ type BaseConfig struct {
 }
 
 func (c *BaseConfig) startConfigCenter(ctx context.Context) error {
-	url, err := common.NewURL(ctx, c.ConfigCenterConfig.Address, common.WithProtocol(c.ConfigCenterConfig.Protocol))
+	url, err := common.NewURL(ctx, c.ConfigCenterConfig.Address, common.WithProtocol(c.ConfigCenterConfig.Protocol), common.WithParams(c.ConfigCenterConfig.GetUrlMap()))
 	if err != nil {
 		return err
 	}

config/base_config_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/apache/dubbo-go/common/config"
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/config_center"
+	_ "github.com/apache/dubbo-go/config_center/apollo"
 )
 
 func Test_refresh(t *testing.T) {

config/config_center_config.go

@@ -19,13 +19,18 @@ package config
 
 import (
 	"context"
+	"net/url"
 	"time"
 )
 
 import (
 	"github.com/creasty/defaults"
 )
 
+import (
+	"github.com/apache/dubbo-go/common/constant"
+)
+
 type ConfigCenterConfig struct {
 	context       context.Context
 	Protocol      string `required:"true"  yaml:"protocol"  json:"protocol,omitempty"`
@@ -35,6 +40,7 @@ type ConfigCenterConfig struct {
 	Username      string `yaml:"username" json:"username,omitempty"`
 	Password      string `yaml:"password" json:"password,omitempty"`
 	ConfigFile    string `default:"dubbo.properties" yaml:"config_file"  json:"config_file,omitempty"`
+	Namespace     string `default:"dubbo.properties" yaml:"namespace"  json:"namespace,omitempty"`
 	AppConfigFile string `default:"dubbo.properties" yaml:"app_config_file"  json:"app_config_file,omitempty"`
 	TimeoutStr    string `yaml:"timeout"  json:"timeout,omitempty"`
 	timeout       time.Duration
@@ -50,3 +56,11 @@ func (c *ConfigCenterConfig) UnmarshalYAML(unmarshal func(interface{}) error) er
 	}
 	return nil
 }
+
+func (c *ConfigCenterConfig) GetUrlMap() url.Values {
+	urlMap := url.Values{}
+	urlMap.Set(constant.CONFIG_NAMESPACE_KEY, c.Namespace)
+	urlMap.Set(constant.CONFIG_GROUP_KEY, c.Group)
+	urlMap.Set(constant.CONFIG_CLUSTER_KEY, c.Cluster)
+	return urlMap
+}

config_center/apollo/factory.go

@@ -0,0 +1,45 @@
+/*
+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 apollo
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/extension"
+	. "github.com/apache/dubbo-go/config_center"
+	"github.com/apache/dubbo-go/config_center/parser"
+)
+
+func init() {
+	extension.SetConfigCenterFactory("apollo", createDynamicConfigurationFactory)
+}
+
+func createDynamicConfigurationFactory() DynamicConfigurationFactory {
+	return &apolloConfigurationFactory{}
+}
+
+type apolloConfigurationFactory struct{}
+
+func (f *apolloConfigurationFactory) GetDynamicConfiguration(url *common.URL) (DynamicConfiguration, error) {
+	dynamicConfiguration, err := newApolloConfiguration(url)
+	if err != nil {
+		return nil, err
+	}
+	dynamicConfiguration.SetParser(&parser.DefaultConfigurationParser{})
+	return dynamicConfiguration, err
+
+}

config_center/apollo/impl.go

@@ -0,0 +1,179 @@
+/*
+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 apollo
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+import (
+	"github.com/pkg/errors"
+	"github.com/zouyx/agollo"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	. "github.com/apache/dubbo-go/config_center"
+	"github.com/apache/dubbo-go/config_center/parser"
+	"github.com/apache/dubbo-go/remoting"
+)
+
+const (
+	apolloProtocolPrefix = "http://"
+	apolloConfigFormat   = "%s.%s"
+)
+
+type apolloConfiguration struct {
+	url *common.URL
+
+	listeners sync.Map
+	appConf   *agollo.AppConfig
+	parser    parser.ConfigurationParser
+}
+
+func newApolloConfiguration(url *common.URL) (*apolloConfiguration, error) {
+	c := &apolloConfiguration{
+		url: url,
+	}
+	configAddr := c.getAddressWithProtocolPrefix(url)
+	configCluster := url.GetParam(constant.CONFIG_CLUSTER_KEY, "")
+
+	appId := url.GetParam(constant.CONFIG_GROUP_KEY, DEFAULT_GROUP)
+	namespaces := url.GetParam(constant.CONFIG_NAMESPACE_KEY, getProperties(DEFAULT_GROUP))
+	c.appConf = &agollo.AppConfig{
+		AppId:         appId,
+		Cluster:       configCluster,
+		NamespaceName: namespaces,
+		Ip:            configAddr,
+	}
+
+	agollo.InitCustomConfig(func() (*agollo.AppConfig, error) {
+		return c.appConf, nil
+	})
+
+	return c, agollo.Start()
+}
+
+func getChangeType(change agollo.ConfigChangeType) remoting.EventType {
+	switch change {
+	case agollo.ADDED:
+		return remoting.EventTypeAdd
+	case agollo.DELETED:
+		return remoting.EventTypeDel
+	case agollo.MODIFIED:
+		return remoting.EventTypeUpdate
+	default:
+		panic("unknown type: " + strconv.Itoa(int(change)))
+	}
+}
+
+func (c *apolloConfiguration) AddListener(key string, listener ConfigurationListener, opts ...Option) {
+	k := &Options{}
+	for _, opt := range opts {
+		opt(k)
+	}
+
+	key = k.Group + key
+	l, _ := c.listeners.LoadOrStore(key, NewApolloListener())
+	l.(*apolloListener).AddListener(listener)
+}
+
+func (c *apolloConfiguration) RemoveListener(key string, listener ConfigurationListener, opts ...Option) {
+	k := &Options{}
+	for _, opt := range opts {
+		opt(k)
+	}
+
+	key = k.Group + key
+	l, ok := c.listeners.Load(key)
+	if ok {
+		l.(*apolloListener).RemoveListener(listener)
+	}
+}
+
+func getProperties(namespace string) string {
+	return getNamespaceName(namespace, agollo.Properties)
+}
+
+func getNamespaceName(namespace string, configFileFormat agollo.ConfigFileFormat) string {
+	return fmt.Sprintf(apolloConfigFormat, namespace, configFileFormat)
+}
+
+func (c *apolloConfiguration) GetConfig(key string, opts ...Option) (string, error) {
+	k := &Options{}
+	for _, opt := range opts {
+		opt(k)
+	}
+	/**
+	 * when group is not null, we are getting startup configs(config file) from Config Center, for example:
+	 * key=dubbo.propertie
+	 */
+	if len(k.Group) != 0 {
+		config := agollo.GetConfig(key)
+		if config == nil {
+			return "", errors.New(fmt.Sprintf("nothiing in namespace:%s ", key))
+		}
+		return config.GetContent(agollo.Properties), nil
+	}
+
+	/**
+	 * when group is null, we are fetching governance rules(config item) from Config Center, for example:
+	 * namespace=use default, key =application.organization
+	 */
+	config := agollo.GetConfig(c.appConf.NamespaceName)
+	if config == nil {
+		return "", errors.New(fmt.Sprintf("nothiing in namespace:%s ", key))
+	}
+	return config.GetStringValue(key, ""), nil
+}
+
+func (c *apolloConfiguration) getAddressWithProtocolPrefix(url *common.URL) string {
+	address := url.Location
+	converted := address
+	if len(address) != 0 {
+		reg := regexp.MustCompile("\\s+")
+		address = reg.ReplaceAllString(address, "")
+		parts := strings.Split(address, ",")
+		addrs := make([]string, 0)
+		for _, part := range parts {
+			addr := part
+			if !strings.HasPrefix(part, apolloProtocolPrefix) {
+				addr = apolloProtocolPrefix + part
+			}
+			addrs = append(addrs, addr)
+		}
+		converted = strings.Join(addrs, ",")
+	}
+	return converted
+}
+
+func (c *apolloConfiguration) Parser() parser.ConfigurationParser {
+	return c.parser
+}
+func (c *apolloConfiguration) SetParser(p parser.ConfigurationParser) {
+	c.parser = p
+}
+
+func (c *apolloConfiguration) GetConfigs(key string, opts ...Option) (string, error) {
+	return c.GetConfig(key, opts...)
+}