2013-10-08 09:17:58 +04:00
|
|
|
package store
|
|
|
|
|
|
|
|
import (
|
|
|
|
"container/list"
|
|
|
|
"path"
|
|
|
|
"strings"
|
2013-12-28 10:55:50 +04:00
|
|
|
"sync"
|
2013-10-08 09:17:58 +04:00
|
|
|
"sync/atomic"
|
|
|
|
|
|
|
|
etcdErr "github.com/coreos/etcd/error"
|
|
|
|
)
|
|
|
|
|
|
|
|
// A watcherHub contains all subscribed watchers
|
|
|
|
// watchers is a map with watched path as key and watcher as value
|
|
|
|
// EventHistory keeps the old events for watcherHub. It is used to help
|
|
|
|
// watcher to get a continuous event history. Or a watcher might miss the
|
|
|
|
// event happens between the end of the first watch command and the start
|
|
|
|
// of the second command.
|
|
|
|
type watcherHub struct {
|
2013-12-28 10:55:50 +04:00
|
|
|
mutex sync.Mutex // protect the hash map
|
2013-11-10 05:55:54 +04:00
|
|
|
watchers map[string]*list.List
|
|
|
|
count int64 // current number of watchers.
|
|
|
|
EventHistory *EventHistory
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
|
|
|
|
// newWatchHub creates a watchHub. The capacity determines how many events we will
|
|
|
|
// keep in the eventHistory.
|
|
|
|
// Typically, we only need to keep a small size of history[smaller than 20K].
|
|
|
|
// Ideally, it should smaller than 20K/s[max throughput] * 2 * 50ms[RTT] = 2000
|
|
|
|
func newWatchHub(capacity int) *watcherHub {
|
|
|
|
return &watcherHub{
|
2013-11-10 05:55:54 +04:00
|
|
|
watchers: make(map[string]*list.List),
|
|
|
|
EventHistory: newEventHistory(capacity),
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-09-09 03:56:10 +04:00
|
|
|
// Watch function returns a Watcher.
|
2013-12-28 10:55:50 +04:00
|
|
|
// If recursive is true, the first change after index under key will be sent to the event channel of the watcher.
|
|
|
|
// If recursive is false, the first change after index at key will be sent to the event channel of the watcher.
|
2013-10-08 09:17:58 +04:00
|
|
|
// If index is zero, watch will start from the current index + 1.
|
2014-09-09 03:56:10 +04:00
|
|
|
func (wh *watcherHub) watch(key string, recursive, stream bool, index uint64) (Watcher, *etcdErr.Error) {
|
2013-12-01 09:47:23 +04:00
|
|
|
event, err := wh.EventHistory.scan(key, recursive, index)
|
2013-10-08 09:17:58 +04:00
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2014-09-09 03:56:10 +04:00
|
|
|
w := &watcher{
|
|
|
|
eventChan: make(chan *Event, 1), // use a buffered channel
|
2013-12-26 18:06:15 +04:00
|
|
|
recursive: recursive,
|
2014-01-09 11:28:33 +04:00
|
|
|
stream: stream,
|
2013-12-26 18:06:15 +04:00
|
|
|
sinceIndex: index,
|
2014-02-15 04:16:55 +04:00
|
|
|
hub: wh,
|
2013-12-26 18:06:15 +04:00
|
|
|
}
|
2013-11-07 09:19:37 +04:00
|
|
|
|
2013-11-10 05:55:54 +04:00
|
|
|
if event != nil {
|
2014-09-09 03:56:10 +04:00
|
|
|
w.eventChan <- event
|
2013-12-26 18:06:15 +04:00
|
|
|
return w, nil
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
|
2013-12-28 10:55:50 +04:00
|
|
|
wh.mutex.Lock()
|
|
|
|
defer wh.mutex.Unlock()
|
|
|
|
|
2013-12-01 09:47:23 +04:00
|
|
|
l, ok := wh.watchers[key]
|
2013-10-08 09:17:58 +04:00
|
|
|
|
2013-12-26 18:06:15 +04:00
|
|
|
var elem *list.Element
|
|
|
|
|
2013-10-08 09:17:58 +04:00
|
|
|
if ok { // add the new watcher to the back of the list
|
2013-12-26 18:06:15 +04:00
|
|
|
elem = l.PushBack(w)
|
2013-10-08 09:17:58 +04:00
|
|
|
|
|
|
|
} else { // create a new list and add the new watcher
|
2013-12-26 18:06:15 +04:00
|
|
|
l = list.New()
|
|
|
|
elem = l.PushBack(w)
|
2013-12-01 09:47:23 +04:00
|
|
|
wh.watchers[key] = l
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
|
2014-01-09 09:29:04 +04:00
|
|
|
w.remove = func() {
|
2014-09-09 03:56:10 +04:00
|
|
|
if w.removed { // avoid removing it twice
|
2014-01-09 11:28:33 +04:00
|
|
|
return
|
|
|
|
}
|
|
|
|
w.removed = true
|
2013-12-26 18:06:15 +04:00
|
|
|
l.Remove(elem)
|
2013-12-28 11:51:16 +04:00
|
|
|
atomic.AddInt64(&wh.count, -1)
|
2013-12-26 18:06:15 +04:00
|
|
|
if l.Len() == 0 {
|
|
|
|
delete(wh.watchers, key)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2013-10-08 09:17:58 +04:00
|
|
|
atomic.AddInt64(&wh.count, 1)
|
|
|
|
|
2013-12-26 18:06:15 +04:00
|
|
|
return w, nil
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
|
|
|
|
// notify function accepts an event and notify to the watchers.
|
|
|
|
func (wh *watcherHub) notify(e *Event) {
|
|
|
|
e = wh.EventHistory.addEvent(e) // add event into the eventHistory
|
|
|
|
|
2013-11-28 08:04:52 +04:00
|
|
|
segments := strings.Split(e.Node.Key, "/")
|
2013-10-08 09:17:58 +04:00
|
|
|
|
|
|
|
currPath := "/"
|
|
|
|
|
|
|
|
// walk through all the segments of the path and notify the watchers
|
|
|
|
// if the path is "/foo/bar", it will notify watchers with path "/",
|
|
|
|
// "/foo" and "/foo/bar"
|
|
|
|
|
|
|
|
for _, segment := range segments {
|
|
|
|
currPath = path.Join(currPath, segment)
|
|
|
|
// notify the watchers who interests in the changes of current path
|
|
|
|
wh.notifyWatchers(e, currPath, false)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-01-22 05:26:56 +04:00
|
|
|
func (wh *watcherHub) notifyWatchers(e *Event, nodePath string, deleted bool) {
|
2013-12-28 10:55:50 +04:00
|
|
|
wh.mutex.Lock()
|
|
|
|
defer wh.mutex.Unlock()
|
|
|
|
|
2014-01-22 05:26:56 +04:00
|
|
|
l, ok := wh.watchers[nodePath]
|
2013-10-08 09:17:58 +04:00
|
|
|
if ok {
|
|
|
|
curr := l.Front()
|
|
|
|
|
2013-12-28 11:49:05 +04:00
|
|
|
for curr != nil {
|
2013-10-08 09:17:58 +04:00
|
|
|
next := curr.Next() // save reference to the next one in the list
|
|
|
|
|
2014-09-09 03:56:10 +04:00
|
|
|
w, _ := curr.Value.(*watcher)
|
2013-10-08 09:17:58 +04:00
|
|
|
|
2014-01-22 18:02:42 +04:00
|
|
|
originalPath := (e.Node.Key == nodePath)
|
2014-02-06 07:34:41 +04:00
|
|
|
if (originalPath || !isHidden(nodePath, e.Node.Key)) && w.notify(e, originalPath, deleted) {
|
2014-01-09 11:28:33 +04:00
|
|
|
if !w.stream { // do not remove the stream watcher
|
|
|
|
// if we successfully notify a watcher
|
|
|
|
// we need to remove the watcher from the list
|
|
|
|
// and decrease the counter
|
|
|
|
l.Remove(curr)
|
|
|
|
atomic.AddInt64(&wh.count, -1)
|
|
|
|
}
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
|
2013-12-28 11:49:05 +04:00
|
|
|
curr = next // update current to the next element in the list
|
|
|
|
}
|
|
|
|
|
|
|
|
if l.Len() == 0 {
|
|
|
|
// if we have notified all watcher in the list
|
|
|
|
// we can delete the list
|
2014-01-22 05:26:56 +04:00
|
|
|
delete(wh.watchers, nodePath)
|
2013-10-08 09:17:58 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// clone function clones the watcherHub and return the cloned one.
|
|
|
|
// only clone the static content. do not clone the current watchers.
|
|
|
|
func (wh *watcherHub) clone() *watcherHub {
|
|
|
|
clonedHistory := wh.EventHistory.clone()
|
|
|
|
|
|
|
|
return &watcherHub{
|
|
|
|
EventHistory: clonedHistory,
|
|
|
|
}
|
|
|
|
}
|
2014-01-22 05:26:56 +04:00
|
|
|
|
2014-02-06 07:34:41 +04:00
|
|
|
// isHidden checks to see if key path is considered hidden to watch path i.e. the
|
2014-01-22 18:29:33 +04:00
|
|
|
// last element is hidden or it's within a hidden directory
|
2014-02-06 07:34:41 +04:00
|
|
|
func isHidden(watchPath, keyPath string) bool {
|
2014-02-06 20:09:47 +04:00
|
|
|
// When deleting a directory, watchPath might be deeper than the actual keyPath
|
|
|
|
// For example, when deleting /foo we also need to notify watchers on /foo/bar.
|
|
|
|
if len(watchPath) > len(keyPath) {
|
|
|
|
return false
|
|
|
|
}
|
2014-02-06 08:31:38 +04:00
|
|
|
// if watch path is just a "/", after path will start without "/"
|
|
|
|
// add a "/" to deal with the special case when watchPath is "/"
|
2014-02-06 07:34:41 +04:00
|
|
|
afterPath := path.Clean("/" + keyPath[len(watchPath):])
|
|
|
|
return strings.Contains(afterPath, "/_")
|
2014-01-22 05:26:56 +04:00
|
|
|
}
|