blob: e8a3557c1e914a7e0f3c91f1b8bd848fbbf81594 [file] [log] [blame]
khenaidooab1f7bd2019-11-14 14:00:27 -05001// Copyright 2017 The etcd Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package v2v3
16
17import (
18 "context"
19 "strings"
20
21 "go.etcd.io/etcd/clientv3"
22 "go.etcd.io/etcd/etcdserver/api/v2error"
23 "go.etcd.io/etcd/etcdserver/api/v2store"
24)
25
26func (s *v2v3Store) Watch(prefix string, recursive, stream bool, sinceIndex uint64) (v2store.Watcher, error) {
27 ctx, cancel := context.WithCancel(s.ctx)
28 wch := s.c.Watch(
29 ctx,
30 // TODO: very pricey; use a single store-wide watch in future
31 s.pfx,
32 clientv3.WithPrefix(),
33 clientv3.WithRev(int64(sinceIndex)),
34 clientv3.WithCreatedNotify(),
35 clientv3.WithPrevKV())
36 resp, ok := <-wch
37 if err := resp.Err(); err != nil || !ok {
38 cancel()
39 return nil, v2error.NewError(v2error.EcodeRaftInternal, prefix, 0)
40 }
41
42 evc, donec := make(chan *v2store.Event), make(chan struct{})
43 go func() {
44 defer func() {
45 close(evc)
46 close(donec)
47 }()
48 for resp := range wch {
49 for _, ev := range s.mkV2Events(resp) {
50 k := ev.Node.Key
51 if recursive {
52 if !strings.HasPrefix(k, prefix) {
53 continue
54 }
55 // accept events on hidden keys given in prefix
56 k = strings.Replace(k, prefix, "/", 1)
57 // ignore hidden keys deeper than prefix
58 if strings.Contains(k, "/_") {
59 continue
60 }
61 }
62 if !recursive && k != prefix {
63 continue
64 }
65 select {
66 case evc <- ev:
67 case <-ctx.Done():
68 return
69 }
70 if !stream {
71 return
72 }
73 }
74 }
75 }()
76
77 return &v2v3Watcher{
78 startRev: resp.Header.Revision,
79 evc: evc,
80 donec: donec,
81 cancel: cancel,
82 }, nil
83}
84
85func (s *v2v3Store) mkV2Events(wr clientv3.WatchResponse) (evs []*v2store.Event) {
86 ak := s.mkActionKey()
87 for _, rev := range mkRevs(wr) {
88 var act, key *clientv3.Event
89 for _, ev := range rev {
90 if string(ev.Kv.Key) == ak {
91 act = ev
92 } else if key != nil && len(key.Kv.Key) < len(ev.Kv.Key) {
93 // use longest key to ignore intermediate new
94 // directories from Create.
95 key = ev
96 } else if key == nil {
97 key = ev
98 }
99 }
100 v2ev := &v2store.Event{
101 Action: string(act.Kv.Value),
102 Node: s.mkV2Node(key.Kv),
103 PrevNode: s.mkV2Node(key.PrevKv),
104 EtcdIndex: mkV2Rev(wr.Header.Revision),
105 }
106 evs = append(evs, v2ev)
107 }
108 return evs
109}
110
111func mkRevs(wr clientv3.WatchResponse) (revs [][]*clientv3.Event) {
112 var curRev []*clientv3.Event
113 for _, ev := range wr.Events {
114 if curRev != nil && ev.Kv.ModRevision != curRev[0].Kv.ModRevision {
115 revs = append(revs, curRev)
116 curRev = nil
117 }
118 curRev = append(curRev, ev)
119 }
120 if curRev != nil {
121 revs = append(revs, curRev)
122 }
123 return revs
124}
125
126type v2v3Watcher struct {
127 startRev int64
128 evc chan *v2store.Event
129 donec chan struct{}
130 cancel context.CancelFunc
131}
132
133func (w *v2v3Watcher) StartIndex() uint64 { return mkV2Rev(w.startRev) }
134
135func (w *v2v3Watcher) Remove() {
136 w.cancel()
137 <-w.donec
138}
139
140func (w *v2v3Watcher) EventChan() chan *v2store.Event { return w.evc }