cadmin/server/package/graveserver/leafstalk/conf/nacos.go

package conf

import (
	"encoding/json"
	"errors"
	"fmt"
	"leafstalk/log"
	"strings"

	"github.com/nacos-group/nacos-sdk-go/v2/clients"
	"github.com/nacos-group/nacos-sdk-go/v2/clients/config_client"
	"github.com/nacos-group/nacos-sdk-go/v2/common/constant"
	"github.com/nacos-group/nacos-sdk-go/v2/vo"
	"github.com/spf13/viper"
)

// LocalNacosConfig 本地nacos配置
type LocalNacosConfig struct {
	NamespaceId string          `json:"namespaceId"` // 空间命名
	LineGroup   string          `json:"lineGroup"`   // 线路分组
	PublicGroup string          `json:"publicGroup"` // 线路分组
	UserName    string          `json:"userName"`    //
	PassWord    string          `json:"passWord"`    //
	Servers     []*ConfigServer `json:"servers"`     // 集群服务器
}

type ConfigServer struct {
	IpAddr string `json:"ipAddr"` // ip地址
	Port   uint64 `json:"port"`   // 端口
}

// ConfigCenterAgent 配置中心代理
type ConfigCenterAgent struct {
	NacosClient config_client.IConfigClient // 客户端api
}

var (
	configCenterAgent *ConfigCenterAgent
	NacosConfig       *LocalNacosConfig
)

// 配置中心代理
func NewNacosConfigCenter(cfg *LocalNacosConfig, dataDirName string) (cli *ConfigCenterAgent, err error) {

	// 基本配置检查
	if cfg == nil || len(cfg.NamespaceId) == 0 { // len(cfg.DataID) == 0 ||
		err = errors.New("configuration check failed")
		return
	}

	// nacos客户端
	nacosClient, err := createNacosClient(cfg, dataDirName)
	if err != nil {
		return nil, err
	}

	cli = new(ConfigCenterAgent)
	cli.NacosClient = nacosClient
	return
}

// createNacosClient create config client.
func createNacosClient(cfg *LocalNacosConfig, dataDirName string) (iClient config_client.IConfigClient, err error) {
	clientConfig := *constant.NewClientConfig(
		constant.WithNamespaceId(cfg.NamespaceId), //When namespace is public, fill in the blank string here.
		constant.WithTimeoutMs(5000),
		constant.WithBeatInterval(1000),
		constant.WithNotLoadCacheAtStart(true),
		constant.WithLogDir(fmt.Sprintf("../nacos/%s/log", dataDirName)),
		constant.WithCacheDir(fmt.Sprintf("../nacos/%s/cache", dataDirName)),
		constant.WithLogLevel("info"),
		constant.WithUsername(cfg.UserName),
		constant.WithPassword(cfg.PassWord),
	)

	var serverConfigs []constant.ServerConfig
	for _, serv := range cfg.Servers {
		c := constant.NewServerConfig(
			serv.IpAddr,
			serv.Port,
			constant.WithScheme("http"),
			constant.WithContextPath("/nacos"),
		)
		serverConfigs = append(serverConfigs, *c)
	}

	// Another way of create config client for dynamic configuration (recommend)
	return clients.NewConfigClient(
		vo.NacosClientParam{
			ClientConfig:  &clientConfig,
			ServerConfigs: serverConfigs,
		},
	)
}

// GetIClient get client api
func (c *ConfigCenterAgent) GetIClient() config_client.IConfigClient {
	return c.NacosClient
}

// GetConfig get config info
func (c *ConfigCenterAgent) GetConfig(dataId string, group string) (string, error) {
	return c.NacosClient.GetConfig(vo.ConfigParam{
		DataId: dataId,
		Group:  group,
	})
}

// PublishConfig publish config
func (c *ConfigCenterAgent) PublishConfig(content string, dataId string, group string) (bool, error) {
	return c.NacosClient.PublishConfig(vo.ConfigParam{
		DataId:  dataId,
		Group:   group,
		Content: content,
	})
}

// DeleteConfig delete config
func (c *ConfigCenterAgent) DeleteConfig(dataId string, group string) (bool, error) {
	return c.NacosClient.DeleteConfig(vo.ConfigParam{
		DataId: dataId,
		Group:  group,
	})
}

// CancelListenConfig Cancel the listening of config change event
func (c *ConfigCenterAgent) CancelListenConfig(dataId string, group string) error {
	return c.NacosClient.CancelListenConfig(vo.ConfigParam{
		DataId: dataId,
		Group:  group,
	})
}

// ListenChanges Listen config change event
func (c *ConfigCenterAgent) ListenChanges(dataId string, group string, onChange func(namespace, group, dataId, data string)) error {
	return c.NacosClient.ListenConfig(vo.ConfigParam{
		DataId:   dataId,
		Group:    group,
		OnChange: onChange,
	})
}

func ListenConfigChanged(group string, dataId string, callback func(group, dataId string, newContent string)) error {
	content1, err := configCenterAgent.GetConfig(dataId, group)
	if err != nil {
		return err
	}

	callback(group, dataId, content1)

	// 开始监听动态配置变化
	err = configCenterAgent.ListenChanges(dataId, group, func(namespace, group, dataId, newContent string) {
		defer func() {
			if err := recover(); err != nil {
				log.Warnf("ListenConfigChanged ListenChanges error %v", err)
			}
		}()

		callback(group, dataId, newContent)
	})
	if err != nil {
		return err
	}
	return nil
}

// LoadSysConfig 加载系统配置
// 先从本地拿到nacos配置，然后到nacos中拿系统配置
func LoadSysConfig(cfgFile string, appName string) (sub1 *Config, err error) {
	conf1 := viper.New()
	conf1.SetConfigFile(cfgFile) //"../conf/grave.json"
	if err = conf1.ReadInConfig(); err != nil {
		log.Warnf("LoadSysConfig ReadInConfig error %v", err)
		return
	}

	// 取nacos相关配置
	var nacosCfg *LocalNacosConfig
	if err = conf1.UnmarshalKey("nacos", &nacosCfg); err != nil {
		log.Warnf("LoadSysConfig UnmarshalKey error %v", err)
		return
	}

	if nacosCfg == nil {
		return nil, errors.New("no config")
	}
	log.Infof("LoadSysConfig nacosCfg %v", *nacosCfg)

	// 构成配置中心
	cfgCenter, err := NewNacosConfigCenter(nacosCfg, appName)
	if err != nil {
		log.Warnf("LoadSysConfig NewNacosConfigCenter error %v", err)
		return
	}
	configCenterAgent = cfgCenter
	NacosConfig = nacosCfg

	// // 初始化动态配置事件
	// err = initDynamicEvent(cfgCenter, nacosCfg.PublicGroup)
	// if err != nil {
	// 	log.Warnf("LoadSysConfig initDynamicEvent error %v", err)
	// 	return
	// }

	// 从配置中心拿静态配置
	content, err := cfgCenter.GetConfig(appName, nacosCfg.LineGroup)
	if err != nil {
		log.Warnf("LoadSysConfig GetConfig %v error %v", nacosCfg.LineGroup, err)
		return
	}

	if err = conf1.MergeConfig(strings.NewReader(content)); err != nil {
		log.Warnf("LoadSysConfig MergeConfig %v error %v %v", nacosCfg.LineGroup, content, err)
		return
	}

	// // 共享组配置
	// content2, err2 := cfgCenter.GetConfig(appName, nacosCfg.PublicGroup)
	// if err2 != nil {
	// 	log.Warnf("LoadSysConfig GetConfig %v error %v", nacosCfg.PublicGroup, err2)
	// 	return
	// }
	// if err = conf1.MergeConfig(strings.NewReader(content2)); err != nil {
	// 	log.Warnf("LoadSysConfig MergeConfig %v error %v %v", nacosCfg.PublicGroup, content, err)
	// 	return
	// }

	// 拿各模块配置
	includeList := conf1.GetStringSlice("include")
	for _, v := range includeList {
		dataId := v
		lineGroup := nacosCfg.LineGroup
		ss := strings.Split(v, ".")
		if len(ss) == 2 {
			lineGroup = ss[0]
			dataId = ss[1]
		}

		// 先获取数据项
		var subContent string
		if subContent, err = cfgCenter.GetConfig(dataId, lineGroup); err != nil {
			log.Warnf("LoadSysConfig GetConfig %v error2 %v", dataId, err)
			return nil, err
		}

		var data map[string]interface{}
		if err = json.Unmarshal([]byte(subContent), &data); err != nil {
			log.Warnf("LoadSysConfig Unmarshal error2 %v", err)
			err = fmt.Errorf("load include sub:%v, err:%+v", dataId, err)
			return nil, err
		}

		conf1.Set(dataId, data)
	}

	// 展示
	for _, key := range conf1.AllKeys() {
		value := conf1.Get(key)
		fmt.Printf("%s: %v\n", key, value)
	}

	sub1 = NewConfig(conf1)
	return
}

// ["etcd", "nats", "sharedRedis", "lines.etcd", "lines.nats.[gmSbject.line1]", "lines.redis.[line1]"  ]
// 从.[分割，前面是数据项；后面是可选项
// 数据项单项：默认数据组做数据组 单项值做数据ID
// 数据项双项：第一项是数据组，第二项是数据ID
// 可选项：可以有多个可选项
func parseIncludeItem(itemVal string, defaultLineGroup string) (lineGroup string, dataId string, opts []string, err error) {

	splitDataId := func(val string) (group string, dataId string, err error) {
		vals := strings.Split(val, ".")
		if len(vals) == 1 {
			return defaultLineGroup, vals[0], nil
		} else if len(vals) == 2 {
			return vals[0], vals[1], nil
		}
		return "", "", errors.New("error dataId format")
	}

	// 用.[分割字符串
	vals := strings.Split(itemVal, ".[")
	num := len(vals)
	if num != 1 && num != 2 {
		return "", "", nil, errors.New("error dataId format1")
	}
	if num == 2 {
		vals[1] = strings.TrimRight(vals[1], "]")
	}

	// 先分割数据项
	lineGroup, dataId, err = splitDataId(vals[0])
	if err != nil {
		return "", "", nil, errors.New("error dataId format2")
	}
	if num == 1 {
		return lineGroup, dataId, nil, nil
	}

	// 分割选项
	opts = strings.Split(vals[1], ",")
	return lineGroup, dataId, opts, nil
}

// "lines.redis.[line1]"; "lines.redis.[db.line1]"; "lines.redis.[db.line1, addr.line1]"
// 所有非可选项 + 可选项 做数据项的值.可以有多个可选项. 针对每个子可选项：
// 分割后只有一项：单项值做数据项的值；只有一项时，不能有其他可选项，例如不能lines.redis.[line1， addr.line1]
// 分割后有2项：第二项的值提取出来，做第一项的值
func adjustOpts(dataId string, data map[string]interface{}, opts []string) (map[string]interface{}, error) {

	for _, opt := range opts {

		path := strings.Split(opt, ".")

		if len(path) == 1 {
			// 子项做主项，只能一个，并且没有兄弟子项
			if len(opts) != 1 {
				return nil, errors.New("errors opts format1")
			}
			data[dataId] = data[opt]
			return data, nil

		} else if len(path) == 2 {
			// 孙子项替换子项 lines.redis.[db.line1]
			childKey := path[0]
			grandsonKey := path[1]

			childVal, ok := data[childKey]
			if !ok {
				return nil, errors.New("errors opts format2")
			}
			grandsons, ok := childVal.(map[string]interface{})
			if !ok {
				return nil, errors.New("errors opts format3")
			}
			data[childKey], ok = grandsons[grandsonKey]
			if !ok {
				return nil, errors.New("errors opts format4")
			}
			continue
		}

		return nil, errors.New("errors opts formatn")
	}
	return data, nil
}