Merge lp:~chipaca/ubuntu-push/advanced-client-session into lp:ubuntu-push

[John Lenton]

The handling of the ping.

Client session, volume 2: Advanced client session.

[lp:~chipaca/ubuntu-push/advanced-client-session]

36. By John Lenton on 2014-01-31

Merged client-session into advanced-client-session.

37. By John Lenton on 2014-01-31

Merged client-session into advanced-client-session.

[Samuele Pedroni (pedronis)]

protocol itself has a SetDeadline that should be used after you got a protocol and not a plain connection

review: Needs Fixing

[lp:~chipaca/ubuntu-push/advanced-client-session]

38. By John Lenton on 2014-01-31

seriously bad protocol breakage spotted by pedronis; fixed

39. By John Lenton on 2014-01-31

Merged watchticker-goes-to-the-races into advanced-client-session.

40. By John Lenton on 2014-01-31

Assert instead of Check when checking the length of a chan you are about to read from

[Samuele Pedroni (pedronis)]

review: Approve

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-01-31 12:44:29 +0000
+++ client/session/session.go	2014-01-31 14:15:56 +0000
@@ -107,3 +107,11 @@
 	}
 	return nil
 }
+
+// handle "ping" messages
+func (sess *ClientSession) handlePing() error {
+	sess.proto.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+	err := sess.proto.WriteMessage(protocol.PingPongMsg{Type: "pong"})
+	sess.Log.Debugf("Ping.")
+	return err
+}

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-01-31 12:44:29 +0000
+++ client/session/session_test.go	2014-01-31 14:15:56 +0000
@@ -17,14 +17,18 @@
 package session
 
 import (
+	"encoding/json"
 	"errors"
+	"fmt"
 	"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"
 	"testing"
 	"time"
 )
@@ -75,6 +79,52 @@
 func (tc *testConn) Read(buf []byte) (n int, err error) { panic("Read not implemented.") }
 func (tc *testConn) Write(buf []byte) (int, error)      { panic("Write not implemented.") }
 
+// 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 {
+	panic("ReadMessage not implemented.")
+}
+
+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
+}
+
 /****************************************************************
   NewSession() tests
 ****************************************************************/
@@ -197,3 +247,46 @@
 	sess.Connection = &testConn{}
 	c.Check(sess.checkRunnable(), IsNil)
 }
+
+/****************************************************************
+  handlePing() tests
+****************************************************************/
+
+type msgSuite struct {
+	sess   *ClientSession
+	upCh   chan interface{}
+	downCh chan interface{}
+	errCh  chan error
+}
+
+var _ = Suite(&msgSuite{})
+
+func (s *msgSuite) SetUpTest(c *C) {
+	var err error
+	s.sess, err = NewSession("", nil, 0, "wah", debuglog)
+	c.Assert(err, IsNil)
+	s.sess.Connection = &testConn{Name: "TestRun* (small r)"}
+	s.errCh = make(chan error, 1)
+	s.upCh = make(chan interface{}, 5)
+	s.downCh = make(chan interface{}, 5)
+	s.sess.proto = &testProtocol{up: s.upCh, down: s.downCh}
+}
+
+func (s *msgSuite) TestHandlePingWorks(c *C) {
+	s.upCh <- nil // no error
+	s.sess.ExchangeTimeout = time.Millisecond
+	c.Check(s.sess.handlePing(), IsNil)
+	c.Assert(len(s.downCh), Equals, 2)
+	c.Check(<-s.downCh, Equals, "deadline 1ms")
+	c.Check(<-s.downCh, Equals, protocol.PingPongMsg{Type: "pong"})
+}
+
+func (s *msgSuite) TestHandlePingHandlesPongWriteError(c *C) {
+	failure := errors.New("Pong")
+	s.upCh <- failure
+
+	c.Check(s.sess.handlePing(), Equals, failure)
+	c.Assert(len(s.downCh), Equals, 2)
+	c.Check(<-s.downCh, Equals, "deadline 0")
+	c.Check(<-s.downCh, Equals, protocol.PingPongMsg{Type: "pong"})
+}