vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

init cancel watcher

release-0.4
Xiang Li 2013-12-26 22:06:15 +08:00
parent c36f306a1d
commit 5e499456f0
7 changed files with 66 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/v1/watch_key_handler.go
View File

@ -25,11 +25,11 @@ func WatchKeyHandler(w http.ResponseWriter, req *http.Request, s Server) error {
}
// Start the watcher on the store.
c, err := s.Store().Watch(key, false, sinceIndex)
watcher, err := s.Store().NewWatcher(key, false, sinceIndex)
if err != nil {
return etcdErr.NewError(500, key, s.Store().Index())
}
event := <-c
event := <-watcher.EventChan
// Convert event to a response and write to client.
b, _ := json.Marshal(event.Response(s.Store().Index()))

5
server/v2/get_handler.go
View File

@ -55,7 +55,7 @@ func GetHandler(w http.ResponseWriter, req *http.Request, s Server) error {
}
// Start the watcher on the store.
eventChan, err := s.Store().Watch(key, recursive, sinceIndex)
watcher, err := s.Store().NewWatcher(key, recursive, sinceIndex)
if err != nil {
return etcdErr.NewError(500, key, s.Store().Index())
}
@ -65,8 +65,9 @@ func GetHandler(w http.ResponseWriter, req *http.Request, s Server) error {
select {
case <-closeChan:
watcher.Remove()
return nil
case event = <-eventChan:
case event = <-watcher.EventChan:
}
} else { //get

13
store/store.go
View File

@ -52,7 +52,8 @@ type Store interface {
value string, expireTime time.Time) (*Event, error)
Delete(nodePath string, recursive, dir bool) (*Event, error)
CompareAndDelete(nodePath string, prevValue string, prevIndex uint64) (*Event, error)
Watch(prefix string, recursive bool, sinceIndex uint64) (<-chan *Event, error)
NewWatcher(prefix string, recursive bool, sinceIndex uint64) (*Watcher, error)
Save() ([]byte, error)
Recovery(state []byte) error
@ -340,21 +341,21 @@ func (s *store) CompareAndDelete(nodePath string, prevValue string, prevIndex ui
return e, nil
}
func (s *store) Watch(key string, recursive bool, sinceIndex uint64) (<-chan *Event, error) {
func (s *store) NewWatcher(key string, recursive bool, sinceIndex uint64) (*Watcher, error) {
key = path.Clean(path.Join("/", key))
nextIndex := s.CurrentIndex + 1
s.worldLock.RLock()
defer s.worldLock.RUnlock()
var c <-chan *Event
var w *Watcher
var err *etcdErr.Error
if sinceIndex == 0 {
c, err = s.WatcherHub.watch(key, recursive, nextIndex)
w, err = s.WatcherHub.watch(key, recursive, nextIndex)
} else {
c, err = s.WatcherHub.watch(key, recursive, sinceIndex)
w, err = s.WatcherHub.watch(key, recursive, sinceIndex)
}
if err != nil {
@ -364,7 +365,7 @@ func (s *store) Watch(key string, recursive bool, sinceIndex uint64) (<-chan *Ev
return nil, err
}
return c, nil
return w, nil
}
// walk function walks all the nodePath and apply the walkFunc on each directory

38
store/store_test.go
View File

@ -446,7 +446,8 @@ func TestStoreCompareAndSwapPrevIndexFailsIfNotMatch(t *testing.T) {
// Ensure that the store can watch for key creation.
func TestStoreWatchCreate(t *testing.T) {
s := newStore()
c, _ := s.Watch("/foo", false, 0)
w, _ := s.NewWatcher("/foo", false, 0)
c := w.EventChan
s.Create("/foo", false, "bar", false, Permanent)
e := nbselect(c)
assert.Equal(t, e.Action, "create", "")
@ -458,9 +459,9 @@ func TestStoreWatchCreate(t *testing.T) {
// Ensure that the store can watch for recursive key creation.
func TestStoreWatchRecursiveCreate(t *testing.T) {
s := newStore()
c, _ := s.Watch("/foo", true, 0)
w, _ := s.NewWatcher("/foo", true, 0)
s.Create("/foo/bar", false, "baz", false, Permanent)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "create", "")
assert.Equal(t, e.Node.Key, "/foo/bar", "")
}
@ -469,9 +470,9 @@ func TestStoreWatchRecursiveCreate(t *testing.T) {
func TestStoreWatchUpdate(t *testing.T) {
s := newStore()
s.Create("/foo", false, "bar", false, Permanent)
c, _ := s.Watch("/foo", false, 0)
w, _ := s.NewWatcher("/foo", false, 0)
s.Update("/foo", "baz", Permanent)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "update", "")
assert.Equal(t, e.Node.Key, "/foo", "")
}
@ -480,9 +481,9 @@ func TestStoreWatchUpdate(t *testing.T) {
func TestStoreWatchRecursiveUpdate(t *testing.T) {
s := newStore()
s.Create("/foo/bar", false, "baz", false, Permanent)
c, _ := s.Watch("/foo", true, 0)
w, _ := s.NewWatcher("/foo", true, 0)
s.Update("/foo/bar", "baz", Permanent)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "update", "")
assert.Equal(t, e.Node.Key, "/foo/bar", "")
}
@ -491,9 +492,9 @@ func TestStoreWatchRecursiveUpdate(t *testing.T) {
func TestStoreWatchDelete(t *testing.T) {
s := newStore()
s.Create("/foo", false, "bar", false, Permanent)
c, _ := s.Watch("/foo", false, 0)
w, _ := s.NewWatcher("/foo", false, 0)
s.Delete("/foo", false, false)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "delete", "")
assert.Equal(t, e.Node.Key, "/foo", "")
}
@ -502,9 +503,9 @@ func TestStoreWatchDelete(t *testing.T) {
func TestStoreWatchRecursiveDelete(t *testing.T) {
s := newStore()
s.Create("/foo/bar", false, "baz", false, Permanent)
c, _ := s.Watch("/foo", true, 0)
w, _ := s.NewWatcher("/foo", true, 0)
s.Delete("/foo/bar", false, false)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "delete", "")
assert.Equal(t, e.Node.Key, "/foo/bar", "")
}
@ -513,9 +514,9 @@ func TestStoreWatchRecursiveDelete(t *testing.T) {
func TestStoreWatchCompareAndSwap(t *testing.T) {
s := newStore()
s.Create("/foo", false, "bar", false, Permanent)
c, _ := s.Watch("/foo", false, 0)
w, _ := s.NewWatcher("/foo", false, 0)
s.CompareAndSwap("/foo", "bar", 0, "baz", Permanent)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "compareAndSwap", "")
assert.Equal(t, e.Node.Key, "/foo", "")
}
@ -524,9 +525,9 @@ func TestStoreWatchCompareAndSwap(t *testing.T) {
func TestStoreWatchRecursiveCompareAndSwap(t *testing.T) {
s := newStore()
s.Create("/foo/bar", false, "baz", false, Permanent)
c, _ := s.Watch("/foo", true, 0)
w, _ := s.NewWatcher("/foo", true, 0)
s.CompareAndSwap("/foo/bar", "baz", 0, "bat", Permanent)
e := nbselect(c)
e := nbselect(w.EventChan)
assert.Equal(t, e.Action, "compareAndSwap", "")
assert.Equal(t, e.Node.Key, "/foo/bar", "")
}
@ -544,15 +545,16 @@ func TestStoreWatchExpire(t *testing.T) {
s.Create("/foo", false, "bar", false, time.Now().Add(500*time.Millisecond))
s.Create("/foofoo", false, "barbarbar", false, time.Now().Add(500*time.Millisecond))
c, _ := s.Watch("/", true, 0)
w, _ := s.NewWatcher("/", true, 0)
c := w.EventChan
e := nbselect(c)
assert.Nil(t, e, "")
time.Sleep(600 * time.Millisecond)
e = nbselect(c)
assert.Equal(t, e.Action, "expire", "")
assert.Equal(t, e.Node.Key, "/foo", "")
c, _ = s.Watch("/", true, 4)
e = nbselect(c)
w, _ = s.NewWatcher("/", true, 4)
e = nbselect(w.EventChan)
assert.Equal(t, e.Action, "expire", "")
assert.Equal(t, e.Node.Key, "/foofoo", "")
}

9
store/watcher.go
View File

@ -16,15 +16,16 @@ limitations under the License.
package store
type watcher struct {
eventChan chan *Event
type Watcher struct {
EventChan chan *Event
recursive bool
sinceIndex uint64
Remove func()
}
// notify function notifies the watcher. If the watcher interests in the given path,
// the function will return true.
func (w *watcher) notify(e *Event, originalPath bool, deleted bool) bool {
func (w *Watcher) notify(e *Event, originalPath bool, deleted bool) bool {
// watcher is interested the path in three cases and under one condition
// the condition is that the event happens after the watcher's sinceIndex
@ -41,7 +42,7 @@ func (w *watcher) notify(e *Event, originalPath bool, deleted bool) bool {
// For example a watcher is watching at "/foo/bar". And we deletes "/foo". The watcher
// should get notified even if "/foo" is not the path it is watching.
if (w.recursive || originalPath || deleted) && e.Index() >= w.sinceIndex {
w.eventChan <- e
w.EventChan <- e
return true
}
return false

39
store/watcher_hub.go
View File

@ -36,41 +36,48 @@ func newWatchHub(capacity int) *watcherHub {
// If recursive is true, the first change after index under key will be sent to the event channel.
// If recursive is false, the first change after index at key will be sent to the event channel.
// If index is zero, watch will start from the current index + 1.
func (wh *watcherHub) watch(key string, recursive bool, index uint64) (<-chan *Event, *etcdErr.Error) {
func (wh *watcherHub) watch(key string, recursive bool, index uint64) (*Watcher, *etcdErr.Error) {
event, err := wh.EventHistory.scan(key, recursive, index)
if err != nil {
return nil, err
}
eventChan := make(chan *Event, 1) // use a buffered channel
if event != nil {
eventChan <- event
return eventChan, nil
}
w := &watcher{
eventChan: eventChan,
w := &Watcher{
EventChan: make(chan *Event, 1), // use a buffered channel
recursive: recursive,
sinceIndex: index,
}
if event != nil {
w.EventChan <- event
return w, nil
}
l, ok := wh.watchers[key]
var elem *list.Element
if ok { // add the new watcher to the back of the list
l.PushBack(w)
elem = l.PushBack(w)
} else { // create a new list and add the new watcher
l := list.New()
l.PushBack(w)
l = list.New()
elem = l.PushBack(w)
wh.watchers[key] = l
}
w.Remove = func() {
l.Remove(elem)
if l.Len() == 0 {
delete(wh.watchers, key)
}
}
atomic.AddInt64(&wh.count, 1)
return eventChan, nil
return w, nil
}
// notify function accepts an event and notify to the watchers.
@ -109,7 +116,7 @@ func (wh *watcherHub) notifyWatchers(e *Event, path string, deleted bool) {
next := curr.Next() // save reference to the next one in the list
w, _ := curr.Value.(*watcher)
w, _ := curr.Value.(*Watcher)
if w.notify(e, e.Node.Key == path, deleted) {

9
store/watcher_test.go
View File

@ -23,10 +23,11 @@ import (
func TestWatcher(t *testing.T) {
s := newStore()
wh := s.WatcherHub
c, err := wh.watch("/foo", true, 1)
w, err := wh.watch("/foo", true, 1)
if err != nil {
t.Fatalf("%v", err)
}
c := w.EventChan
select {
case <-c:
@ -45,7 +46,8 @@ func TestWatcher(t *testing.T) {
t.Fatal("recv != send")
}
c, _ = wh.watch("/foo", false, 2)
w, _ = wh.watch("/foo", false, 2)
c = w.EventChan
e = newEvent(Create, "/foo/bar", 2, 2)
@ -69,7 +71,8 @@ func TestWatcher(t *testing.T) {
}
// ensure we are doing exact matching rather than prefix matching
c, _ = wh.watch("/fo", true, 1)
w, _ = wh.watch("/fo", true, 1)
c = w.EventChan
select {
case re = <-c: