Merge lp:~chipaca/ubuntu-push/client-v0-p11 into lp:ubuntu-push

[John Lenton]

Several things:
* features:
  * Client.doLoop, the puppet master.
* fixes and cleanups:
  * added log_level to client config
  * added the mysterious sessionRetryCh, used in doLoop to avoid a rather common starvation scenario.
  * found a way not to panic out in initSession (not that it's much better)
  * unified logging in the client tests a bit
  * added logging to session's start error states.

Several things:
* features:
  * Client.doLoop, the puppet master.
* fixes and cleanups:
  * added log_level to client config
  * added the mysterious sessionRetryCh, used in doLoop to avoid a rather common starvation scenario.
  * found a way not to panic out in initSession (not that it's much better)
  * unified logging in the client tests a bit
  * added logging to session's start error states.

[Samuele Pedroni (pedronis)]

ok but land with todo markers

=== modified file 'client/client.go'
--- client/client.go 2014-02-04 13:10:18 +0000
+++ client/client.go 2014-02-04 17:34:01 +0000
@@ -142,6 +142,7 @@

 // connectSession kicks off the session connection dance
 func (client *Client) connectSession() {
+ // xxx lp:1276199
  if client.sessionRetrierStopper != nil {
   client.sessionRetrierStopper <- true
   client.sessionRetrierStopper = nil
@@ -156,6 +157,7 @@

 // disconnectSession disconnects the session
 func (client *Client) disconnectSession() {
+ // xxx lp:1276199
  if client.sessionRetrierStopper != nil {
   client.sessionRetrierStopper <- true

[Samuele Pedroni (pedronis)]

// ClientConfig holds the client configuration
type ClientConfig struct {
        connectivity.ConnectivityConfig // q.v.
        // A reasonably large maximum ping time

that's probably more: A reasonably large timeout for receive/answer pairs

[Samuele Pedroni (pedronis)]

ok, but see bug/markers above and as discussed

review: Approve

[lp:~chipaca/ubuntu-push/client-v0-p11]

60. By John Lenton on 2014-02-04

added XXX comments, and fixed ExchangeTimeout docstring, as per pedronis suggestions

[John Lenton (chipaca)]

Done. Also added auto-highlighting of XXX comments to my .emacs... :-)

=== modified file 'client/client.go'
--- client/client.go	2014-02-04 18:19:28 +0000
+++ client/client.go	2014-02-04 18:19:28 +0000
@@ -39,12 +39,14 @@
 // ClientConfig holds the client configuration
 type ClientConfig struct {
 	connectivity.ConnectivityConfig // q.v.
-	// A reasonably large maximum ping time
+	// A reasonably larg timeout for receive/answer pairs
 	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
 	// The server to connect to
 	Addr config.ConfigHostPort
 	// The PEM-encoded server certificate
 	CertPEMFile string `json:"cert_pem_file"`
+	// The logging level (one of "debug", "info", "error")
+	LogLevel string `json:"log_level"`
 }
 
 // Client is the Ubuntu Push Notifications client-side daemon.
@@ -62,6 +64,7 @@
 	actionsCh             <-chan notifications.RawActionReply
 	session               *session.ClientSession
 	sessionRetrierStopper chan bool
+	sessionRetryCh        chan uint32
 }
 
 // Configure loads the configuration specified in configPath, and sets it up.
@@ -74,8 +77,8 @@
 	if err != nil {
 		return fmt.Errorf("reading config: %v", err)
 	}
-	// later, we'll be specifying logging options in the config file
-	client.log = logger.NewSimpleLogger(os.Stderr, "error")
+	// later, we'll be specifying more logging options in the config file
+	client.log = logger.NewSimpleLogger(os.Stderr, client.config.LogLevel)
 
 	// overridden for testing
 	client.idder = identifier.New()
@@ -84,6 +87,7 @@
 	client.connectivityEndp = bus.SystemBus.Endpoint(networkmanager.BusAddress, client.log)
 
 	client.connCh = make(chan bool)
+	client.sessionRetryCh = make(chan uint32)
 
 	if client.config.CertPEMFile != "" {
 		client.pem, err = ioutil.ReadFile(client.config.CertPEMFile)
@@ -126,17 +130,19 @@
 }
 
 // initSession creates the session object
-func (client *Client) initSession() {
+func (client *Client) initSession() error {
 	sess, err := session.NewSession(string(client.config.Addr), client.pem,
 		client.config.ExchangeTimeout.Duration, client.deviceId, client.log)
 	if err != nil {
-		panic("Don't know how to handle session creation failure.")
+		return err
 	}
 	client.session = sess
+	return nil
 }
 
 // connectSession kicks off the session connection dance
 func (client *Client) connectSession() {
+	// XXX: lp:1276199
 	if client.sessionRetrierStopper != nil {
 		client.sessionRetrierStopper <- true
 		client.sessionRetrierStopper = nil
@@ -146,11 +152,12 @@
 		client.session.Dial,
 		util.Jitter}
 	client.sessionRetrierStopper = ar.Stop
-	go ar.Retry()
+	go func() { client.sessionRetryCh <- ar.Retry() }()
 }
 
 // disconnectSession disconnects the session
 func (client *Client) disconnectSession() {
+	// XXX: lp:1276199
 	if client.sessionRetrierStopper != nil {
 		client.sessionRetrierStopper <- true
 		client.sessionRetrierStopper = nil
@@ -212,3 +219,21 @@
 	urld := urldispatcher.New(client.urlDispatcherEndp, client.log)
 	return urld.DispatchURL("settings:///system/system-update")
 }
+
+// doLoop connects events with their handlers
+func (client *Client) doLoop(connhandler func(bool), clickhandler, notifhandler func() error, errhandler func(error)) {
+	for {
+		select {
+		case state := <-client.connCh:
+			connhandler(state)
+		case <-client.actionsCh:
+			clickhandler()
+		case <-client.session.MsgCh:
+			notifhandler()
+		case err := <-client.session.ErrCh:
+			errhandler(err)
+		case count := <-client.sessionRetryCh:
+			client.log.Debugf("Session connected after %d attempts", count)
+		}
+	}
+}

=== modified file 'client/client_test.go'
--- client/client_test.go	2014-02-04 18:19:28 +0000
+++ client/client_test.go	2014-02-04 18:19:28 +0000
@@ -18,10 +18,12 @@
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	. "launchpad.net/gocheck"
 	"launchpad.net/ubuntu-push/bus/networkmanager"
+	"launchpad.net/ubuntu-push/bus/notifications"
 	testibus "launchpad.net/ubuntu-push/bus/testing"
 	"launchpad.net/ubuntu-push/client/session"
 	"launchpad.net/ubuntu-push/logger"
@@ -57,6 +59,7 @@
 
 var nullog = logger.NewSimpleLogger(ioutil.Discard, "error")
 var noisylog = logger.NewSimpleLogger(os.Stderr, "debug")
+var debuglog = noisylog
 var _ = Suite(&clientSuite{})
 
 const (
@@ -82,6 +85,7 @@
 }
 
 func (cs *clientSuite) SetUpTest(c *C) {
+	debuglog.Debugf("---")
 	dir := c.MkDir()
 	cs.configPath = filepath.Join(dir, "config")
 	cfg := fmt.Sprintf(`
@@ -92,7 +96,8 @@
     "connectivity_check_md5": "",
     "addr": ":0",
     "cert_pem_file": %#v,
-    "recheck_timeout": "3h"
+    "recheck_timeout": "3h",
+    "log_level": "debug"
 }`, helpers.SourceRelative("../server/acceptance/config/testing.cert"))
 	ioutil.WriteFile(cs.configPath, []byte(cfg), 0600)
 }
@@ -205,6 +210,7 @@
 
 func (cs *clientSuite) TestGetDeviceIdWorks(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	cli.idder = identifier.New()
 	c.Check(cli.deviceId, Equals, "")
 	c.Check(cli.getDeviceId(), IsNil)
@@ -213,6 +219,7 @@
 
 func (cs *clientSuite) TestGetDeviceIdCanFail(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	cli.idder = idtesting.Failing()
 	c.Check(cli.deviceId, Equals, "")
 	c.Check(cli.getDeviceId(), NotNil)
@@ -237,8 +244,10 @@
 	cEndp := testibus.NewTestingEndpoint(cCond, condition.Work(true),
 		uint32(networkmanager.ConnectedGlobal),
 	)
+	testibus.SetWatchTicker(cEndp, make(chan bool))
 	// ok, create the thing
 	cli := new(Client)
+	cli.log = debuglog
 	err := cli.Configure(cs.configPath)
 	c.Assert(err, IsNil)
 	// the user actions channel has not been set up
@@ -267,6 +276,7 @@
 func (cs *clientSuite) TestTakeTheBusCanFail(c *C) {
 	cli := new(Client)
 	err := cli.Configure(cs.configPath)
+	cli.log = debuglog
 	c.Assert(err, IsNil)
 	// the user actions channel has not been set up
 	c.Check(cli.actionsCh, IsNil)
@@ -285,15 +295,13 @@
 ******************************************************************/
 
 func (cs *clientSuite) TestHandleErr(c *C) {
+	buf := &bytes.Buffer{}
 	cli := new(Client)
-	cli.log = noisylog
+	cli.log = logger.NewSimpleLogger(buf, "debug")
 	cli.initSession()
 	cli.hasConnectivity = true
-	cli.handleErr(nil)
-	c.Assert(cli.session, NotNil)
-	// let the session connection fail
-	time.Sleep(100 * time.Millisecond)
-	c.Check(cli.session.State(), Equals, session.Error)
+	cli.handleErr(errors.New("bananas"))
+	c.Check(buf.String(), Matches, ".*session exited.*bananas\n")
 }
 
 /*****************************************************************
@@ -302,19 +310,23 @@
 
 func (cs *clientSuite) TestHandleConnStateD2C(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	cli.initSession()
 	// let's pretend the client had a previous attempt at connecting still pending
 	// (hard to trigger in real life, but possible)
-	cli.sessionRetrierStopper = make(chan bool, 1)
+	ch := make(chan bool, 1)
+	cli.sessionRetrierStopper = ch
 
 	c.Assert(cli.hasConnectivity, Equals, false)
 	cli.handleConnState(true)
+	c.Check(len(ch), Equals, 1)
 	c.Check(cli.hasConnectivity, Equals, true)
 	c.Assert(cli.session, NotNil)
 }
 
 func (cs *clientSuite) TestHandleConnStateSame(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	// here we want to check that we don't do anything
 	c.Assert(cli.session, IsNil)
 	c.Assert(cli.hasConnectivity, Equals, false)
@@ -328,7 +340,8 @@
 
 func (cs *clientSuite) TestHandleConnStateC2D(c *C) {
 	cli := new(Client)
-	cli.session, _ = session.NewSession(string(cli.config.Addr), cli.pem, cli.config.ExchangeTimeout.Duration, cli.deviceId, noisylog)
+	cli.log = debuglog
+	cli.session, _ = session.NewSession(string(cli.config.Addr), cli.pem, cli.config.ExchangeTimeout.Duration, cli.deviceId, debuglog)
 	cli.session.Dial()
 	cli.hasConnectivity = true
 
@@ -340,7 +353,8 @@
 
 func (cs *clientSuite) TestHandleConnStateC2DPending(c *C) {
 	cli := new(Client)
-	cli.session, _ = session.NewSession(string(cli.config.Addr), cli.pem, cli.config.ExchangeTimeout.Duration, cli.deviceId, noisylog)
+	cli.log = debuglog
+	cli.session, _ = session.NewSession(string(cli.config.Addr), cli.pem, cli.config.ExchangeTimeout.Duration, cli.deviceId, debuglog)
 	cli.sessionRetrierStopper = make(chan bool, 1)
 	cli.hasConnectivity = true
 
@@ -369,9 +383,9 @@
 
 func (cs *clientSuite) TestHandleNotificationFail(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(false))
 	cli.notificationsEndp = endp
-	cli.log = noisylog
 	c.Check(cli.handleNotification(), NotNil)
 }
 
@@ -381,9 +395,9 @@
 
 func (cs *clientSuite) TestHandleClick(c *C) {
 	cli := new(Client)
+	cli.log = debuglog
 	endp := testibus.NewTestingEndpoint(nil, condition.Work(true), nil)
 	cli.urlDispatcherEndp = endp
-	cli.log = noisylog
 	c.Check(cli.handleClick(), IsNil)
 	// check we sent the notification
 	args := testibus.GetCallArgs(endp)
@@ -391,3 +405,56 @@
 	c.Check(args[0].Member, Equals, "DispatchURL")
 	c.Check(args[0].Args, DeepEquals, []interface{}{"settings:///system/system-update"})
 }
+
+/*****************************************************************
+    doLoop tests
+******************************************************************/
+
+func (cs *clientSuite) TestDoLoopConn(c *C) {
+	cli := new(Client)
+	cli.log = debuglog
+	cli.connCh = make(chan bool, 1)
+	cli.connCh <- true
+	cli.initSession()
+
+	ch := make(chan bool, 1)
+	go cli.doLoop(func(bool) { ch <- true }, func() error { return nil }, func() error { return nil }, func(error) {})
+	c.Check(takeNextBool(ch), Equals, true)
+}
+
+func (cs *clientSuite) TestDoLoopClick(c *C) {
+	cli := new(Client)
+	cli.log = debuglog
+	cli.initSession()
+	aCh := make(chan notifications.RawActionReply, 1)
+	aCh <- notifications.RawActionReply{}
+	cli.actionsCh = aCh
+
+	ch := make(chan bool, 1)
+	go cli.doLoop(func(bool) {}, func() error { ch <- true; return nil }, func() error { return nil }, func(error) {})
+	c.Check(takeNextBool(ch), Equals, true)
+}
+
+func (cs *clientSuite) TestDoLoopNotif(c *C) {
+	cli := new(Client)
+	cli.log = debuglog
+	cli.initSession()
+	cli.session.MsgCh = make(chan *session.Notification, 1)
+	cli.session.MsgCh <- &session.Notification{}
+
+	ch := make(chan bool, 1)
+	go cli.doLoop(func(bool) {}, func() error { return nil }, func() error { ch <- true; return nil }, func(error) {})
+	c.Check(takeNextBool(ch), Equals, true)
+}
+
+func (cs *clientSuite) TestDoLoopErr(c *C) {
+	cli := new(Client)
+	cli.log = debuglog
+	cli.initSession()
+	cli.session.ErrCh = make(chan error, 1)
+	cli.session.ErrCh <- nil
+
+	ch := make(chan bool, 1)
+	go cli.doLoop(func(bool) {}, func() error { return nil }, func() error { return nil }, func(error) { ch <- true })
+	c.Check(takeNextBool(ch), Equals, true)
+}

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-02-04 13:02:38 +0000
+++ client/session/session.go	2014-02-04 18:19:28 +0000
@@ -147,14 +147,17 @@
 	err := sess.proto.WriteMessage(protocol.AckMsg{"ack"})
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to ack broadcast: %s", err)
 		return err
 	}
 	sess.Log.Debugf("broadcast chan:%v app:%v topLevel:%d payloads:%s",
 		bcast.ChanId, bcast.AppId, bcast.TopLevel, bcast.Payloads)
 	if bcast.ChanId == protocol.SystemChannelId {
 		// the system channel id, the only one we care about for now
+		sess.Log.Debugf("sending it over")
 		sess.Levels.Set(bcast.ChanId, bcast.TopLevel)
 		sess.MsgCh <- &Notification{}
+		sess.Log.Debugf("sent it over")
 	} else {
 		sess.Log.Debugf("what is this weird channel, %#v?", bcast.ChanId)
 	}
@@ -192,6 +195,7 @@
 	err := conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to start: set deadline: %s", err)
 		return err
 	}
 	_, err = conn.Write(wireVersionBytes)
@@ -199,6 +203,7 @@
 	// n < len(p). So, no need to check number of bytes written, hooray.
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to start: write version: %s", err)
 		return err
 	}
 	proto := sess.Protocolator(conn)
@@ -210,12 +215,14 @@
 	})
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to start: connect: %s", err)
 		return err
 	}
 	var connAck protocol.ConnAckMsg
 	err = proto.ReadMessage(&connAck)
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to start: connack: %s", err)
 		return err
 	}
 	if connAck.Type != "connack" {
@@ -225,6 +232,7 @@
 	pingInterval, err := time.ParseDuration(connAck.Params.PingInterval)
 	if err != nil {
 		sess.setState(Error)
+		sess.Log.Errorf("unable to start: parse ping interval: %s", err)
 		return err
 	}
 	sess.proto = proto