Repository: ably/ably-go
Pull request #152 (2020-04-21 12:07:03)

RTN15a: Reconnect after networking error.
This implements RTN15a in a kind of ad-hoc way, pending the rest
of the recovery logic.


Patch:
diff --git a/ably/options.go b/ably/options.go
index d5726d5..0fc8aa5 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -29,6 +29,7 @@ var defaultOptions = &ClientOptions{
 	RealtimeHost:             "realtime.ably.io",
 	TimeoutDisconnect:        30 * time.Second,
 	RealtimeRequestTimeout:   10 * time.Second, // DF1b
+	DisconnectedRetryTimeout: 15 * time.Second, // TO3l1
 	TimeoutSuspended:         2 * time.Minute,
 	FallbackRetryTimeout:     10 * time.Minute,
 	IdempotentRestPublishing: false,
@@ -212,6 +213,10 @@ type ClientOptions struct {
 	// and each subsequent operation.
 	RealtimeRequestTimeout time.Duration
 
+	// DisconnectedRetryTimeout is the time to wait after a disconnection before
+	// attempting an automatic reconnection, if still disconnected.
+	DisconnectedRetryTimeout time.Duration
+
 	// Dial specifies the dial function for creating message connections used
 	// by RealtimeClient.
 	//
@@ -275,6 +280,13 @@ func (opts *ClientOptions) realtimeRequestTimeout() time.Duration {
 	return defaultOptions.RealtimeRequestTimeout
 }
 
+func (opts *ClientOptions) disconnectedRetryTimeout() time.Duration {
+	if opts.DisconnectedRetryTimeout != 0 {
+		return opts.DisconnectedRetryTimeout
+	}
+	return defaultOptions.DisconnectedRetryTimeout
+}
+
 func (opts *ClientOptions) restURL() string {
 	host := opts.RestHost
 	if host == "" {
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 6980625..2178731 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -82,6 +82,18 @@ var connectResultStates = []StateEnum{
 }
 
 func (c *Conn) connect(result bool) (Result, error) {
+	return c.connectWithRecovery(result, "", 0)
+}
+
+func (c *Conn) reconnect(result bool) (Result, error) {
+	c.state.Lock()
+	connKey := c.details.ConnectionKey
+	connSerial := c.serial
+	c.state.Unlock()
+	return c.connectWithRecovery(result, connKey, connSerial)
+}
+
+func (c *Conn) connectWithRecovery(result bool, connKey string, connSerial int64) (Result, error) {
 	c.state.Lock()
 	defer c.state.Unlock()
 	if c.isActive() {
@@ -118,6 +130,10 @@ func (c *Conn) connect(result bool) (Result, error) {
 	if err := c.auth.authQuery(query); err != nil {
 		return nil, c.setState(StateConnFailed, err)
 	}
+	if connKey != "" {
+		query.Set("resume", connKey)
+		query.Set("connectionSerial", fmt.Sprint(connSerial))
+	}
 	u.RawQuery = query.Encode()
 	conn, err := c.dial(proto, u)
 	if err != nil {
@@ -350,9 +366,11 @@ func (c *Conn) eventloop() {
 				c.state.Unlock()
 				return
 			}
-			c.setState(StateConnFailed, err)
+
+			c.setState(StateConnDisconnected, err)
 			c.state.Unlock()
-			return // TODO recovery
+			c.reconnect(false)
+			return
 		}
 		if msg.ConnectionSerial != 0 {
 			c.state.Lock()


Test patch:
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
new file mode 100644
index 0000000..c476868
--- /dev/null
+++ b/ably/realtime_conn_spec_test.go
@@ -0,0 +1,128 @@
+package ably_test
+
+import (
+	"io"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/ably/ably-go/ably"
+	"github.com/ably/ably-go/ably/ablytest"
+	"github.com/ably/ably-go/ably/internal/ablyutil"
+	"github.com/ably/ably-go/ably/proto"
+)
+
+func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
+		},
+	})
+	defer safeclose(t, client, app)
+
+	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+		t.Fatalf("Connect=%s", err)
+	}
+
+	channel := client.Channels.Get("channel")
+
+	if err := ablytest.Wait(channel.Attach()); err != nil {
+		t.Fatal(err)
+	}
+
+	sub, err := channel.Subscribe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stateChanges := make(chan ably.State, 16)
+	client.Connection.On(stateChanges)
+
+	doEOF <- struct{}{}
+
+	var state ably.State
+
+	select {
+	case state = <-stateChanges:
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("didn't transition on EOF")
+	}
+
+	if expected, got := ably.StateConnDisconnected, state; expected != got.State {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+
+	// Publish a message to the channel through REST. If connection recovery
+	// succeeds, we should then receive it without reattaching.
+
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case state = <-stateChanges:
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("didn't reconnect")
+	}
+
+	if expected, got := ably.StateConnConnecting, state; expected != got.State {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+
+	select {
+	case state = <-stateChanges:
+	case <-time.After(ablytest.Timeout):
+		t.Fatal("didn't transition from CONNECTING")
+	}
+
+	if expected, got := ably.StateConnConnected, state; expected != got.State {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+
+	select {
+	case msg := <-sub.MessageChannel():
+		if expected, got := "data", msg.Data; expected != got {
+			t.Fatalf("expected message with data %v, got %v", expected, got)
+		}
+	case <-time.After(ablytest.Timeout):
+		t.Fatal("expected message after connection recovery; got none")
+	}
+}
+
+type protoConnWithFakeEOF struct {
+	proto.Conn
+	doEOF <-chan struct{}
+}
+
+func (c protoConnWithFakeEOF) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	type result struct {
+		msg *proto.ProtocolMessage
+		err error
+	}
+
+	received := make(chan result, 1)
+
+	go func() {
+		msg, err := c.Conn.Receive(deadline)
+		received <- result{msg: msg, err: err}
+
+	}()
+
+	select {
+	case r := <-received:
+		return r.msg, r.err
+	case <-c.doEOF:
+		return nil, io.EOF
+	}
+}
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index ffa626a..c4eecc2 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -165,7 +165,7 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 	defer safeclose(t, app)
 
 	states := make(chan ably.State, 10)
-	client.Connection.On(states, ably.StateConnConnected, ably.StateConnFailed)
+	client.Connection.On(states, ably.StateConnConnected, ably.StateConnDisconnected)
 
 	client.Connection.Connect()
 
@@ -188,9 +188,7 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 		t.Fatal("didn't receive state change event")
 	}
 
-	// TODO: Should be Disconnected, not Failed
-	// Part of https://ably.atlassian.net/browse/FEA-391
-	if expected, got := ably.StateConnFailed, state.State; expected != got {
+	if expected, got := ably.StateConnDisconnected, state.State; expected != got {
 		t.Fatalf("expected %v, got %v", expected, got)
 	}
 }