/
watcher.go
89 lines (74 loc) · 1.71 KB
/
watcher.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
87
88
89
package etcdv3
import (
"context"
"time"
"github.com/pkg/errors"
"github.com/coreos/etcd/clientv3"
"github.com/liangjfblue/cheetah/cores/discovery"
)
type etcdWatcher struct {
w clientv3.WatchChan
client *clientv3.Client
timeout time.Duration
stop chan struct{}
}
func newEtcdWatcher(d *etcdDiscovery, timeout time.Duration, opts ...discovery.WatchOption) (discovery.Watcher, error) {
var wp discovery.WatchOptions
for _, o := range opts {
o(&wp)
}
ctx, cancel := context.WithCancel(context.TODO())
stop := make(chan bool, 1)
go func() {
<-stop
cancel()
}()
watchPath := discovery.ServicePrefixPath()
if len(wp.Service) > 0 {
watchPath = discovery.ServicePath(wp.Service) + "/"
}
return &etcdWatcher{
w: d.client.Watch(ctx, watchPath, clientv3.WithPrefix(), clientv3.WithPrevKV()),
client: d.client,
timeout: timeout,
stop: make(chan struct{}, 1),
}, nil
}
func (e *etcdWatcher) Next() (*discovery.Result, error) {
for v := range e.w {
if v.Err() != nil {
return nil, v.Err()
}
for _, event := range v.Events {
var action discovery.EventType
service := discovery.Decode(event.Kv.Value)
switch event.Type {
case clientv3.EventTypePut:
if event.IsCreate() {
action = discovery.Create
} else if event.IsModify() {
action = discovery.Update
}
case clientv3.EventTypeDelete:
action = discovery.Delete
service = discovery.Decode(event.PrevKv.Value)
}
if service == nil {
continue
}
return &discovery.Result{
Action: action,
Service: service,
}, nil
}
}
return nil, errors.New("could not get next")
}
func (e *etcdWatcher) Stop() {
select {
case <-e.stop:
return
default:
close(e.stop)
}
}