Watcher package as base for all state watcher.

state/watcher/watcher.go

+package watcher
+
+import (
+        "launchpad.net/gozk/zookeeper"
+        "launchpad.net/tomb"
+        "sort"
+)
+
+// updater defines the interface that has to be implemented
+// by the concrete watchers to handle and deliver node·
+// updates individually.

[niemeyer, 2012/03/28 00:28:59]

There's some mix up going on which would be good to get cleaned. The role of
updater is unclear, and that is shining through its documentation, that is
actually made in terms of watcher, ChildrenWatcher, and ContentWatcher, rather
than its own purpose. In fact, these types are all mixed up, without clear
boundaries, and if you dig through you'll note that all of that weight comes
from an attempt to share about 15 lines of code in the loop method.

Please follow on and break both types completely, getting rid of updater and
watcher. I'm betting the code will look cleaner, simpler, and thus easier to
understand.

[TheMue, 2012/03/28 11:12:57]

Done.

+type updater interface {
+        // update checks if the observed data has changed. Only in case·
+        // of a change the new data will be sent to a receiver. The methods
+        // returns the next watch and a continuation flag to the watcher.
+        // The latter will be set to false in case of an error or the
+        // receiving of a stop signal.
+        update(eventType zookeeper.Event) (nextWatch <-chan zookeeper.Event, cont bool)
+        // close is called if the watcher ends its work.
+        close()
+}
+
+// watcher provides the backend goroutine handling for the
+// observation of ZooKeeper node changes. The concrete handling
+// has to be done by the updater.
+type watcher struct {
+        tomb    tomb.Tomb
+        updater updater
+}
+
+// init assigns the updater and then starts the watching loop.
+func (w *watcher) init(updater updater, event zookeeper.Event) {
+        w.updater = updater
+        go w.loop(event)
+}
+
+// Stop ends the watching.
+func (w *watcher) Stop() error {
+        w.tomb.Kill(nil)
+        return w.tomb.Wait()
+}
+
+// loop is the backend for watching.
+func (w *watcher) loop(event zookeeper.Event) {
+        defer w.tomb.Done()
+        defer w.updater.close()
+
+        watch, cont := w.updater.update(event)
+        if !cont {
+                return
+        }
+        // Fire an initial event.
+        // watch := func() <-chan zookeeper.Event {

[niemeyer, 2012/03/28 00:28:59]

The dead code may be removed.

[TheMue, 2012/03/28 11:12:57]

Done.

+        //      eventChan := make(chan zookeeper.Event, 1)
+        //      eventChan <- zookeeper.Event{
+        //              State: zookeeper.STATE_CONNECTED,
+        //              Type:  zookeeper.EVENT_CHANGED,
+        //      }
+        //      return eventChan
+        // }()
+
+        for {
+                select {
+                case <-w.tomb.Dying():
+                        return
+                case evt := <-watch:
+                        if !evt.Ok() {
+                                w.tomb.Killf("watcher: critical session event: %v", evt)
+                                return
+                        }
+                        watch, cont = w.updater.update(event)
+                        if !cont {
+                                return
+                        }
+                }
+        }
+}
+
+// ContentWatcher observes a ZooKeeper node for changes of the content
+// and delivers those via the Change() method.

[niemeyer, 2012/03/28 00:28:59]

// ContentWatcher observes a ZooKeeper node and delivers a
// notification when a content change is detected.

[TheMue, 2012/03/28 11:12:57]

Done.

+type ContentWatcher struct {
+        zk         *zookeeper.Conn
+        path       string
+        changeChan chan string
+        content    string
+        watcher
+}
+
+// NewContentWatcher creates a new content watcher.

[niemeyer, 2012/03/28 00:28:59]

This comment may be dropped.

[TheMue, 2012/03/28 11:12:57]

Done.

+func NewContentWatcher(zk *zookeeper.Conn, path string) (*ContentWatcher, error) {
+        w := &ContentWatcher{
+                zk:         zk,
+                path:       path,
+                changeChan: make(chan string),
+        }
+        evt := zookeeper.Event{
+                State: zookeeper.STATE_CONNECTED,
+                Type:  zookeeper.EVENT_CHANGED,
+        }
+        w.watcher.init(w, evt)
+        return w, nil

[niemeyer, 2012/03/28 00:28:59]

If there are no possibilities of error, we don't need an error result.

[TheMue, 2012/03/28 11:12:57]

Done.

+}
+
+// Changes emits the content of the node on the
+// returned channel each time it changes.

[niemeyer, 2012/03/28 00:28:59]

// Changes returns a channel that will receive the new node
// content when a change is detected. Note that multiple
// changes may be observed as a single event in the channel.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ContentWatcher) Changes() <-chan string {
+        return w.changeChan
+}
+
+// update is documented in the updater interface above. For the
+// ContentWatcher it retrieves the nodes content as string and
+// emits changed content via the changeChan.

[niemeyer, 2012/03/28 00:28:59]

Needs a new comment, and the comment should not mention the code flow.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ContentWatcher) update(event zookeeper.Event) (nextWatch <-chan zookeeper.Event, cont bool) {

[niemeyer, 2012/03/28 00:28:59]

The use of "cont" in those methods feels a bit non-conventional, and seems to
encourage bad practices. For example, the logic below returns false if the node
is deleted. But.. and then what? The loop stops, returns, the watcher never
fires again.. no errors, just silent misbehavior.

This should really return (chan, error), and then return nil or the error as
appropriate. If a watched node goes away, that's an error that should be
returned. The call site, within loop, should be the one doing the "if err != nil
{ w.tomb.Kill(err); return }".

[TheMue, 2012/03/28 11:12:57]

Done.

+        if event.Type == zookeeper.EVENT_DELETED {
+                return nil, false
+        }
+        content, _, watch, err := w.zk.GetW(w.path)
+        if err != nil {
+                w.tomb.Kill(err)
+                return nil, false
+        }
+        if content == w.content {
+                return watch, true
+        }
+        w.content = content
+        select {
+        case <-w.tomb.Dying():
+                return nil, false
+        case w.changeChan <- w.content:
+        }
+        return watch, true
+}
+
+// close is documented in the updater interface above. For the
+// ContentWatcher it just closes the changeChan.

[niemeyer, 2012/03/28 00:28:59]

Observe:

  // For the ContentWatcher it just closes the changeChan.

vs.

  func (w *ContentWatcher) close() { close(w.changeChan) }

No need to code twice.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ContentWatcher) close() {
+        close(w.changeChan)
+}
+
+// ChildrenChange contains information about
+// children that have been created or deleted.
+type ChildrenChange struct {
+        // Del holds names of children that have been deleted.
+        Del []string
+        // New holds names of children that have been created.
+        New []string

[niemeyer, 2012/03/28 00:28:59]

s/Del/Removed/
s/New/Added/

And drop comments please. ChildrenChange.Added/Removed is clear.

[TheMue, 2012/03/28 11:12:57]

Done.

+}
+
+// ChildrenWatcher observes a ZooKeeper node for changes of the
+// children and delivers those via the Change() method.

[niemeyer, 2012/03/28 00:28:59]

// ChildrenWatcher observes a ZooKeeper node and delivers a
// notification when child nodes are added or removed.

[TheMue, 2012/03/28 11:12:57]

Done.

+type ChildrenWatcher struct {
+        zk         *zookeeper.Conn
+        path       string
+        changeChan chan ChildrenChange
+        children   map[string]bool
+        watcher
+}
+
+// NewWatcher creates a new watcher.

[niemeyer, 2012/03/28 00:28:59]

Comment is wrong, but may actually be dropped.

[TheMue, 2012/03/28 11:12:57]

Done.

+func NewChildrenWatcher(zk *zookeeper.Conn, path string) (*ChildrenWatcher, error) {
+        w := &ChildrenWatcher{
+                zk:         zk,
+                path:       path,
+                changeChan: make(chan ChildrenChange),
+                children:   make(map[string]bool),
+        }
+        evt := zookeeper.Event{
+                State: zookeeper.STATE_CONNECTED,
+                Type:  zookeeper.EVENT_CHILD,
+        }
+        w.watcher.init(w, evt)
+        return w, nil

[niemeyer, 2012/03/28 00:28:59]

If there are no errors, no need for an error result.

[TheMue, 2012/03/28 11:12:57]

Done.

+}
+
+// Changes emits the deleted and the new created children of·
+// the node on the returned channel each time they are changing.

[niemeyer, 2012/03/28 00:28:59]

// Changes returns a channel that will receive the changes
// performed to the set of children of the watched node.
// Note that multiple changes may be observed as a single
// event in the channel.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ChildrenWatcher) Changes() <-chan ChildrenChange {
+        return w.changeChan
+}
+
+// update is documented in the updater interface above. For the
+// ChildrenWatcher it retrieves the nodes children, checks which
+// are added or deleted and emits these changes via the changeChan.

[niemeyer, 2012/03/28 00:28:59]

Needs a new comment, and the comment should not mention the code flow.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ChildrenWatcher) update(event zookeeper.Event) (nextWatch <-chan zookeeper.Event, cont bool) {

[niemeyer, 2012/03/28 00:28:59]

Please return error rather than bool, and fix appropriately.

[TheMue, 2012/03/28 11:12:57]

Done.

+        if event.Type == zookeeper.EVENT_DELETED {
+                return nil, false
+        }
+        retrievedChildren, _, watch, err := w.zk.ChildrenW(w.path)
+        if err != nil {
+                w.tomb.Kill(err)
+                return nil, false
+        }
+        children := make(map[string]bool)
+        for _, child := range retrievedChildren {
+                children[child] = true
+        }
+        var change ChildrenChange
+        for child, _ := range w.children {
+                if !children[child] {
+                        change.Del = append(change.Del, child)
+                        delete(w.children, child)
+                }
+        }
+        for child, _ := range children {
+                if !w.children[child] {
+                        change.New = append(change.New, child)
+                        w.children[child] = true
+                }
+        }
+        if len(change.Del) == 0 && len(change.New) == 0 {
+                return watch, true
+        }
+        sort.Strings(change.Del)
+        sort.Strings(change.New)
+        select {
+        case <-w.tomb.Dying():
+                return nil, false
+        case w.changeChan <- change:
+        }
+        return watch, true
+}
+
+// close is documented in the updater interface above. For the
+// ChildrenWatcher it just closes the changeChan.

[niemeyer, 2012/03/28 00:28:59]

Comment needs love.

[TheMue, 2012/03/28 11:12:57]

Done.

+func (w *ChildrenWatcher) close() {
+        close(w.changeChan)
+}