feat: read config from nacos

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/go-resty/resty/v2 v2.7.0
 	github.com/gogf/gf v1.16.9
+	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.3
 	github.com/gorilla/websocket v1.5.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
@@ -32,6 +33,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/modern-go/reflect2 v1.0.2
+	github.com/nacos-group/nacos-sdk-go/v2 v2.2.1
 	github.com/onsi/ginkgo/v2 v2.9.2
 	github.com/onsi/gomega v1.27.6
 	github.com/philchia/agollo/v4 v4.1.5
@@ -72,10 +74,12 @@ require (
 
 require (
 	github.com/StackExchange/wmi v1.2.1 // indirect
+	github.com/aliyun/alibaba-cloud-sdk-go v1.61.1704 // indirect
 	github.com/andybalholm/brotli v1.0.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bep/godartsass v0.16.0 // indirect
 	github.com/bep/golibsass v1.1.0 // indirect
+	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/bytedance/sonic v1.8.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -101,7 +105,6 @@ require (
 	github.com/goccy/go-json v0.10.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/gohugoio/hugo v0.111.3 // indirect
-	github.com/golang/mock v1.6.0 // indirect
 	github.com/golang/snappy v0.0.3 // indirect
 	github.com/gomodule/redigo v2.0.0+incompatible // indirect
 	github.com/google/flatbuffers v1.11.0 // indirect
@@ -114,6 +117,7 @@ require (
 	github.com/imdario/mergo v0.3.13 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jtolds/gls v4.20.0+incompatible // indirect
 	github.com/klauspost/compress v1.16.3 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
@@ -163,7 +167,9 @@ require (
 	golang.org/x/exp v0.0.0-20221031165847-c99f073a8326 // indirect
 	golang.org/x/net v0.8.0 // indirect
 	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.7.0 // indirect
+	gopkg.in/ini.v1 v1.66.2 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	stathat.com/c/consistent v1.0.0 // indirect

pkg/conf/datasource/nacos/nacos.go

@@ -0,0 +1,78 @@
+// Copyright 2022 Douyu
+//
+// Licensed 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 nacos
+
+import (
+	"github.com/douyu/jupiter/pkg/conf"
+	"github.com/nacos-group/nacos-sdk-go/v2/clients/config_client"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+)
+
+type nacosDataSource struct {
+	client config_client.IConfigClient
+	group  string
+	dataID string
+
+	changed chan struct{}
+}
+
+// NewDataSource creates an nacos DataSource
+func NewDataSource(client config_client.IConfigClient, group, dataID string, watch bool) conf.DataSource {
+	ds := &nacosDataSource{
+		client:  client,
+		group:   group,
+		dataID:  dataID,
+		changed: make(chan struct{}, 1),
+	}
+	if watch {
+		_ = ds.client.ListenConfig(vo.ConfigParam{
+			Group:  ds.group,
+			DataId: ds.dataID,
+			OnChange: func(namespace, group, dataId, data string) {
+				ds.changed <- struct{}{}
+			},
+		})
+	}
+	return ds
+}
+
+// ReadConfig reads config content from nacos
+func (ds *nacosDataSource) ReadConfig() ([]byte, error) {
+	configData, err := ds.client.GetConfig(vo.ConfigParam{
+		Group:  ds.group,
+		DataId: ds.dataID,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return []byte(configData), nil
+}
+
+// IsConfigChanged returns a chanel for notification when the config changed
+func (ds *nacosDataSource) IsConfigChanged() <-chan struct{} {
+	return ds.changed
+}
+
+// Close stops watching the config changed
+func (ds *nacosDataSource) Close() error {
+	_ = ds.client.CancelListenConfig(vo.ConfigParam{
+		Group:  ds.group,
+		DataId: ds.dataID,
+	})
+	ds.client.CloseClient()
+	close(ds.changed)
+	return nil
+}

pkg/conf/datasource/nacos/nacos_test.go

@@ -0,0 +1,98 @@
+// Copyright 2022 Douyu
+//
+// Licensed 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 nacos
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/douyu/jupiter/pkg/conf"
+	"github.com/douyu/jupiter/pkg/conf/datasource/nacos/mock"
+	"github.com/golang/mock/gomock"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	ds         conf.DataSource
+	wg         sync.WaitGroup
+	localParam = vo.ConfigParam{
+		DataId:  "data-id",
+		Group:   "group",
+		Content: "hello world",
+	}
+	newContent = "hello-world-changed"
+)
+
+func TestReadConfig(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	client := mock.NewMockIConfigClient(ctrl)
+	t.Run("with watch", func(t *testing.T) {
+		client.EXPECT().CancelListenConfig(gomock.Any()).Return(nil)
+		client.EXPECT().CloseClient().Return()
+		client.EXPECT().GetConfig(gomock.Any()).Return(localParam.Content, nil)
+		client.EXPECT().ListenConfig(gomock.Any()).DoAndReturn(func(param vo.ConfigParam) error {
+			go func() {
+				defer wg.Done()
+				time.Sleep(4 * time.Second)
+				param.OnChange("namespace", localParam.Group, localParam.DataId, newContent)
+				client.EXPECT().GetConfig(gomock.Any()).Return(newContent, nil)
+				time.Sleep(2 * time.Second)
+				teardown(t)
+			}()
+			wg.Add(1)
+			return nil
+		})
+
+		ds = NewDataSource(client, localParam.Group, localParam.DataId, true)
+		content, err := ds.ReadConfig()
+		assert.Nil(t, err)
+		assert.Equal(t, localParam.Content, string(content))
+		t.Logf("read config: %s", content)
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range ds.IsConfigChanged() {
+				content, err := ds.ReadConfig()
+				assert.Nil(t, err)
+				assert.Equal(t, newContent, string(content))
+				t.Logf("read new config: %s", content)
+			}
+		}()
+
+		wg.Wait()
+	})
+
+	t.Run("without with", func(t *testing.T) {
+		client.EXPECT().CancelListenConfig(gomock.Any()).Return(nil)
+		client.EXPECT().CloseClient().Return()
+		client.EXPECT().GetConfig(gomock.Any()).Return(localParam.Content, nil)
+		ds = NewDataSource(client, localParam.Group, localParam.DataId, false)
+		defer teardown(t)
+		content, err := ds.ReadConfig()
+		assert.Nil(t, err)
+		assert.Equal(t, localParam.Content, string(content))
+		t.Logf("read config: %s", content)
+	})
+}
+
+func teardown(t *testing.T) {
+	t.Helper()
+	ds.Close()
+}