fetch apolloConfig

Author: cyruslo

apollo/client.go

@@ -1,20 +1,115 @@
 package apollo
 
 import (
-	"crypto/tls"
-
-	"github.com/libgox/addr"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
 )
 
 type Client interface {
+	GetStringValue(key string) string
 }
 
-type Config struct {
-	Address addr.Address
-	// TlsConfig configuration information for tls.
-	TlsConfig *tls.Config
+type innerClient struct {
+	ctx      context.Context
+	config   *Config
+	storage  *storage
+	poller   *longPoll
+	listener Listener
 }
 
 func NewClient(config *Config) (Client, error) {
-	return newClient(config)
+	if config.Logger != nil {
+		SetLogger(config.Logger)
+	}
+	ctx := context.Background()
+	c := &innerClient{
+		ctx:     ctx,
+		config:  config,
+		storage: newStorage(config.Namespace),
+	}
+	c.poller = newLongPoll(config, c.updateHandle)
+
+	go c.poller.start(c.ctx)
+
+	return c, nil
+}
+
+func (i *innerClient) updateHandle() error {
+	change, err := i.sync()
+	if err != nil {
+		return err
+	}
+	if change == nil || len(change.Changes) == 0 {
+		return fmt.Errorf("no changes to sync")
+	}
+	if i.listener != nil {
+		i.listener.OnChange(change)
+	}
+	return nil
+}
+
+func (i *innerClient) sync() (*ChangeEvent, error) {
+	log.Infof("sync namespace %s with remote config server", i.config.Namespace)
+	url := i.config.GetSyncURI()
+	r := &requester{
+		client: &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig: i.config.TLSConfig,
+			},
+		},
+		retries: 3,
+	}
+	result, err := r.do(i.ctx, url, r.retries)
+	if err != nil || len(result) == 0 {
+		return nil, err
+	}
+
+	ac := &apolloConfiguration{}
+	if err = json.Unmarshal(result, ac); err != nil {
+		return nil, err
+	}
+	return i.updateCache(ac)
+}
+
+func (i *innerClient) updateCache(ac *apolloConfiguration) (*ChangeEvent, error) {
+	var change = &ChangeEvent{
+		Namespace: ac.Namespace,
+		Changes:   make(map[string]*Change),
+	}
+	c := i.storage.loadCache(ac.Namespace)
+
+	c.data.Range(func(k, v interface{}) bool {
+		key := k.(string)
+		value := v.(string)
+		if _, ok := ac.Configurations[key]; !ok {
+			c.data.Delete(key)
+		}
+		change.Changes[key] = onDelete(key, value)
+		return true
+	})
+
+	for k, v := range ac.Configurations {
+		old, ok := c.data.Load(k)
+		if !ok {
+			change.Changes[k] = onAdd(k, v)
+			c.data.Store(k, v)
+			continue
+		}
+		if old.(string) != v {
+			change.Changes[k] = onModify(k, old.(string), v)
+		}
+		c.data.Store(k, v)
+	}
+	// todo: save releaseKey?
+	return change, nil
+}
+
+func (i *innerClient) GetStringValue(key string) string {
+	v, ok := i.storage.loadCache(i.config.Namespace).data.Load(key)
+	if !ok {
+		return ""
+	}
+	return v.(string)
 }

apollo/client_impl.go

@@ -0,0 +1,17 @@
+package apollo
+
+import "testing"
+
+func TestNewClient(t *testing.T) {
+	c, _ := NewClient(&Config{
+		AppID:     "appId",
+		Cluster:   "default",
+		Namespace: "apollo",
+		Address:   "localhost:8080",
+		Secret:    "",
+		TLSConfig: nil,
+		Logger:    nil,
+	})
+	value := c.GetStringValue("")
+	t.Log(value)
+}

apollo/client_test.go

@@ -0,0 +1,45 @@
+package apollo
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net/url"
+)
+
+type Config struct {
+	AppID     string
+	Cluster   string
+	Namespace string
+	Address   string
+	Secret    string
+	// TlsConfig configuration information for tls.
+	TLSConfig *tls.Config
+	Logger    Logger
+}
+
+func (c *Config) GetNotifyURLSuffix(notifications string) string {
+	return fmt.Sprintf("%s/notifications/v2?appId=%s&cluster=%s&notifications=%s",
+		c.GetUrlPrefix(),
+		url.QueryEscape(c.AppID),
+		url.QueryEscape(c.Cluster),
+		url.QueryEscape(notifications))
+}
+
+func (c *Config) GetSyncURI() string {
+	return fmt.Sprintf("%s/configs/%s/%s/%s?releaseKey=&ip=%s",
+		c.GetUrlPrefix(),
+		url.QueryEscape(c.AppID),
+		url.QueryEscape(c.Cluster),
+		url.QueryEscape(c.Namespace),
+		GetLocalIP())
+}
+
+func (c *Config) GetUrlPrefix() string {
+	var urlPrefix string
+	if c.TLSConfig != nil {
+		urlPrefix = "https://"
+	} else {
+		urlPrefix = "http://"
+	}
+	return urlPrefix
+}

apollo/config.go

@@ -0,0 +1,53 @@
+package apollo
+
+type ChangeType int
+
+const (
+	ADD ChangeType = iota
+
+	MODIFY
+
+	DELETE
+)
+
+type Listener interface {
+	OnChange(event *ChangeEvent)
+}
+
+type Change struct {
+	Key        string
+	OldValue   string
+	NewValue   string
+	ChangeType ChangeType
+}
+
+type ChangeEvent struct {
+	Namespace      string
+	NotificationID int
+	Changes        map[string]*Change
+}
+
+func onDelete(key, value string) *Change {
+	return &Change{
+		Key:        key,
+		ChangeType: DELETE,
+		OldValue:   value,
+	}
+}
+
+func onModify(key, oldValue, newValue string) *Change {
+	return &Change{
+		Key:        key,
+		ChangeType: MODIFY,
+		OldValue:   oldValue,
+		NewValue:   newValue,
+	}
+}
+
+func onAdd(key, value string) *Change {
+	return &Change{
+		Key:        key,
+		ChangeType: ADD,
+		NewValue:   value,
+	}
+}

apollo/event.go

@@ -0,0 +1,42 @@
+package apollo
+
+type Logger interface {
+	Info(format string, args ...interface{})
+
+	Error(format string, args ...interface{})
+
+	Warning(format string, args ...interface{})
+
+	Infof(format string, args ...interface{})
+
+	Errorf(format string, args ...interface{})
+
+	Warningf(format string, args ...interface{})
+}
+
+var log Logger = &defaultLogger{}
+
+type defaultLogger struct {
+}
+
+func SetLogger(logger Logger) {
+	log = logger
+}
+
+func (d defaultLogger) Info(format string, args ...interface{}) {
+}
+
+func (d defaultLogger) Error(format string, args ...interface{}) {
+}
+
+func (d defaultLogger) Warning(format string, args ...interface{}) {
+}
+
+func (d defaultLogger) Infof(format string, args ...interface{}) {
+}
+
+func (d defaultLogger) Errorf(format string, args ...interface{}) {
+}
+
+func (d defaultLogger) Warningf(format string, args ...interface{}) {
+}

apollo/logger.go

@@ -0,0 +1,18 @@
+package apollo
+
+import "encoding/json"
+
+const defaultNotificationID int = -1
+
+type notification struct {
+	Namespace      string `json:"namespace"`
+	NotificationID int    `json:"notificationId"`
+}
+
+func (n *notification) String() string {
+	res, err := json.Marshal([]*notification{n})
+	if err != nil {
+		return ""
+	}
+	return string(res)
+}