forked from go-kit/kit
-
Notifications
You must be signed in to change notification settings - Fork 0
/
subscriber.go
86 lines (73 loc) · 2.05 KB
/
subscriber.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
package zk
import (
"github.com/samuel/go-zookeeper/zk"
"github.com/go-kit/kit/endpoint"
"github.com/go-kit/kit/log"
"github.com/go-kit/kit/sd"
"github.com/go-kit/kit/sd/cache"
)
// Subscriber yield endpoints stored in a certain ZooKeeper path. Any kind of
// change in that path is watched and will update the Subscriber endpoints.
type Subscriber struct {
client Client
path string
cache *cache.Cache
logger log.Logger
quitc chan struct{}
}
var _ sd.Subscriber = &Subscriber{}
// NewSubscriber returns a ZooKeeper subscriber. ZooKeeper will start watching
// the given path for changes and update the Subscriber endpoints.
func NewSubscriber(c Client, path string, factory sd.Factory, logger log.Logger) (*Subscriber, error) {
s := &Subscriber{
client: c,
path: path,
cache: cache.New(factory, logger),
logger: logger,
quitc: make(chan struct{}),
}
err := s.client.CreateParentNodes(s.path)
if err != nil {
return nil, err
}
instances, eventc, err := s.client.GetEntries(s.path)
if err != nil {
logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
return nil, err
}
logger.Log("path", s.path, "instances", len(instances))
s.cache.Update(instances)
go s.loop(eventc)
return s, nil
}
func (s *Subscriber) loop(eventc <-chan zk.Event) {
var (
instances []string
err error
)
for {
select {
case <-eventc:
// We received a path update notification. Call GetEntries to
// retrieve child node data, and set a new watch, as ZK watches are
// one-time triggers.
instances, eventc, err = s.client.GetEntries(s.path)
if err != nil {
s.logger.Log("path", s.path, "msg", "failed to retrieve entries", "err", err)
continue
}
s.logger.Log("path", s.path, "instances", len(instances))
s.cache.Update(instances)
case <-s.quitc:
return
}
}
}
// Endpoints implements the Subscriber interface.
func (s *Subscriber) Endpoints() ([]endpoint.Endpoint, error) {
return s.cache.Endpoints(), nil
}
// Stop terminates the Subscriber.
func (s *Subscriber) Stop() {
close(s.quitc)
}