Repository: ably/ably-go
Pull request #165 (2020-06-22 11:24:34)

 Add rtn16 
- [x] RTN16a
- [x] RTN16b
- [x] RTN16c
- [x] RTN16d
- [x] RTN16e
- [x] RTN16f



Patch:
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 0cc9035..6ffc0ff 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/ably/ably-go/ably/internal/ablyutil"
@@ -15,11 +16,23 @@ var (
 	errQueueing = errors.New("unable to send messages in current state with disabled queueing")
 )
 
+// connectionMode is the mode in which the connection is operating
+type connectionMode uint
+
+const (
+	// normalMode this is set when the Connection operating normally
+	normalMode connectionMode = iota
+	// resumeMode this is set when the Connection is trying to resume
+	resumeMode
+	// recoveryMode this is set when the Connection is trying to recover
+	recoveryMode
+)
+
 // Connection represents a single connection Realtime instantiates for
 // communication with Ably servers.
 type Connection struct {
-	details   proto.ConnectionDetails
 	id        string
+	key       string
 	serial    int64
 	msgSerial int64
 	err       error
@@ -35,6 +48,7 @@ type Connection struct {
 	callbacks    connCallbacks
 	reconnecting bool
 }
+
 type connCallbacks struct {
 	onChannelMsg func(*proto.ProtocolMessage)
 	// onReconnectMsg is called when we get a response from reconnect request. We
@@ -95,15 +109,17 @@ var connectResultStates = []StateEnum{
 }
 
 func (c *Connection) connect(result bool) (Result, error) {
-	return c.connectWithRecovery(result, "", 0)
+	c.state.Lock()
+	mode := c.getMode()
+	c.state.Unlock()
+	return c.connectWith(result, mode)
 }
 
 func (c *Connection) reconnect(result bool) (Result, error) {
 	c.state.Lock()
-	connKey := c.details.ConnectionKey
-	connSerial := c.serial
+	mode := c.getMode()
 	c.state.Unlock()
-	r, err := c.connectWithRecovery(result, connKey, connSerial)
+	r, err := c.connectWith(result, mode)
 	if err != nil {
 		return nil, err
 	}
@@ -116,22 +132,17 @@ func (c *Connection) reconnect(result bool) (Result, error) {
 	return r, nil
 }
 
-func (c *Connection) connectWithRecovery(result bool, connKey string, connSerial int64) (Result, error) {
-	c.state.Lock()
-	defer c.state.Unlock()
-	if c.isActive() {
-		return nopResult, nil
+func (c *Connection) getMode() connectionMode {
+	if c.key != "" {
+		return resumeMode
 	}
-	c.setState(StateConnConnecting, nil)
-	u, err := url.Parse(c.opts.realtimeURL())
-	if err != nil {
-		return nil, c.setState(StateConnFailed, err)
+	if c.opts.Recover != "" {
+		return recoveryMode
 	}
-	var res Result
-	if result {
-		res = c.state.listenResult(connectResultStates...)
-	}
-	proto := c.opts.protocol()
+	return normalMode
+}
+
+func (c *Connection) params(mode connectionMode) (url.Values, error) {
 	query := url.Values{
 		"timestamp": []string{strconv.FormatInt(TimeNow(), 10)},
 		"echo":      []string{"true"},
@@ -151,13 +162,44 @@ func (c *Connection) connectWithRecovery(result bool, connKey string, connSerial
 		query.Set(k, v)
 	}
 	if err := c.auth.authQuery(query); err != nil {
+		return nil, err
+	}
+	switch mode {
+	case resumeMode:
+		query.Set("resume", c.key)
+		query.Set("connectionSerial", fmt.Sprint(c.serial))
+	case recoveryMode:
+		m := strings.Split(c.opts.Recover, ":")
+		if len(m) != 3 {
+			return nil, errors.New("conn: Invalid recovery key")
+		}
+		query.Set("recover", m[0])
+		query.Set("connectionSerial", m[1])
+	}
+	return query, nil
+}
+
+func (c *Connection) connectWith(result bool, mode connectionMode) (Result, error) {
+	c.state.Lock()
+	defer c.state.Unlock()
+	if c.isActive() {
+		return nopResult, nil
+	}
+	c.setState(StateConnConnecting, nil)
+	u, err := url.Parse(c.opts.realtimeURL())
+	if err != nil {
 		return nil, c.setState(StateConnFailed, err)
 	}
-	if connKey != "" {
-		query.Set("resume", connKey)
-		query.Set("connectionSerial", fmt.Sprint(connSerial))
+	var res Result
+	if result {
+		res = c.state.listenResult(connectResultStates...)
+	}
+	query, err := c.params(mode)
+	if err != nil {
+		return nil, c.setState(StateConnFailed, err)
 	}
 	u.RawQuery = query.Encode()
+	proto := c.opts.protocol()
 	conn, err := c.dial(proto, u)
 	if err != nil {
 		return nil, c.setState(StateConnFailed, err)
@@ -167,6 +209,7 @@ func (c *Connection) connectWithRecovery(result bool, connKey string, connSerial
 	} else {
 		c.setConn(conn)
 	}
+	c.reconnecting = mode == recoveryMode || mode == resumeMode
 	return res, nil
 }
 
@@ -178,6 +221,7 @@ var closeResultStates = []StateEnum{
 
 func (c *Connection) close() {
 	c.state.Lock()
+	c.key, c.id = "", "" //(RTN16c)
 	defer c.state.Unlock()
 	switch c.state.current {
 	case
@@ -218,7 +262,7 @@ func (c *Connection) ID() string {
 func (c *Connection) Key() string {
 	c.state.Lock()
 	defer c.state.Unlock()
-	return c.details.ConnectionKey
+	return c.key
 }
 
 // Ping issues a ping request against configured endpoint and returns TTR times
@@ -238,6 +282,15 @@ func (c *Connection) ErrorReason() *ErrorInfo {
 	return c.state.err
 }
 
+func (c *Connection) RecoveryKey() string {
+	c.state.Lock()
+	defer c.state.Unlock()
+	if c.key == "" {
+		return ""
+	}
+	return strings.Join([]string{c.key, fmt.Sprint(c.serial), fmt.Sprint(c.msgSerial)}, ":")
+}
+
 // Serial gives serial number of a message received most recently. Last known
 // serial number is used when recovering connection state.
 func (c *Connection) Serial() int64 {
@@ -417,6 +470,10 @@ func (c *Connection) logger() *LoggerOptions {
 	return c.auth.logger()
 }
 
+func (c *Connection) setSerial(serial int64) {
+	c.serial = serial
+}
+
 func (c *Connection) eventloop() {
 	var receiveTimeout time.Duration
 
@@ -440,7 +497,7 @@ func (c *Connection) eventloop() {
 		}
 		if msg.ConnectionSerial != 0 {
 			c.state.Lock()
-			c.serial = msg.ConnectionSerial
+			c.setSerial(msg.ConnectionSerial)
 			c.state.Unlock()
 		}
 		switch msg.Action {
@@ -448,7 +505,7 @@ func (c *Connection) eventloop() {
 		case proto.ActionAck:
 			c.state.Lock()
 			c.pending.Ack(msg.MsgSerial, msg.Count, newErrorProto(msg.Error))
-			c.serial++
+			c.setSerial(c.serial + 1)
 			c.state.Unlock()
 		case proto.ActionNack:
 			c.state.Lock()
@@ -478,13 +535,18 @@ func (c *Connection) eventloop() {
 			}
 		case proto.ActionConnected:
 			c.auth.updateClientID(msg.ConnectionDetails.ClientID)
+			c.state.Lock()
+			// we need to get this before we set c.key so as to be sure if we were
+			// resuming or recovering the connection.
+			mode := c.getMode()
+			c.state.Unlock()
 			if msg.ConnectionDetails != nil {
 				c.state.Lock()
-				c.details = *msg.ConnectionDetails
+				c.key = msg.ConnectionDetails.ConnectionKey //(RTN15e) (RTN16d)
 				c.state.Unlock()
 
 				// Spec RSA7b3, RSA7b4, RSA12a
-				c.auth.updateClientID(c.details.ClientID)
+				c.auth.updateClientID(msg.ConnectionDetails.ClientID)
 
 				maxIdleInterval := time.Duration(msg.ConnectionDetails.MaxIdleInterval) * time.Millisecond
 				receiveTimeout = c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
@@ -492,14 +554,15 @@ func (c *Connection) eventloop() {
 			c.state.Lock()
 			reconnecting := c.reconnecting
 			if reconnecting {
+				// reset the mode
 				c.reconnecting = false
 			}
+			id := c.id
 			c.state.Unlock()
 			if reconnecting {
 				// (RTN15c1) (RTN15c2)
 				c.state.Lock()
 				c.setState(StateConnConnected, newErrorProto(msg.Error))
-				id := c.id
 				c.state.Unlock()
 				if id != msg.ConnectionID {
 					// (RTN15c3)
@@ -516,8 +579,16 @@ func (c *Connection) eventloop() {
 			}
 			c.state.Lock()
 			c.id = msg.ConnectionID
-			c.serial = -1
 			c.msgSerial = 0
+			if reconnecting && mode == recoveryMode {
+				// we are setting msgSerial as per (RTN16f)
+				msgSerial, err := strconv.ParseInt(strings.Split(c.opts.Recover, ":")[2], 10, 64)
+				if err != nil {
+					//TODO: how to handle this? Panic?
+				}
+				c.msgSerial = msgSerial
+			}
+			c.setSerial(-1)
 			c.state.Unlock()
 			c.queue.Flush()
 		case proto.ActionDisconnected:
@@ -527,7 +598,7 @@ func (c *Connection) eventloop() {
 			c.state.Unlock()
 		case proto.ActionClosed:
 			c.state.Lock()
-			c.id = ""
+			c.id, c.key = "", "" //(RTN16c)
 			c.setState(StateConnClosed, nil)
 			c.state.Unlock()
 			if c.conn != nil {


Test patch:
diff --git a/ably/export_test.go b/ably/export_test.go
index d3833c0..fdb88f9 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -106,3 +106,15 @@ func (c *Connection) OnState(ch chan<- State, states ...StateEnum) {
 func (c *Connection) OffState(ch chan<- State, states ...StateEnum) {
 	c.offState(ch, states...)
 }
+
+func (c *Connection) RemoveKey() {
+	c.state.Lock()
+	defer c.state.Unlock()
+	c.key = ""
+}
+
+func (c *Connection) MsgSerial() int64 {
+	c.state.Lock()
+	defer c.state.Unlock()
+	return c.msgSerial
+}
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index a6d5f74..538b20b 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"strings"
 	"testing"
 	"time"
 
@@ -957,3 +958,123 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 }
+
+func TestRealtimeConn_RTN16(t *testing.T) {
+	t.Parallel()
+	app, c := ablytest.NewRealtime(&ably.ClientOptions{})
+	defer safeclose(t, ablytest.FullRealtimeCloser(c), app)
+
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+	channel := c.Channels.Get("channel")
+	if err := ablytest.Wait(channel.Attach()); err != nil {
+		t.Fatal(err)
+	}
+	if err := ablytest.Wait(channel.Publish("name", "data")); err != nil {
+		t.Fatal(err)
+	}
+
+	client := app.NewRealtime(&ably.ClientOptions{
+		Recover: c.Connection.RecoveryKey(),
+	})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client))
+
+	err = ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+	{ //RTN16b, RTN16f
+		if !sameConnection(client.Connection.Key(), c.Connection.Key()) {
+			t.Errorf("expected the same connection")
+		}
+
+		if client.Connection.MsgSerial() != c.Connection.MsgSerial() {
+			t.Errorf("expected %d got %d", c.Connection.MsgSerial(), client.Connection.MsgSerial())
+		}
+		if true {
+			return
+		}
+	}
+
+	{ //(RTN16c)
+		client.Close()
+		if key := client.Connection.Key(); key != "" {
+			t.Errorf("expected key to be empty got %q instead", key)
+		}
+
+		if recover := client.Connection.RecoveryKey(); recover != "" {
+			t.Errorf("expected recovery key to be empty got %q instead", recover)
+		}
+		if id := client.Connection.ID(); id != "" {
+			t.Errorf("expected id to be empty got %q instead", id)
+		}
+	}
+	{ //(RTN16e)
+		// This test was adopted from the ably-js project
+		// https://github.com/ably/ably-js/blob/340e5ce31dc9d7434a06ae4e1eec32bdacc9c6c5/spec/realtime/connection.test.js#L119
+		var query url.Values
+		fakeRecoveryKey := "_____!ablygo_test_fake-key____:5:3"
+		client2 := app.NewRealtime(&ably.ClientOptions{
+			Recover: fakeRecoveryKey,
+			Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+				query = u.Query()
+				return ablyutil.DialWebsocket(protocol, u)
+			},
+		})
+		defer safeclose(t, ablytest.FullRealtimeCloser(client2))
+		err = ablytest.ConnWaiter(client2, client2.Connect, ably.ConnectionEventConnected).Wait()
+		if err == nil {
+			t.Fatal("expected reason to be set")
+		}
+		{ // (RTN16a)
+			recover := query.Get("recover")
+			if recover == "" {
+				t.Fatal("expected resume query param to be set")
+			}
+			parts := strings.Split(fakeRecoveryKey, ":")
+			if recover != parts[0] {
+				t.Errorf("resume: expected %q got %q", parts[0], recover)
+			}
+			serial := query.Get("connectionSerial")
+			if serial == "" {
+				t.Fatal("expected connectionSerial query param to be set")
+			}
+			if serial != parts[1] {
+				t.Errorf("connectionSerial: expected %q got %q", parts[1], serial)
+			}
+		}
+		{ //(RTN16e)
+			info := err.(*ably.ErrorInfo)
+			code := 80008
+			if info.Code != code {
+				// verify unrecoverable-connection error set in stateChange.reason
+				t.Errorf("expected 80000 got %d", info.Code)
+			}
+			reason := client2.Connection.ErrorReason()
+			if reason.Code != code {
+				// verify unrecoverable-connection error set in connection.errorReason
+				t.Errorf("expected 80000 got %d", reason.Code)
+			}
+			if serial := client2.Connection.Serial(); serial != -1 {
+				// verify serial is -1 (new connection), not 5
+				t.Errorf("expected -1 got %d", serial)
+			}
+			if serial := client2.Connection.MsgSerial(); serial != 0 {
+				// (RTN16f)
+				// verify msgSerial is 0 (new connection), not 3
+				t.Errorf("expected 0 got %d", serial)
+			}
+			fake := "ablygo_test_fake"
+			if key := client2.Connection.Key(); strings.Contains(key, fake) {
+				// verify connection using a new connectionkey
+				t.Errorf("expected %q not to contain %q", key, fake)
+			}
+		}
+	}
+}
+
+func sameConnection(a, b string) bool {
+	return strings.Split(a, "-")[0] == strings.Split(b, "-")[0]
+}