76 lines
1.8 KiB
Go
76 lines
1.8 KiB
Go
package etcd
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"github.com/fox/fox/log"
|
|
"sync"
|
|
)
|
|
|
|
type Registry[T INode] struct {
|
|
*etcdRegistryImpl
|
|
nodes *sync.Map
|
|
me T
|
|
}
|
|
|
|
func NewRegistry[T INode](endpoints []string, username, password string, me T) (*Registry[T], error) {
|
|
bs, err := json.Marshal(me)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
e := &Registry[T]{}
|
|
e.nodes = &sync.Map{}
|
|
e.etcdRegistryImpl, err = newServiceRegistryImpl(endpoints, me.EtcdRootKey(), username, password, e, me.EtcdKey(), string(bs))
|
|
|
|
return e, err
|
|
}
|
|
|
|
// 保存当前服务
|
|
func (r *Registry[T]) saveNode(newNodes *sync.Map, jsonBytes []byte) {
|
|
var tmp = new(T)
|
|
if err := json.Unmarshal(jsonBytes, tmp); err != nil {
|
|
log.ErrorF(err.Error())
|
|
}
|
|
newNodes.Store((*tmp).MapKey(), *tmp)
|
|
}
|
|
|
|
// 保存当前服务
|
|
func (r *Registry[T]) replace(newNodes *sync.Map) {
|
|
//log.DebugF("更新前.nodes ptr:%d", &r.nodes)
|
|
r.nodes = newNodes
|
|
//r.nodes.Range(func(key, value interface{}) bool {
|
|
// bV, _ := json.Marshal(value)
|
|
// log.DebugF("has node key:%s value:%s", key, string(bV))
|
|
// return true
|
|
//})
|
|
//log.DebugF("更新所有节点信息.nodes ptr:%d", &r.nodes)
|
|
}
|
|
|
|
// 获取当前根节点下所有节点信息
|
|
func (r *Registry[T]) RangeNode(cb func(key string, value *T) bool) {
|
|
//log.DebugF("循环处理.nodes ptr:%d", &r.nodes)
|
|
r.nodes.Range(func(key, value interface{}) bool {
|
|
k, _ := key.(string)
|
|
if node, ok := value.(T); ok {
|
|
return cb(k, &node)
|
|
} else {
|
|
//log.DebugF("转换失败.key:%v", key)
|
|
}
|
|
return true
|
|
})
|
|
}
|
|
|
|
// 根据inode的mapKey()查找对应的节点
|
|
func (r *Registry[T]) FindNode(key string) (T, error) {
|
|
var tmp = new(T)
|
|
v, ok := r.nodes.Load(key)
|
|
if !ok {
|
|
return *tmp, fmt.Errorf("%v not exist", key)
|
|
}
|
|
if node, ok := v.(T); ok {
|
|
return node, nil
|
|
}
|
|
return *tmp, fmt.Errorf("%v 类型转换失败", key)
|
|
}
|