bogus CL just to demonstrate some changes

state/watcher.go

 package state
 
 import (
         "launchpad.net/goyaml"
         "launchpad.net/juju-core/log"
         "launchpad.net/juju-core/state/presence"
         "launchpad.net/juju-core/state/watcher"
         "launchpad.net/tomb"
 )
 
@@ skipping 644 matching lines @@
         case w.changeChan <- ch:
                 w.current = latest
         }
         return nil
 }
 
 func (w *ServiceRelationsWatcher) done() {
         close(w.changeChan)
 }
 
+type relationUnit struct {
+        rw *relationUnitsWatcher
+        key string
+        name string
+        stop chan struct{}
+        stopped bool
+        w *watcher.ContentWatcher
+}
+
 // relationUnitsWatcher watches the presence and settings of units
 // playing a particular role in a particular scope of a relation,
 // on behalf of another relation unit (which can potentially be in
 // that scope/role, and will if so be exluded from reported events).
 type relationUnitsWatcher struct {
         st        *State
-        tomb      tomb.Tomb
+        stop <-chan struct{}
+        err error
         role      RelationRole
         scope     unitScopePath
         ignore    string
         updates   chan unitSettingsChange
-        unitTombs map[string]*tomb.Tomb
-        names     map[string]string
+        units map[string]*relationUnit
         changes   chan RelationUnitsChange
 }
 
 // RelationUnitsChange holds settings information for newly-added and -changed
 // units, and the names of those newly departed from the relation.
 type RelationUnitsChange struct {
         Changed  map[string]UnitSettings
         Departed []string
 }
 
 // UnitSettings holds information about a service unit's settings within a
 // relation.
 type UnitSettings struct {
         Version  int
         Settings map[string]interface{}
 }
 
 // unitSettingsChange is used internally by relationUnitsWatcher to communicate
 // information about a particular unit's settings within a relation.
 type unitSettingsChange struct {
-        name     string
+        unit *relationUnit
         settings UnitSettings
+        eof bool
+        err error
 }
 
 // newRelationUnitsWatcher returns a relationUnitsWatcher which notifies of
 // all presence and settings changes to units playing role within scope,
 // excluding the given unit.
-func newRelationUnitsWatcher(scope unitScopePath, role RelationRole, u *Unit) *relationUnitsWatcher {
+func newRelationUnitsWatcher(scope unitScopePath, role RelationRole, u *Unit, stop <-chan struct{}) *relationUnitsWatcher {
         w := &relationUnitsWatcher{
                 st:        u.st,
                 role:      role,
                 scope:     scope,
                 ignore:    u.key,
-                names:     make(map[string]string),
-                unitTombs: make(map[string]*tomb.Tomb),
+                units: make(map[string]*relationUnit),
                 updates:   make(chan unitSettingsChange),
                 changes:   make(chan RelationUnitsChange),
+                stop: stop,
         }
         go w.loop()
         return w
 }
 
 func (w *relationUnitsWatcher) loop() {
         defer w.finish()
-        roleWatcher := presence.NewChildrenWatcher(w.st.zk, w.scope.presencePath(w.role, ""))
-        defer watcher.Stop(roleWatcher, &w.tomb)
+        roleStop := make(chan struct{})
+        roleWatcher := presence.NewChildrenWatcher(w.st.zk, w.scope.presencePath(w.role, ""), roleStop)
+        defer func() {
+                close(roleStop)
+                if err := roleWatcher.Wait(); err != nil && w.err == nil {
+                        w.err = err
+                }
+        }()
         emittedValue := false
         for {
-                var err error
                 var change RelationUnitsChange
                 select {
-                case <-w.tomb.Dying():
+                case <-w.stop:
                         return
                 case ch, ok := <-roleWatcher.Changes():
+                        var err error
                         if !ok {
                                 err = watcher.MustErr(roleWatcher)
                         } else {
                                 change, err = w.updateWatches(ch)
                         }
                         if err != nil {
-                                w.tomb.Kill(err)
+                                w.err = err
                                 return
                         }
                         if emittedValue && len(change.Changed) == 0 && len(change.Departed) == 0 {
                                 continue
                         }
-                case ch, ok := <-w.updates:
-                        if !ok {
-                                panic("updates channel closed")
+                case ch := <-w.updates:
+                        // If the unit has been stopped, it may yet produce an event,
+                        // which we ignore, as we've already sent the Departed change.
+                        if ch.unit.stopped {
+                                continue
                         }
                         change = RelationUnitsChange{
-                                Changed: map[string]UnitSettings{ch.name: ch.settings},
+                                Changed: map[string]UnitSettings{ch.unit.name: ch.settings},
                         }
                 }
-                select {
-                case <-w.tomb.Dying():
-                        return
-                case w.changes <- change:
-                        emittedValue = true
-                }
+                w.changes <- change
+                emittedValue = true
         }
 }
 
 func (w *relationUnitsWatcher) finish() {
-        for _, t := range w.unitTombs {
-                t.Kill(nil)
-                w.tomb.Kill(t.Wait())
+        for _, u := range w.units {
+                close(u.stop)
         }
-        close(w.updates)
+        for len(w.units) > 0 {
+                ch := <-w.updates
+                if !ch.eof {
+                        continue
+                }
+                if ch.err != nil && w.err == nil {
+                        w.err = ch.err
+                }
+                delete(w.units, ch.unit.key)
+        }
         close(w.changes)
-        w.tomb.Done()
 }
 
-// Stop stops the watcher and returns any errors encountered while watching.
-func (w *relationUnitsWatcher) Stop() error {
-        w.tomb.Kill(nil)
-        return w.tomb.Wait()
-}
-
-// Dying returns a channel that is closed when the
-// watcher has stopped or is about to stop.
-func (w *relationUnitsWatcher) Dying() <-chan struct{} {
-        return w.tomb.Dying()
-}
-
-// Err returns any error encountered while stopping the watcher, or
-// tome.ErrStillAlive if the watcher is still running.
-func (w *relationUnitsWatcher) Err() error {
-        return w.tomb.Err()
+// Wait consumes all remaining changes and
+// returns any error encountered while the watcher
+// was running.
+func (w *relationUnitsWatcher) Wait() error {
+        for _ = range w.changes {
+        }
+        return w.err
 }
 
 // Changes returns a channel that will receive the changes to
 // the relation when detected.
 // The first event on the channel holds the initial state of the
 // relation in its Changed field.
 func (w *relationUnitsWatcher) Changes() <-chan RelationUnitsChange {
         return w.changes
 }
 
 // updateWatches starts or stops watches on the settings of the relation
 // units declared present or absent by ch, and returns a RelationUnitsChange
 // event expressing those changes.
 func (w *relationUnitsWatcher) updateWatches(ch watcher.ChildrenChange) (change RelationUnitsChange, err error) {
         for _, key := range ch.Removed {
                 if key == w.ignore {
                         continue
                 }
-                // When we stop a unit settings watcher, we have to wait for its tomb,
-                // lest its latest change (potentially waiting to be sent on the updates
-                // channel) be received and sent on as a RelationUnitsChange event *after*
-                // we notify of its departure in the event we are currently preparing.
-                t := w.unitTombs[key]
-                delete(w.unitTombs, key)
-                t.Kill(nil)
-                if err := t.Wait(); err != nil {
-                        return RelationUnitsChange{}, err
-                }
-                name := w.names[key]
-                delete(w.names, key)
-                change.Departed = append(change.Departed, name)
+                u := w.units[key]
+                close(u.stop)
+                u.stopped = true
+                change.Departed = append(change.Departed, u.name)
         }
         var topo *topology
         for _, key := range ch.Added {
                 if key == w.ignore {
                         continue
                 }
                 if topo == nil {
                         // Create topology lazily; no sense reading it N times for
                         // N added presence nodes where N != 1.
                         if topo, err = readTopology(w.st.zk); err != nil {
                                 return RelationUnitsChange{}, err
                         }
                 }
                 name, err := topo.UnitName(key)
                 if err != nil {
                         return RelationUnitsChange{}, err
                 }
                 // Start watching unit settings, and consume initial event to get
                 // initial settings for the event we're preparing; subsequent
                 // changes will be received on the unitLoop goroutine and sent to
                 // this one via w.updates.
-                w.names[key] = name
-                uw := watcher.NewContentWatcher(w.st.zk, w.scope.settingsPath(key))
-                select {
-                case <-w.tomb.Dying():
-                        return RelationUnitsChange{}, tomb.ErrDying
-                case cch, ok := <-uw.Changes():
-                        if !ok {
-                                return RelationUnitsChange{}, watcher.MustErr(uw)
-                        }
-                        us := UnitSettings{Version: cch.Version}
-                        if err = goyaml.Unmarshal([]byte(cch.Content), &us.Settings); err != nil {
-                                return RelationUnitsChange{}, err
-                        }
-                        if change.Changed == nil {
-                                change.Changed = map[string]UnitSettings{}
-                        }
-                        change.Changed[name] = us
-                        t := &tomb.Tomb{}
-                        w.unitTombs[key] = t
-                        go w.unitLoop(name, uw, t)
+                u := &relationUnit {
+                        rw: w,
+                        key: key,
+                        name: name,
+                        stop: make(chan struct{}),
+                        w: watcher.NewContentWatcher(w.st.zk, w.scope.settingsPath(key)),
                 }
+                cch, ok := <-u.w.Changes()
+                if !ok {
+                        return RelationUnitsChange{}, watcher.MustErr(u.w)
+                }
+                us := UnitSettings{Version: cch.Version}
+                if err = goyaml.Unmarshal([]byte(cch.Content), &us.Settings); err != nil {
+                        return RelationUnitsChange{}, err
+                }
+                if change.Changed == nil {
+                        change.Changed = map[string]UnitSettings{}
+                }
+                change.Changed[name] = us
+                w.units[key] = u
+                go u.loop()
         }
         return change, nil
 }
 
 // unitLoop sends a unitSettingsChange event on w.updates for each ContentChange
 // event received from uw.
-func (w *relationUnitsWatcher) unitLoop(name string, uw *watcher.ContentWatcher, t *tomb.Tomb) {
-        defer t.Done()
-        defer uw.Stop()
-        for {
-                select {
-                case <-t.Dying():
+func (u *relationUnit) loop() {
+        var err error
+        defer func() {
+                u.rw.updates <- unitSettingsChange{unit: u, eof: true, err: err}
+        }()
+        defer u.w.Stop()
+        for ch := range u.w.Changes() {
+                us := UnitSettings{Version: ch.Version}
+                if e := goyaml.Unmarshal([]byte(ch.Content), &us.Settings); e != nil {
+                        err = e
                         return
-                case ch, ok := <-uw.Changes():
-                        if !ok {
-                                w.tomb.Kill(watcher.MustErr(uw))
-                                return
-                        }
-                        us := UnitSettings{Version: ch.Version}
-                        if err := goyaml.Unmarshal([]byte(ch.Content), &us.Settings); err != nil {
-                                w.tomb.Kill(err)
-                                return
-                        }
-                        select {
-                        case <-t.Dying():
-                                return
-                        case w.updates <- unitSettingsChange{name, us}:
-                        }
                 }
+                u.rw.updates <- unitSettingsChange{unit: u, settings: us}
         }
+        err = watcher.MustErr(u.w)
 }