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

[John Lenton]

Ladies and gentlemen, the client session.

Refactored the whole dance.

[Samuele Pedroni (pedronis)]

31 + sess.Log.Errorf("%s", err)

seems that error needs a prefix of sort

[Samuele Pedroni (pedronis)]

98 +func (cs *clientSessionSuite) TestResets(c *C) {

it's hard to tell what this is testing concretely? it's not cliear it's checking the various bits of channel resetting for example

maybe run/Run should set a flag on the session or something to use here?

[Samuele Pedroni (pedronis)]

Run and Reset need descriptions

[Samuele Pedroni (pedronis)]

vaguer comments:

bit unsure about the assymmetry that the happy start path has you call Dial and Run,

and Reset does both for you,

also not sure about Run doing "go ..." for you

[Samuele Pedroni (pedronis)]

40 + sess.Close()

that one needs testing, calling Dial twice is a bit annoying though, maybe easier if it lives in run() ?

minor:
8 + sess.ErrCh = make(chan error, 1)
9 + sess.MsgCh = make(chan *Notification)

this could be done in run() as well, run is the only place that uses sess.ErrCh

38 + panic("can't Reset() without a protocol constructor.")

should say Dial

114 + func() error { return failure })

could use a channel there given that loop is run in a goroutine

[John Lenton (chipaca)]

Oops, yes, calling close was untested (no need to call Dial twice). Fixed that.
Also moved it into run (as a parameter) and tested it there as well (what's the convention to avoid shadowing builtins? gone with close_ for now).

Moved channel creation, good point.

Fixed the panic message.

Added a second channel to the loop test function; note "return failure" goes into the error channel, so it was already doing one. Made the second one more explicit though.

[Samuele Pedroni (pedronis)]

> Also moved it into run (as a parameter) and tested it there as well (what's
> the convention to avoid shadowing builtins? gone with close_ for now).

yes the single namespace (for types, modules and variables and builtins) plus the convention for protected/public makes for clunky clashes sometimes, I haven't seen deeper recommendations than to inflict the most clunkiness on variable naming; doClose|closeFunc though perhaps

> Added a second channel to the loop test function; note "return failure" goes
> into the error channel, so it was already doing one. Made the second one more
> explicit though.

I was meaning this, to check that run() is running is a goroutine, given that the MsgCh is also unbuffered it doesn't matter for now (but this is more independent of the details of ErrCh/MsgCh)

--- client/session/session_test.go 2014-02-01 09:42:43 +0000
+++ client/session/session_test.go 2014-02-01 14:00:10 +0000
@@ -584,18 +584,20 @@
  // biggie if this stops being true)
  c.Check(sess.ErrCh, IsNil)
  c.Check(sess.MsgCh, IsNil)
- failure := errors.New("TestRunRunsEvenIfLoopFails")
+ failureCh := make(chan error)
  notf := &Notification{}
  err = sess.run(
   func() {},
   func() error { return nil },
   func() error { return nil },
- func() error { sess.MsgCh <- notf; return failure })
+ func() error { sess.MsgCh <- notf; return <-failureCh })
  c.Check(err, Equals, nil)
  // if run doesn't error it sets up the channels
  c.Assert(sess.ErrCh, NotNil)
  c.Assert(sess.MsgCh, NotNil)
  c.Check(<-sess.MsgCh, Equals, notf)
+ failure := errors.New("TestRunRunsEvenIfLoopFails")
+ failureCh <- failure
  c.Check(<-sess.ErrCh, Equals, failure)
 }

[Samuele Pedroni (pedronis)]

there's a data race in TestDialsWorks:

https://pastebin.canonical.com/104078/

I fear the stuff in condition may need Mutexes or be channel based more

[Samuele Pedroni (pedronis)]

74 + ch := make(chan bool, 1)
75 + err = sess.run(
76 + func() { ch <- true },

close isn't called in a goroutine so a flag would work there as well

[John Lenton (chipaca)]

> > Also moved it into run (as a parameter) and tested it there as well (what's
> > the convention to avoid shadowing builtins? gone with close_ for now).
>
> yes the single namespace (for types, modules and variables and builtins) plus
> the convention for protected/public makes for clunky clashes sometimes, I
> haven't seen deeper recommendations than to inflict the most clunkiness on
> variable naming; doClose|closeFunc though perhaps

went with using the noun forms.

> I was meaning this, to check that run() is running is a goroutine, given that
> the MsgCh is also unbuffered it doesn't matter for now (but this is more
> independent of the details of ErrCh/MsgCh)

now I got it. And stole your patch :)

[John Lenton (chipaca)]

> there's a data race in TestDialsWorks:
>
> https://pastebin.canonical.com/104078/
>
> I fear the stuff in condition may need Mutexes or be channel based more

Yup. Done.

[John Lenton (chipaca)]

> 74 + ch := make(chan bool, 1)
> 75 + err = sess.run(
> 76 + func() { ch <- true },
>
> close isn't called in a goroutine so a flag would work there as well

and the code is more legible as a result. Done.

[Samuele Pedroni (pedronis)]

thanks for bearing with me

looks good !

review: Approve

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

42. By John Lenton on 2014-02-01

[r=pedronis] Ladies and gentlemen, the client session.

=== modified file 'client/session/session.go'
--- client/session/session.go	2014-01-31 21:47:15 +0000
+++ client/session/session.go	2014-02-01 21:04:46 +0000
@@ -196,3 +196,31 @@
 	sess.Log.Debugf("Connected %v.", conn.LocalAddr())
 	return nil
 }
+
+// run calls connect, and if it works it calls start, and if it works
+// it runs loop in a goroutine, and ships its return value over ErrCh.
+func (sess *ClientSession) run(closer func(), connecter, starter, looper func() error) error {
+	closer()
+	err := connecter()
+	if err == nil {
+		err = starter()
+		if err == nil {
+			sess.ErrCh = make(chan error, 1)
+			sess.MsgCh = make(chan *Notification)
+			go func() { sess.ErrCh <- looper() }()
+		}
+	}
+	return err
+}
+
+// Dial takes the session from newly created (or newly disconnected)
+// to running the main loop.
+func (sess *ClientSession) Dial() error {
+	if sess.Protocolator == nil {
+		// a missing protocolator means you've willfully overridden
+		// it; returning an error here would prompt AutoRedial to just
+		// keep on trying.
+		panic("can't Dial() without a protocol constructor.")
+	}
+	return sess.run(sess.Close, sess.connect, sess.start, sess.loop)
+}

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-01-31 21:47:15 +0000
+++ client/session/session_test.go	2014-02-01 21:04:46 +0000
@@ -17,6 +17,7 @@
 package session
 
 import (
+	"crypto/tls"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -39,7 +40,8 @@
 type clientSessionSuite struct{}
 
 var nullog = logger.NewSimpleLogger(ioutil.Discard, "error")
-var debuglog = logger.NewSimpleLogger(os.Stderr, "debug")
+var noisylog = logger.NewSimpleLogger(os.Stderr, "debug")
+var debuglog = nullog
 var _ = Suite(&clientSessionSuite{})
 
 //
@@ -542,5 +544,159 @@
 	}
 	// start is now done.
 	err = <-errCh
-	c.Assert(err, IsNil)
+	c.Check(err, IsNil)
+}
+
+/****************************************************************
+  run() tests
+****************************************************************/
+
+func (cs *clientSessionSuite) TestRunBailsIfConnectFails(c *C) {
+	sess, err := NewSession("", nil, 0, "wah", debuglog)
+	c.Assert(err, IsNil)
+	failure := errors.New("TestRunBailsIfConnectFails")
+	has_closed := false
+	err = sess.run(
+		func() { has_closed = true },
+		func() error { return failure },
+		nil,
+		nil)
+	c.Check(err, Equals, failure)
+	c.Check(has_closed, Equals, true)
+}
+
+func (cs *clientSessionSuite) TestRunBailsIfStartFails(c *C) {
+	sess, err := NewSession("", nil, 0, "wah", debuglog)
+	c.Assert(err, IsNil)
+	failure := errors.New("TestRunBailsIfStartFails")
+	err = sess.run(
+		func() {},
+		func() error { return nil },
+		func() error { return failure },
+		nil)
+	c.Check(err, Equals, failure)
+}
+
+func (cs *clientSessionSuite) TestRunRunsEvenIfLoopFails(c *C) {
+	sess, err := NewSession("", nil, 0, "wah", debuglog)
+	c.Assert(err, IsNil)
+	// just to make a point: until here we haven't set ErrCh & MsgCh (no
+	// biggie if this stops being true)
+	c.Check(sess.ErrCh, IsNil)
+	c.Check(sess.MsgCh, IsNil)
+	failureCh := make(chan error) // must be unbuffered
+	notf := &Notification{}
+	err = sess.run(
+		func() {},
+		func() error { return nil },
+		func() error { return nil },
+		func() error { sess.MsgCh <- notf; return <-failureCh })
+	c.Check(err, Equals, nil)
+	// if run doesn't error it sets up the channels
+	c.Assert(sess.ErrCh, NotNil)
+	c.Assert(sess.MsgCh, NotNil)
+	c.Check(<-sess.MsgCh, Equals, notf)
+	failure := errors.New("TestRunRunsEvenIfLoopFails")
+	failureCh <- failure
+	c.Check(<-sess.ErrCh, Equals, failure)
+	// so now you know it was running in a goroutine :)
+}
+
+/****************************************************************
+  Dial() tests
+****************************************************************/
+
+func (cs *clientSessionSuite) TestDialPanics(c *C) {
+	// one last unhappy test
+	sess, err := NewSession("", nil, 0, "wah", debuglog)
+	c.Assert(err, IsNil)
+	sess.Protocolator = nil
+	c.Check(sess.Dial, PanicMatches, ".*protocol constructor.")
+}
+
+func (cs *clientSessionSuite) TestDialWorks(c *C) {
+	// happy path thoughts
+	cert, err := tls.X509KeyPair(helpers.TestCertPEMBlock, helpers.TestKeyPEMBlock)
+	c.Assert(err, IsNil)
+	tlsCfg := &tls.Config{
+		Certificates:           []tls.Certificate{cert},
+		SessionTicketsDisabled: true,
+	}
+
+	timeout := 100 * time.Millisecond
+	lst, err := tls.Listen("tcp", "localhost:0", tlsCfg)
+	c.Assert(err, IsNil)
+	sess, err := NewSession(lst.Addr().String(), nil, timeout, "wah", debuglog)
+	c.Assert(err, IsNil)
+	tconn := &testConn{CloseCondition: condition.Fail2Work(10)}
+	sess.Connection = tconn
+	// just to be sure:
+	c.Check(tconn.CloseCondition.String(), Matches, ".* 10 to go.")
+
+	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 sess.Dial()
+
+	srv, err := lst.Accept()
+	c.Assert(err, IsNil)
+
+	// connect done
+
+	// Dial should have had the session's old connection (tconn) closed
+	// before connecting a new one; if that was done, tconn's condition
+	// ticked forward:
+	c.Check(tconn.CloseCondition.String(), Matches, ".* 9 to go.")
+
+	// now, start: 1. protocol version
+	v, err := protocol.ReadWireFormatVersion(srv, timeout)
+	c.Assert(err, IsNil)
+	c.Assert(v, Equals, protocol.ProtocolWireVersion)
+
+	// 2. "connect" (but on the fake protcol above! woo)
+
+	c.Check(takeNext(downCh), Equals, "deadline 100ms")
+	_, ok := takeNext(downCh).(protocol.ConnectMsg)
+	c.Check(ok, Equals, true)
+	upCh <- nil // no error
+	upCh <- protocol.ConnAckMsg{
+		Type:   "connack",
+		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
+	}
+	// start is now done.
+
+	// 3. "loop"
+
+	// ping works,
+	c.Check(takeNext(downCh), Equals, "deadline 110ms")
+	upCh <- protocol.PingPongMsg{Type: "ping"}
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+	upCh <- nil
+
+	// and broadcasts...
+	b := &protocol.BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "--ignored--",
+		ChanId:   "0",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{"b":1}`)},
+	}
+	c.Check(takeNext(downCh), Equals, "deadline 110ms")
+	upCh <- b
+	c.Check(takeNext(downCh), Equals, protocol.AckMsg{"ack"})
+	upCh <- nil
+	// ...get bubbled up,
+	c.Check(<-sess.MsgCh, NotNil)
+	// and their TopLevel remembered
+	c.Check(sess.Levels.GetAll(), DeepEquals, map[string]int64{"0": 2})
+
+	// and ping still work even after that.
+	c.Check(takeNext(downCh), Equals, "deadline 110ms")
+	upCh <- protocol.PingPongMsg{Type: "ping"}
+	c.Check(takeNext(downCh), Equals, protocol.PingPongMsg{Type: "pong"})
+	failure := errors.New("pongs")
+	upCh <- failure
+	c.Check(<-sess.ErrCh, Equals, failure)
 }

=== modified file 'testing/condition/condition.go'
--- testing/condition/condition.go	2014-01-20 06:23:17 +0000
+++ testing/condition/condition.go	2014-02-01 21:04:46 +0000
@@ -20,6 +20,7 @@
 import (
 	"fmt"
 	"strings"
+	"sync"
 )
 
 type Interface interface {
@@ -62,9 +63,12 @@
 
 type fail2Work struct {
 	Left int32
+	lock sync.RWMutex
 }
 
 func (c *fail2Work) OK() bool {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.Left > 0 {
 		c.Left--
 		return false
@@ -74,6 +78,8 @@
 }
 
 func (c *fail2Work) String() string {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 	if c.Left > 0 {
 		return fmt.Sprintf("Still Broken, %d to go.", c.Left)
 	} else {
@@ -104,10 +110,13 @@
 
 type chain struct {
 	subs []*_iter
+	lock sync.RWMutex
 }
 
 func (c *chain) OK() bool {
 	var sub *_iter
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	for _, sub = range c.subs {
 		if sub.remaining > 0 {
 			sub.remaining--
@@ -119,6 +128,8 @@
 
 func (c *chain) String() string {
 	ss := make([]string, len(c.subs))
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 	for i, sub := range c.subs {
 		ss[i] = sub.String()
 	}
@@ -138,5 +149,5 @@
 		args = args[2:]
 	}
 
-	return &chain{iters}
+	return &chain{subs: iters}
 }