mirror of
https://github.com/zeromicro/go-zero.git
synced 2026-05-09 08:00:00 +08:00
116 lines
2.5 KiB
Go
116 lines
2.5 KiB
Go
package subscriber
|
|
|
|
import (
|
|
"sync"
|
|
"sync/atomic"
|
|
|
|
"github.com/zeromicro/go-zero/core/discov"
|
|
"github.com/zeromicro/go-zero/core/logx"
|
|
)
|
|
|
|
type (
|
|
// etcdSubscriber is a subscriber that subscribes to etcd.
|
|
etcdSubscriber struct {
|
|
*discov.Subscriber
|
|
}
|
|
|
|
// EtcdConf is the configuration for etcd.
|
|
EtcdConf = discov.EtcdConf
|
|
)
|
|
|
|
// MustNewEtcdSubscriber returns an etcd Subscriber, exits on errors.
|
|
func MustNewEtcdSubscriber(conf EtcdConf) Subscriber {
|
|
s, err := NewEtcdSubscriber(conf)
|
|
logx.Must(err)
|
|
return s
|
|
}
|
|
|
|
// NewEtcdSubscriber returns an etcd Subscriber.
|
|
func NewEtcdSubscriber(conf EtcdConf) (Subscriber, error) {
|
|
opts := buildSubOptions(conf)
|
|
s, err := discov.NewSubscriber(conf.Hosts, conf.Key, opts...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &etcdSubscriber{Subscriber: s}, nil
|
|
}
|
|
|
|
// buildSubOptions constructs the options for creating a new etcd subscriber.
|
|
func buildSubOptions(conf EtcdConf) []discov.SubOption {
|
|
opts := []discov.SubOption{
|
|
discov.WithExactMatch(),
|
|
discov.WithContainer(newContainer()),
|
|
}
|
|
|
|
if len(conf.User) > 0 {
|
|
opts = append(opts, discov.WithSubEtcdAccount(conf.User, conf.Pass))
|
|
}
|
|
if len(conf.CertFile) > 0 || len(conf.CertKeyFile) > 0 || len(conf.CACertFile) > 0 {
|
|
opts = append(opts, discov.WithSubEtcdTLS(conf.CertFile, conf.CertKeyFile,
|
|
conf.CACertFile, conf.InsecureSkipVerify))
|
|
}
|
|
|
|
return opts
|
|
}
|
|
|
|
// AddListener adds a listener to the subscriber.
|
|
func (s *etcdSubscriber) AddListener(listener func()) error {
|
|
s.Subscriber.AddListener(listener)
|
|
return nil
|
|
}
|
|
|
|
// Value returns the value of the subscriber.
|
|
func (s *etcdSubscriber) Value() (string, error) {
|
|
vs := s.Subscriber.Values()
|
|
if len(vs) > 0 {
|
|
return vs[len(vs)-1], nil
|
|
}
|
|
|
|
return "", nil
|
|
}
|
|
|
|
type container struct {
|
|
value atomic.Value
|
|
listeners []func()
|
|
lock sync.Mutex
|
|
}
|
|
|
|
func newContainer() *container {
|
|
return &container{}
|
|
}
|
|
|
|
func (c *container) OnAdd(kv discov.KV) {
|
|
c.value.Store([]string{kv.Val})
|
|
c.notifyChange()
|
|
}
|
|
|
|
func (c *container) OnDelete(_ discov.KV) {
|
|
c.value.Store([]string(nil))
|
|
c.notifyChange()
|
|
}
|
|
|
|
func (c *container) AddListener(listener func()) {
|
|
c.lock.Lock()
|
|
c.listeners = append(c.listeners, listener)
|
|
c.lock.Unlock()
|
|
}
|
|
|
|
func (c *container) GetValues() []string {
|
|
if vals, ok := c.value.Load().([]string); ok {
|
|
return vals
|
|
}
|
|
|
|
return []string(nil)
|
|
}
|
|
|
|
func (c *container) notifyChange() {
|
|
c.lock.Lock()
|
|
listeners := append(([]func())(nil), c.listeners...)
|
|
c.lock.Unlock()
|
|
|
|
for _, listener := range listeners {
|
|
listener()
|
|
}
|
|
}
|