state/api: New client API watchers; api.State

state/apiserver/apiserver.go

 // Copyright 2013 Canonical Ltd.
 // Licensed under the AGPLv3, see LICENCE file for details.
 
 package apiserver
 
 import (
         "fmt"
         "launchpad.net/juju-core/charm"
         "launchpad.net/juju-core/juju"
         "launchpad.net/juju-core/log"
         "launchpad.net/juju-core/state"
         "launchpad.net/juju-core/state/api"
         "launchpad.net/juju-core/state/api/params"
         "launchpad.net/juju-core/state/multiwatcher"
         "launchpad.net/juju-core/state/presence"
         "launchpad.net/juju-core/state/statecmd"
         statewatcher "launchpad.net/juju-core/state/watcher"
         "strconv"
         "sync"
 )
 
 // srvRoot represents a single client's connection to the state.
 type srvRoot struct {
         admin     *srvAdmin
         client    *srvClient
+        state     *srvState
         srv       *Server
         resources *resources
 
         user authUser
 }
 
 // srvAdmin is the only object that unlogged-in
 // clients can access. It holds any methods
 // that are needed to log in.
 type srvAdmin struct {
@@ skipping 16 matching lines @@
 type srvUser struct {
         root *srvRoot
         u    *state.User
 }
 
 // srvClient serves client-specific API methods.
 type srvClient struct {
         root *srvRoot
 }
 
+// srvState serves agent-specific top-level state API methods.
+type srvState struct {
+        root *srvRoot
+}
+
+// WatchMachines registers a srvLifecycleWatcher that notifies of
+// changes to the lifecycles of the machines in the environment. The
+// result contains the id of the registered watcher and the initial
+// list of machine ids.
+func (s *srvState) WatchMachines() (params.LifecycleWatchResults, error) {

[rog, 2013/05/24 13:35:13]

these two methods should be further down the file, perhaps just before the
srvMachine methods.

also, i'm not sure we want to allow any agent to watch all machines - perhaps
only an EnvironManager machine?

[dimitern, 2013/05/24 14:07:44]

Done.

+        watcher := s.root.srv.state.WatchMachines()
+        // To save an extra round-trip to call Next after Watch, we check
+        // for initial changes.
+        initial, ok := <-watcher.Changes()
+        if !ok {
+                return params.LifecycleWatchResults{}, statewatcher.MustErr(watcher)
+        }
+        return params.LifecycleWatchResults{
+                LifecycleWatcherId: s.root.resources.register(watcher).id,
+                Ids:                initial,
+        }, nil
+}
+
+// WatchEnvironConfig registers a srvEnvironConfigWatcher for
+// observing changes to the environment configuration. The result
+// contains the id of the registered watcher and the current
+// environment configuration.
+func (s *srvState) WatchEnvironConfig() (params.EnvironConfigWatchResults, error) {
+        watcher := s.root.srv.state.WatchEnvironConfig()

[rog, 2013/05/24 13:35:13]

// TODO restrict access to environment manager machines only.

?

[dimitern, 2013/05/24 14:07:44]

No need, I added the authEnvironManager check anyway.

+        // To save an extra round-trip to call Next after Watch, we check

[rog, 2013/05/24 13:35:13]

// The watcher always sends an initial value on the channel,
// so we send that as the result of the watch request.
// This saves the client a round trip.
?

[dimitern, 2013/05/24 14:07:44]

Done. Also updated the other places.

+        // for initial changes.
+        initial, ok := <-watcher.Changes()
+        if !ok {
+                return params.EnvironConfigWatchResults{}, statewatcher.MustErr(watcher)
+        }
+        return params.EnvironConfigWatchResults{
+                EnvironConfigWatcherId: s.root.resources.register(watcher).id,
+                Config:                 initial.AllAttrs(),
+        }, nil
+}
+
 func newStateServer(srv *Server) *srvRoot {
         r := &srvRoot{
                 srv:       srv,
                 resources: newResources(),
         }
         r.admin = &srvAdmin{
                 root: r,
         }
         r.client = &srvClient{
                 root: r,
         }
+        r.state = &srvState{
+                root: r,
+        }
         return r
 }
 
 // Kill implements rpc.Killer.  It cleans up any resources that need
 // cleaning up to ensure that all outstanding requests return.
 func (r *srvRoot) Kill() {
         r.resources.stopAll()
 }
 
 // Admin returns an object that provides API access
@@ skipping 121 matching lines @@
         watcher := r.resources.get(id)
         if watcher == nil {
                 return srvEntityWatcher{}, errUnknownWatcher
         }
         if _, ok := watcher.resource.(*state.EntityWatcher); !ok {
                 return srvEntityWatcher{}, errUnknownWatcher
         }
         return srvEntityWatcher{watcher}, nil
 }
 
+// LifecycleWatcher returns an object that provides
+// API access to methods on a state.LifecycleWatcher.
+// Each client has its own current set of watchers, stored
+// in r.resources.
+func (r *srvRoot) LifecycleWatcher(id string) (srvLifecycleWatcher, error) {
+        if err := r.requireAgent(); err != nil {
+                return srvLifecycleWatcher{}, err
+        }
+        watcher := r.resources.get(id)
+        if watcher == nil {
+                return srvLifecycleWatcher{}, errUnknownWatcher
+        }
+        if _, ok := watcher.resource.(*state.LifecycleWatcher); !ok {
+                return srvLifecycleWatcher{}, errUnknownWatcher
+        }
+        return srvLifecycleWatcher{watcher}, nil
+}
+
+// EnvironConfigWatcher returns an object that provides
+// API access to methods on a state.EnvironConfigWatcher.
+// Each client has its own current set of watchers, stored
+// in r.resources.
+func (r *srvRoot) EnvironConfigWatcher(id string) (srvEnvironConfigWatcher, error) {
+        if err := r.requireAgent(); err != nil {
+                return srvEnvironConfigWatcher{}, err
+        }
+        watcher := r.resources.get(id)
+        if watcher == nil {
+                return srvEnvironConfigWatcher{}, errUnknownWatcher
+        }
+        if _, ok := watcher.resource.(*state.EnvironConfigWatcher); !ok {
+                return srvEnvironConfigWatcher{}, errUnknownWatcher
+        }
+        return srvEnvironConfigWatcher{watcher}, nil
+}
+
 // AllWatcher returns an object that provides API access to methods on
 // a state/multiwatcher.Watcher, which watches any changes to the
 // state. Each client has its own current set of watchers, stored in
 // r.resources.
 func (r *srvRoot) AllWatcher(id string) (srvClientAllWatcher, error) {
         if err := r.requireClient(); err != nil {
                 return srvClientAllWatcher{}, err
         }
         watcher := r.resources.get(id)
         if watcher == nil {
                 return srvClientAllWatcher{}, errUnknownWatcher
         }
         if _, ok := watcher.resource.(*multiwatcher.Watcher); !ok {
                 return srvClientAllWatcher{}, errUnknownWatcher
         }
         return srvClientAllWatcher{watcher}, nil
 
 }
 
+// State returns an object that provides API access to top-level state methods.
+func (r *srvRoot) State(id string) (*srvState, error) {
+        if err := r.requireAgent(); err != nil {
+                return nil, err
+        }
+        if id != "" {
+                // Safeguard id for possible future use.
+                return nil, errBadId
+        }
+        return r.state, nil
+}
+
 // Client returns an object that provides access
 // to methods accessible to non-agent clients.
 func (r *srvRoot) Client(id string) (*srvClient, error) {
         if err := r.requireClient(); err != nil {
                 return nil, err
         }
         if id != "" {
                 // Safeguard id for possible future use.
                 return nil, errBadId
         }
@@ skipping 30 matching lines @@
         if _, ok := <-watcher.Changes(); ok {
                 return nil
         }
         err := watcher.Err()
         if err == nil {
                 err = errStoppedWatcher
         }
         return err
 }
 
+// srvLifecycleWatcher notifies about lifecycle changes for all
+// entities of a given kind. See state.LifecycleWatcher.
+type srvLifecycleWatcher struct {
+        *srvResource
+}
+
+// Next returns when a change has occured to the lifecycle of an
+// entity of the collection being watched since the most recent call
+// to Next or the Watch call that created the srvLifecycleWatcher.
+func (w srvLifecycleWatcher) Next() (params.LifecycleWatchResults, error) {
+        watcher := w.resource.(*state.LifecycleWatcher)
+        if changes, ok := <-watcher.Changes(); ok {
+                return params.LifecycleWatchResults{
+                        Ids: changes,
+                }, nil
+        }
+        err := watcher.Err()
+        if err == nil {
+                err = errStoppedWatcher
+        }
+        return params.LifecycleWatchResults{}, err
+}
+
+// srvEnvironConfigWatcher notifies about changes to the environment
+// configuration. See state.EnvironConfigWatcher.
+type srvEnvironConfigWatcher struct {
+        *srvResource
+}
+
+// Next returns when a change has occured to the environment
+// configuration since the most recent call to Next or the Watch call
+// that created the srvEnvironConfigWatcher.
+func (w srvEnvironConfigWatcher) Next() (params.EnvironConfigWatchResults, error) {
+        watcher := w.resource.(*state.EnvironConfigWatcher)
+        if changes, ok := <-watcher.Changes(); ok {
+                return params.EnvironConfigWatchResults{
+                        Config: changes.AllAttrs(),
+                }, nil
+        }
+        err := watcher.Err()
+        if err == nil {
+                err = errStoppedWatcher
+        }
+        return params.EnvironConfigWatchResults{}, err
+}
+
 func (c *srvClient) Status() (api.Status, error) {
         ms, err := c.root.srv.state.AllMachines()
         if err != nil {
                 return api.Status{}, err
         }
         status := api.Status{
                 Machines: make(map[string]api.MachineInfo),
         }
         for _, m := range ms {
                 instId, _ := m.InstanceId()
@@ skipping 220 matching lines @@
 
 // Get retrieves all the details of a machine.
 func (m *srvMachine) Get() (info params.Machine) {
         instId, _ := m.m.InstanceId()
         info.InstanceId = string(instId)
         info.Life = params.Life(m.m.Life().String())
         return
 }
 
 func (m *srvMachine) Watch() (params.EntityWatcherId, error) {
-        w := m.m.Watch()
-        if _, ok := <-w.Changes(); !ok {
-                return params.EntityWatcherId{}, statewatcher.MustErr(w)
+        watcher := m.m.Watch()
+        // To save an extra round-trip to call Next after Watch, we check
+        // for initial changes.
+        if _, ok := <-watcher.Changes(); !ok {
+                return params.EntityWatcherId{}, statewatcher.MustErr(watcher)
         }
         return params.EntityWatcherId{
-                EntityWatcherId: m.root.resources.register(w).id,
+                EntityWatcherId: m.root.resources.register(watcher).id,
         }, nil
 }
 
 // SetAgentAlive signals that the agent for machine m is alive.
 func (m *srvMachine) SetAgentAlive() (params.PingerId, error) {
         if !m.root.authOwner(m.m) {
                 return params.PingerId{}, errPerm
         }
         pinger, err := m.m.SetAgentAlive()
         if err != nil {
@@ skipping 202 matching lines @@
 func (rs *resources) stopAll() {
         rs.mu.Lock()
         defer rs.mu.Unlock()
         for _, r := range rs.rs {
                 if err := r.resource.Stop(); err != nil {
                         log.Errorf("state/api: error stopping %T resource: %v", r, err)
                 }
         }
         rs.rs = make(map[string]*srvResource)
 }