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

[John Lenton]

The client (session), v0.

The client (session), v0.

A lot of comonality with some of the testing bits of protocol and
server/session; later, let's see if we can refactor.

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

25. By John Lenton on 2014-01-25

a few cleanups

26. By John Lenton on 2014-01-25

Moving it into client/session because that's what it is.

[Samuele Pedroni (pedronis)]

I think it would be saner for the testing if Config was an interface and not the concrete thing, I don't think it's the task of the session to read pem files.

same for run, it should be split in a start and a loop so that tests can poke at simpler bits

review: Needs Fixing

[Samuele Pedroni (pedronis)]

probably run can be split even more,

- in start/setup

- get next message

- acting

- reply

Unmerged revisions

=== added directory 'client'
=== added directory 'client/session'
=== added file 'client/session/session.go'
--- client/session/session.go	1970-01-01 00:00:00 +0000
+++ client/session/session.go	2014-01-25 02:14:54 +0000
@@ -0,0 +1,214 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package session
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"io/ioutil"
+	"launchpad.net/ubuntu-push/config"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	"net"
+	"time"
+)
+
+var wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+
+type Notification struct {
+	// something something something
+}
+
+type LevelMap interface {
+	Set(level string, top int64)
+	GetAll() map[string]int64
+}
+
+type mapLevelMap map[string]int64
+
+func (m *mapLevelMap) Set(level string, top int64) {
+	(*m)[level] = top
+}
+func (m *mapLevelMap) GetAll() map[string]int64 {
+	return map[string]int64(*m)
+}
+
+var _ LevelMap = &mapLevelMap{}
+
+type Config struct {
+	// session configuration
+	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
+	// server connection config
+	Addr        config.ConfigHostPort
+	CertPEMFile string `json:"cert_pem_file"`
+}
+
+// ClienSession holds a client<->server session and its configuration.
+type ClientSession struct {
+	// configuration
+	DeviceId        string
+	ServerAddr      string
+	ExchangeTimeout time.Duration
+	Levels          LevelMap
+	// connection
+	Connection   net.Conn
+	Protocolator func(net.Conn) protocol.Protocol
+	Log          logger.Logger
+	TLS          *tls.Config
+	// status
+	ErrCh chan error
+	MsgCh chan *Notification
+}
+
+func NewSession(config Config, log logger.Logger, deviceId string) (*ClientSession, error) {
+	sess := &ClientSession{
+		ExchangeTimeout: config.ExchangeTimeout.TimeDuration(),
+		ServerAddr:      config.Addr.HostPort(),
+		DeviceId:        deviceId,
+		Log:             log,
+		Protocolator:    protocol.NewProtocol0,
+		Levels:          &mapLevelMap{},
+		TLS:             &tls.Config{InsecureSkipVerify: true}, // XXX
+	}
+	if config.CertPEMFile != "" {
+		cert, err := ioutil.ReadFile(config.CertPEMFile)
+		if err != nil {
+			return nil, err
+		}
+		cp := x509.NewCertPool()
+		ok := cp.AppendCertsFromPEM(cert)
+		if !ok {
+			return nil, errors.New("dial: could not parse certificate")
+		}
+		sess.TLS.RootCAs = cp
+	}
+	return sess, nil
+}
+
+// Dial connects to a server using the configuration in the ClientSession
+// and sets up the connection.
+func (sess *ClientSession) Dial() error {
+	conn, err := net.DialTimeout("tcp", sess.ServerAddr, sess.ExchangeTimeout)
+	if err != nil {
+		return err
+	}
+	sess.Connection = tls.Client(conn, sess.TLS)
+	return nil
+}
+
+type serverMsg struct {
+	Type string `json:"T"`
+	protocol.BroadcastMsg
+	protocol.NotificationsMsg
+}
+
+func (sess *ClientSession) Reset() error {
+	if sess.Protocolator == nil {
+		return errors.New("Can't Reset() without a protocol constructor.")
+	}
+	if sess.Connection != nil {
+		sess.Connection.Close() // just in case
+	}
+	err := sess.Dial()
+	if err != nil {
+		sess.Log.Errorf("%s", err)
+		return err
+	}
+	sess.ErrCh = make(chan error, 1)
+	sess.MsgCh = make(chan *Notification)
+	sess.Run()
+	return nil
+}
+
+func (sess *ClientSession) Run() {
+	go func() { sess.ErrCh <- sess.run() }()
+}
+
+// Run the session with the server, emits a stream of events.
+func (sess *ClientSession) run() error {
+	conn := sess.Connection
+	if conn == nil {
+		return errors.New("Can't run() disconnected.")
+	}
+	if sess.Protocolator == nil {
+		return errors.New("Can't run() without a protocol constructor.")
+	}
+	defer conn.Close()
+	err := conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+	if err != nil {
+		return err
+	}
+	_, err = conn.Write(wireVersionBytes)
+	// The Writer docs: Write must return a non-nil error if it returns
+	// n < len(p). So, no need to check number of bytes written, hooray.
+	if err != nil {
+		return err
+	}
+	proto := sess.Protocolator(conn)
+	err = proto.WriteMessage(protocol.ConnectMsg{
+		Type:     "connect",
+		DeviceId: sess.DeviceId,
+		Levels:   sess.Levels.GetAll(),
+	})
+	if err != nil {
+		return err
+	}
+	var connAck protocol.ConnAckMsg
+	err = proto.ReadMessage(&connAck)
+	if err != nil {
+		return err
+	}
+	pingInterval, err := time.ParseDuration(connAck.Params.PingInterval)
+	if err != nil {
+		return err
+	}
+	sess.Log.Debugf("Connected %v.", conn.LocalAddr())
+	var recv serverMsg
+	for {
+		deadAfter := pingInterval + sess.ExchangeTimeout
+		conn.SetDeadline(time.Now().Add(deadAfter))
+		err = proto.ReadMessage(&recv)
+		if err != nil {
+			return err
+		}
+		switch recv.Type {
+		case "ping":
+			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+			err := proto.WriteMessage(protocol.PingPongMsg{Type: "pong"})
+			if err != nil {
+				return err
+			}
+			sess.Log.Debugf("Ping.")
+		case "broadcast":
+			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+			err := proto.WriteMessage(protocol.PingPongMsg{Type: "ack"})
+			if err != nil {
+				return err
+			}
+			sess.Log.Debugf("broadcast chan:%v app:%v topLevel:%d payloads:%s",
+				recv.ChanId, recv.AppId, recv.TopLevel, recv.Payloads)
+			if recv.ChanId == protocol.SystemChannelId {
+				// the system channel id, the only one we care about for now
+				sess.Levels.Set(recv.ChanId, recv.TopLevel)
+				sess.MsgCh <- &Notification{}
+			} else {
+				sess.Log.Debugf("What is this weird channel, %s?", recv.ChanId)
+			}
+		}
+	}
+}

=== added file 'client/session/session_test.go'
--- client/session/session_test.go	1970-01-01 00:00:00 +0000
+++ client/session/session_test.go	2014-01-25 02:14:54 +0000
@@ -0,0 +1,586 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package session
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	helpers "launchpad.net/ubuntu-push/testing"
+	"launchpad.net/ubuntu-push/testing/condition"
+	"net"
+	"os"
+	"reflect"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSession(t *testing.T) { TestingT(t) }
+
+type clientSessionSuite struct{}
+
+var nullog = logger.NewSimpleLogger(ioutil.Discard, "error")
+var debuglog = logger.NewSimpleLogger(os.Stderr, "debug")
+var _ = Suite(&clientSessionSuite{})
+
+/****************************************************************
+  NewSession() tests
+****************************************************************/
+
+func (cs *clientSessionSuite) TestNewSessionPlainWorks(c *C) {
+	cfg := Config{}
+	sess, err := NewSession(cfg, nullog, "wah")
+	c.Check(sess, NotNil)
+	c.Check(err, IsNil)
+}
+
+var certfile string = helpers.SourceRelative("../../server/acceptance/config/testing.cert")
+
+func (cs *clientSessionSuite) TestNewSessionPEMWorks(c *C) {
+	cfg := Config{CertPEMFile: certfile}
+	sess, err := NewSession(cfg, nullog, "wah")
+	c.Check(sess, NotNil)
+	c.Assert(err, IsNil)
+	c.Check(sess.TLS.RootCAs, NotNil)
+}
+
+func (cs *clientSessionSuite) TestNewSessionBadPEMFilePathFails(c *C) {
+	cfg := Config{CertPEMFile: "/no/such/path"}
+	sess, err := NewSession(cfg, nullog, "wah")
+	c.Check(sess, IsNil)
+	c.Check(err, NotNil)
+}
+
+func (cs *clientSessionSuite) TestNewSessionBadPEMFileContentFails(c *C) {
+	cfg := Config{CertPEMFile: "/etc/passwd"}
+	sess, err := NewSession(cfg, nullog, "wah")
+	c.Check(sess, IsNil)
+	c.Check(err, NotNil)
+}
+
+/****************************************************************
+  Run() tests
+****************************************************************/
+
+func testname() string {
+	pcs := make([]uintptr, 200)
+	runtime.Callers(0, pcs)
+	testname := "<unknown>"
+	for _, pc := range pcs {
+		me := runtime.FuncForPC(pc)
+		if me == nil {
+			break
+		}
+		parts := strings.Split(me.Name(), ".")
+		funcname := parts[len(parts)-1]
+		if strings.HasPrefix(funcname, "Test") {
+			testname = funcname
+		}
+	}
+	return testname
+}
+
+type xAddr string
+
+func (x xAddr) Network() string { return "<:>" }
+func (x xAddr) String() string  { return string(x) }
+
+// testConn (roughly based on the one in protocol_test)
+
+type testConn struct {
+	Name              string
+	Deadlines         []time.Duration
+	Writes            [][]byte
+	WriteCondition    condition.Interface
+	DeadlineCondition condition.Interface
+}
+
+func (tc *testConn) LocalAddr() net.Addr { return xAddr(tc.Name) }
+
+func (tc *testConn) RemoteAddr() net.Addr { return xAddr(tc.Name) }
+
+func (tc *testConn) Close() error { return nil }
+
+func (tc *testConn) SetDeadline(t time.Time) error {
+	tc.Deadlines = append(tc.Deadlines, t.Sub(time.Now()))
+	if tc.DeadlineCondition == nil || tc.DeadlineCondition.OK() {
+		return nil
+	} else {
+		return errors.New("deadliner on fire")
+	}
+}
+
+func (tc *testConn) SetReadDeadline(t time.Time) error  { panic("NIH"); return nil }
+func (tc *testConn) SetWriteDeadline(t time.Time) error { panic("NIH"); return nil }
+func (tc *testConn) Read(buf []byte) (n int, err error) { panic("NIH"); return -1, nil }
+
+func (tc *testConn) Write(buf []byte) (int, error) {
+	store := make([]byte, len(buf))
+	copy(store, buf)
+	tc.Writes = append(tc.Writes, store)
+	if tc.WriteCondition == nil || tc.WriteCondition.OK() {
+		return len(store), nil
+	} else {
+		return -1, errors.New("writer on fire")
+	}
+}
+
+// test protocol (from session_test)
+
+type testProtocol struct {
+	up   chan interface{}
+	down chan interface{}
+}
+
+// takeNext takes a value from given channel with a 5s timeout
+func takeNext(ch <-chan interface{}) interface{} {
+	select {
+	case <-time.After(5 * time.Second):
+		panic("test protocol exchange stuck: too long waiting")
+	case v := <-ch:
+		return v
+	}
+	return nil
+}
+
+func (c *testProtocol) SetDeadline(t time.Time) {
+	deadAfter := t.Sub(time.Now())
+	deadAfter = (deadAfter + time.Millisecond/2) / time.Millisecond * time.Millisecond
+	c.down <- fmt.Sprintf("deadline %v", deadAfter)
+}
+
+func (c *testProtocol) ReadMessage(dest interface{}) error {
+	switch v := takeNext(c.up).(type) {
+	case error:
+		return v
+	default:
+		// make sure JSON.Unmarshal works with dest
+		var marshalledMsg []byte
+		marshalledMsg, err := json.Marshal(v)
+		if err != nil {
+			return fmt.Errorf("can't jsonify test value %v: %s", v, err)
+		}
+		return json.Unmarshal(marshalledMsg, dest)
+	}
+	return nil
+}
+
+func (c *testProtocol) WriteMessage(src interface{}) error {
+	// make sure JSON.Marshal works with src
+	_, err := json.Marshal(src)
+	if err != nil {
+		return err
+	}
+	val := reflect.ValueOf(src)
+	if val.Kind() == reflect.Ptr {
+		src = val.Elem().Interface()
+	}
+	c.down <- src
+	switch v := takeNext(c.up).(type) {
+	case error:
+		return v
+	}
+	return nil
+}
+
+/****************************************************************
+ *
+ *  Go way down to the bottom if you want to see the full, working case.
+ *  This has a rather slow buildup.
+ *
+ *  TODO: check deadlines
+ *
+ ****************************************************************/
+
+func (cs *clientSessionSuite) TestRunFailsIfNilConnection(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	// not connected!
+	err = sess.run()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*disconnected.*")
+}
+
+func (cs *clientSessionSuite) TestRunFailsIfNilProtocolator(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()} // ok, have a constructor
+	sess.Protocolator = nil                       // but no protocol, seeficare.
+	err = sess.run()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*protocol constructor.*")
+}
+
+func (cs *clientSessionSuite) TestRunFailsIfSetDeadlineFails(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname(),
+		DeadlineCondition: condition.Work(false)} // setdeadline will fail
+	err = sess.run()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*deadline.*")
+}
+
+func (cs *clientSessionSuite) TestRunFailsIfWriteFails(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname(),
+		WriteCondition: condition.Work(false)} // write will fail
+	err = sess.run()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*write.*")
+}
+
+func (cs *clientSessionSuite) TestRunConnectMessageFails(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	c.Check(takeNext(downCh), DeepEquals, protocol.ConnectMsg{
+		Type:     "connect",
+		DeviceId: sess.DeviceId,
+		Levels:   map[string]int64{},
+	})
+	upCh <- errors.New("Overflow error in /dev/null")
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, "Overflow.*null")
+}
+
+func (cs *clientSessionSuite) TestRunConnackReadError(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- io.EOF
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*EOF.*")
+}
+
+func (cs *clientSessionSuite) TestRunBadConnack(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{}
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*invalid.*")
+}
+
+func (cs *clientSessionSuite) TestRunMainloopReadError(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
+	}
+	// in the mainloop!
+	upCh <- errors.New("Read")
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Equals, "Read")
+}
+
+func (cs *clientSessionSuite) TestRunPongWriteError(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
+	}
+	// in the mainloop!
+	upCh <- protocol.PingPongMsg{Type: "ping"}
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+	upCh <- errors.New("Pong")
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Equals, "Pong")
+}
+
+func (cs *clientSessionSuite) TestRunPingPong(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
+	}
+	// in the mainloop!
+	upCh <- protocol.PingPongMsg{Type: "ping"}
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+	upCh <- nil    // pong ok
+	upCh <- io.EOF // close it down
+	err = <-errCh
+}
+
+func (cs *clientSessionSuite) TestRunBadAckWrite(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+	sess.MsgCh = make(chan *Notification, 5)
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{time.Second.String()},
+	}
+	// in the mainloop!
+
+	b := &protocol.BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "APP",
+		ChanId:   "0",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
+	}
+	upCh <- b
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "ack"})
+	upCh <- errors.New("ACK ACK ACK")
+	err = <-errCh
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Equals, "ACK ACK ACK")
+}
+
+func (cs *clientSessionSuite) TestRunBroadcastWrongChannel(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	errCh := make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+	sess.MsgCh = make(chan *Notification, 5)
+
+	go func() {
+		errCh <- sess.run()
+	}()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{time.Second.String()},
+	}
+	// in the mainloop!
+
+	b := &protocol.BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "APP",
+		ChanId:   "42",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
+	}
+	upCh <- b
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "ack"})
+	upCh <- nil    // ack ok
+	upCh <- io.EOF // close it down
+	err = <-errCh
+	c.Check(len(sess.MsgCh), Equals, 0)
+}
+
+func (cs *clientSessionSuite) TestRunBroadcastRightChannel(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.Connection = &testConn{Name: testname()}
+	sess.ErrCh = make(chan error, 1)
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+	sess.MsgCh = make(chan *Notification, 5)
+
+	sess.Run()
+
+	takeNext(downCh) // connectMsg
+	upCh <- nil      // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{time.Second.String()},
+	}
+	// in the mainloop!
+
+	b := &protocol.BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "--ignored--",
+		ChanId:   "0",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
+	}
+	upCh <- b
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "ack"})
+	upCh <- nil    // ack ok
+	upCh <- io.EOF // close it down
+	err = <-sess.ErrCh
+	c.Assert(len(sess.MsgCh), Equals, 1)
+	c.Check(<-sess.MsgCh, Equals, &Notification{})
+	// and finally, the session keeps track of the levels
+	c.Check(sess.Levels.GetAll(), DeepEquals, map[string]int64{"0": 2})
+}
+
+/*
+ *
+ *
+ *
+ * breathe in...
+ */
+
+func (cs *clientSessionSuite) TestDialFailsWithNoAddress(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	err = sess.Dial()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*dial.*address.*")
+}
+
+func (cs *clientSessionSuite) TestDialConnects(c *C) {
+	lp, err := net.Listen("tcp", ":0")
+	c.Assert(err, IsNil)
+	defer lp.Close()
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.ServerAddr = lp.Addr().String()
+	err = sess.Dial()
+	c.Check(err, IsNil)
+	c.Check(sess.Connection, NotNil)
+}
+
+func (cs *clientSessionSuite) TestResetFailsWithoutProtocolator(c *C) {
+	sess, _ := NewSession(Config{}, debuglog, "wah")
+	sess.Protocolator = nil
+	err := sess.Reset()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*protocol constructor\\.")
+}
+
+func (cs *clientSessionSuite) TestResetFailsWithNoAddress(c *C) {
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	err = sess.Reset()
+	c.Assert(err, NotNil)
+	c.Check(err.Error(), Matches, ".*dial.*address.*")
+}
+
+func (cs *clientSessionSuite) TestResets(c *C) {
+	upCh := make(chan interface{}, 5)
+	downCh := make(chan interface{}, 5)
+	proto := &testProtocol{up: upCh, down: downCh}
+	lp, err := net.Listen("tcp", ":0")
+	c.Assert(err, IsNil)
+	defer lp.Close()
+
+	sess, err := NewSession(Config{}, debuglog, "wah")
+	c.Assert(err, IsNil)
+	sess.ServerAddr = lp.Addr().String()
+	sess.Connection = &testConn{Name: testname()}
+	sess.Protocolator = func(_ net.Conn) protocol.Protocol { return proto }
+
+	sess.Reset()
+
+	// wheee
+	err = <-sess.ErrCh
+	c.Assert(err, NotNil) // some random tcp error because
+	// there's nobody talking to the port
+}

=== modified file 'protocol/protocol.go'
--- protocol/protocol.go	2014-01-14 15:35:20 +0000
+++ protocol/protocol.go	2014-01-25 02:14:54 +0000
@@ -55,7 +55,7 @@
 }
 
 // NewProtocol0 creates and initialises a protocol with wire format version 0.
-func NewProtocol0(conn net.Conn) *protocol0 {
+func NewProtocol0(conn net.Conn) Protocol {
 	buf := bytes.NewBuffer(make([]byte, 5000))
 	return &protocol0{
 		buffer: buf,