api/state: implement Server.Stop

state/api/serve.go

 package api
 
 import (
         "code.google.com/p/go.net/websocket"
         "crypto/tls"
         "launchpad.net/juju-core/log"
         "launchpad.net/juju-core/state"
+        "launchpad.net/tomb"
         "net"
         "net/http"
+        "sync"
 )
 
 // srvState represents a single client's connection to the state.
 type srvState struct {
-        state *state.State
-        conn  *websocket.Conn
+        srv  *Server
+        conn *websocket.Conn
 }
 
 type Server struct {
+        tomb  tomb.Tomb
+        wg    sync.WaitGroup
         state *state.State
+        addr  net.Addr
 }
 
 // Serve serves the given state by accepting requests
 // on the given listener, using the given certificate
 // and key (in PEM format) for authentication.·
-func Serve(s *state.State, lis net.Listener, cert, key []byte) error {
+func NewServer(s *state.State, addr string, cert, key []byte) (*Server, error) {
+        lis, err := net.Listen("tcp", addr)
+        if err != nil {
+                return nil, err
+        }
         tlsCert, err := tls.X509KeyPair(cert, key)
         if err != nil {
-                return err
+                return nil, err
+        }
+        srv := &Server{
+                state: s,
+                addr:  lis.Addr(),
         }
         lis = tls.NewListener(lis, &tls.Config{
                 Certificates: []tls.Certificate{tlsCert},
         })
-        return http.Serve(lis, newHandler(s))
+        go func() {
+                defer srv.tomb.Done()
+                srv.tomb.Kill(srv.run(lis))

[dfc, 2012/12/10 21:53:06]

nice

+        }()
+        return srv, nil
 }
 
-// newHandler returns an http handler that serves the API
-// interface to the given state as a websocket.
-func newHandler(s *state.State) http.Handler {
-        return websocket.Handler(func(conn *websocket.Conn) {
-                srv := &srvState{
-                        state: s,
-                        conn:  conn,
+// Stop stops the server and returns when all requests that

[dfc, 2012/12/10 21:53:06]

s/when../once all outstanding requests have been completed ?

[fwereade, 2012/12/11 09:16:33]

+1

+// it is running have completed.
+func (srv *Server) Stop() error {
+        srv.tomb.Kill(nil)
+        return srv.tomb.Wait()
+}
+
+func (srv *Server) run(lis net.Listener) error {
+        srv.wg.Add(1)
+        go func() {
+                <-srv.tomb.Dying()
+                lis.Close()
+                srv.wg.Done()

[dfc, 2012/12/10 21:53:06]

defer this line ? lis.Close() may panic, although unlikely

[fwereade, 2012/12/11 09:16:33]

+1 also

[rog, 2012/12/13 17:16:22]

lis.Close cannot panic (we've checked the listener is ok elsewhere), and if it
does, the whole system goes down regardless anyway, so the defer doesn't help at
all.

i prefer the simpler straight-line logic if that's ok.

+        }()
+        handler := websocket.Handler(func(conn *websocket.Conn) {
+                srv.wg.Add(1)
+                defer srv.wg.Done()

[niemeyer, 2012/12/13 16:05:55]

Why do we have two Adds and two Dones for a single call? The logic below doesn't
seem to justify that.

+                // If we've got to this stage and the tomb is still
+                // alive, we know that any tomb.Kill must occur after we
+                // have called wg.Add, so we avoid the possibility of a
+                // handler goroutine running after Stop has returned.
+                if srv.tomb.Err() != tomb.ErrStillAlive {
+                        return
                 }
-                srv.run()
+                st := &srvState{
+                        srv:  srv,
+                        conn: conn,
+                }
+                srv.wg.Add(1)
+                go func() {
+                        st.run()
+                        srv.wg.Done()
+                }()
+                <-srv.tomb.Dying()

[niemeyer, 2012/12/13 16:05:55]

Won't this block every single goroutine per connection made until the server is
closed?

+                conn.Close()
         })
+        // The error from http.Serve is not interesting.
+        http.Serve(lis, handler)
+        return nil
+}
+
+// Addr returns the address that the server is listening on.
+func (srv *Server) Addr() string {
+        return srv.addr.String()
 }
 
 type rpcRequest struct {
         Request string // placeholder only
 }
 
 type rpcResponse struct {
         Response string // placeholder only
         Error    string
 }
 
-func (srv *srvState) run() {
+func (st *srvState) run() {
         for {
                 var req rpcRequest
-                err := websocket.JSON.Receive(srv.conn, &req)
+                err := websocket.JSON.Receive(st.conn, &req)
                 if err != nil {
                         log.Printf("api: error receiving request: %v", err)
                         return
                 }
                 var resp rpcResponse
                 // placeholder for executing some arbitrary operation
                 // on state.
-                m, err := srv.state.Machine(req.Request)
+                m, err := st.srv.state.Machine(req.Request)
                 if err != nil {
                         resp.Error = err.Error()
                 } else {
                         instId, err := m.InstanceId()
                         if err != nil {
                                 resp.Error = err.Error()
                         } else {
                                 resp.Response = string(instId)
                         }
                 }
-                err = websocket.JSON.Send(srv.conn, &resp)
+                err = websocket.JSON.Send(st.conn, &resp)
                 if err != nil {
                         log.Printf("api: error sending reply: %v", err)
                         return
                 }
         }
 }