Repository: ably/ably-go (go)

Issue:
Fix RTN14c test to include full connection establishment
We were wrongly using `realtimeRequestTimeout` as a timeout for establishing a transport; it should be the timeout for establishing a realtime connection, which includes receiving the CONNECTED protocol message. For the transport, we now use `httpOpenTimeout` (see https://github.com/ably/ably-go/pull/312).

RTN14c needs to be adapted to this change.

Pull request description:
Rewrite RTN14c to test full connection establishment
Also, mock time to make it deterministic.

Fixes #315.

Patch:
diff --git a/ably/internal/ablyutil/websocket.go b/ably/internal/ablyutil/websocket.go
index d1f7cda..61aa913 100644
--- a/ably/internal/ablyutil/websocket.go
+++ b/ably/internal/ablyutil/websocket.go
@@ -49,7 +49,6 @@ func DialWebsocket(proto string, u *url.URL, timeout time.Duration) (*WebsocketC
 	default:
 		return nil, errors.New(`invalid protocol "` + proto + `"`)
 	}
-	// Starts a raw websocket connection with server
 	conn, err := dialWebsocketTimeout(u.String(), "", "https://"+u.Host, timeout)
 	if err != nil {
 		return nil, err
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index f41b418..0b5ed79 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -129,6 +129,30 @@ func (c *Connection) dial(proto string, u *url.URL) (conn proto.Conn, err error)
 	return conn, err
 }
 
+func (c *Connection) connectAfterSuspension(arg connArgs) (result, error) {
+	retryIn := c.opts.suspendedRetryTimeout()
+	lg := c.logger().sugar()
+	lg.Debugf("Attemting to periodically establish connection after suspension with timeout %v", retryIn)
+	ctx, cancel := c.ctxCancelOnStateTransition()
+	defer cancel()
+	tick := ablyutil.NewTicker(c.opts.After)(ctx, retryIn)
+	for {
+		_, ok := <-tick
+		if !ok {
+			return nil, ctx.Err()
+		}
+		res, err := c.connectWith(arg)
+		if err != nil {
+			if recoverable(err) {
+				c.setState(ConnectionStateSuspended, err, retryIn)
+				continue
+			}
+			return nil, c.setState(ConnectionStateFailed, err, 0)
+		}
+		return res, nil
+	}
+}
+
 // recoverable returns true if err is recoverable, err is from making a
 // connection
 func recoverable(err error) bool {
@@ -150,15 +174,8 @@ func (c *Connection) Connect() {
 		return
 	}
 
-	IsInReconnectionLoop := c.state == ConnectionStateDisconnected || c.state == ConnectionStateSuspended
-
-	// set state to connecting for initial connect
 	c.lockSetState(ConnectionStateConnecting, nil, 0)
 
-	if IsInReconnectionLoop {
-		return
-	}
-
 	go func() {
 		c.connect(connArgs{})
 	}()
@@ -270,8 +287,6 @@ func (c *Connection) params(mode connectionMode) (url.Values, error) {
 
 const connectionStateTTLErrFmt = "Exceeded connectionStateTtl=%v while in DISCONNECTED state"
 
-var errClosedWhileReconnecting = errors.New("connection explicitly closed while trying to reconnect")
-
 func (c *Connection) connectWithRetryLoop(arg connArgs) (result, error) {
 	lg := c.logger().sugar()
 	res, err := c.connectWith(arg)
@@ -285,71 +300,81 @@ func (c *Connection) connectWithRetryLoop(arg connArgs) (result, error) {
 	lg.Errorf("Received recoverable error %v", err)
 	retryIn := c.opts.disconnectedRetryTimeout()
 	c.setState(ConnectionStateDisconnected, err, retryIn)
-	idleState := ConnectionStateDisconnected
 
-	// If we spend more than the connection state TTL retrying, we move from
-	// DISCONNECTED to SUSPENDED, which also changes the retry timeout period.
-	stateTTLCtx, cancelStateTTLTimer := context.WithCancel(context.Background())
-	defer cancelStateTTLTimer()
-	stateTTLTimer := c.opts.After(stateTTLCtx, c.opts.connectionStateTTL())
+	ctx, cancel := c.ctxCancelOnStateTransition()
+	defer cancel()
 
-	for {
-		// If the connection transitions, it's because Connect or Close was called
-		// explicitly. In that case, skip the wait and either retry connecting
-		// immediately (RTN11c) or exit the loop (RTN12d).
-		timerCtx, cancelTimer := c.ctxCancelOnStateTransition()
-		<-c.opts.After(timerCtx, retryIn)
-		cancelTimer()
-
-		switch state := c.State(); state {
-		case ConnectionStateConnecting, ConnectionStateDisconnected, ConnectionStateSuspended:
-		case ConnectionStateClosed:
-			// Close was explicitly called, so stop trying to connect (RTN12d).
-			return nil, errClosedWhileReconnecting
-		default:
-			panic(fmt.Errorf("unexpected state transition: %v -> %v", idleState, state))
-		}
+	// The initial DISCONNECTED event has been fired. If we reach stateTTL without
+	// any state changes, we transition to SUSPENDED state
+	stateTTL := c.opts.connectionStateTTL()
+	stateDeadlineCtx, cancelStateDeadline := context.WithCancel(ctx)
+	defer cancelStateDeadline()
+	stateDeadline := c.opts.After(stateDeadlineCtx, stateTTL)
+
+	nextCtx, cancelNext := context.WithCancel(ctx)
+	next := c.opts.After(nextCtx, retryIn)
+	reset := func() {
+		lg.Debugf("Retry in %v", retryIn)
+		cancelNext()
+		nextCtx, cancelNext = context.WithCancel(ctx)
+		next = c.opts.After(nextCtx, retryIn)
+	}
 
-		// Before attempting to connect, move from DISCONNETCED to SUSPENDED if
-		// more than connectionStateTTL has passed.
-		if idleState == ConnectionStateDisconnected {
+loop:
+	for {
+		select {
+		case _, ok := <-stateDeadline:
+			if !ok {
+				return nil, ctx.Err()
+			}
+			break loop
+		case _, ok := <-next:
+			if !ok {
+				return nil, ctx.Err()
+			}
+			// Prioritize stateDeadline.
 			select {
-			case <-stateTTLTimer:
-				// (RTN14e)
-				err = fmt.Errorf(connectionStateTTLErrFmt, c.opts.connectionStateTTL())
-				c.setState(ConnectionStateSuspended, err, c.opts.suspendedRetryTimeout())
-				idleState = ConnectionStateSuspended
-				// (RTN14f)
-				lg.Debug("Reached SUSPENDED state while opening connection")
-				retryIn = c.opts.suspendedRetryTimeout()
-				continue // wait for re-connection with new retry timeout for suspended
+			case _, ok := <-stateDeadline:
+				if !ok {
+					return nil, ctx.Err()
+				}
 			default:
 			}
-		}
 
-		lg.Debug("Attemting to open connection")
-		res, err := c.connectWith(arg)
-		if err == nil {
-			return res, nil
-		}
-		if recoverable(err) {
-			// Go back to previous state and wait again until the next
-			// connection attempt.
-			lg.Errorf("Received recoverable error %v", err)
-			c.setState(idleState, err, retryIn)
-			continue
+			lg.Debug("Attemting to open connection")
+			res, err := c.connectWith(arg)
+			if err == nil {
+				return res, nil
+			}
+			if recoverable(err) {
+				lg.Errorf("Received recoverable error %v", err)
+				// No need to reset stateDeadline as we are still in DISCONNECTED state. so
+				// another DISCONNECTED implies we haven't transitioned from DISCONNECTED
+				// state
+				c.setState(ConnectionStateDisconnected, err, retryIn)
+				reset()
+				continue
+			}
+			return nil, c.setState(ConnectionStateFailed, err, 0)
 		}
-		return nil, c.setState(ConnectionStateFailed, err, 0)
 	}
+
+	// (RTN14e)
+	lg.Debug("Transition to SUSPENDED state")
+	err = fmt.Errorf(connectionStateTTLErrFmt, stateTTL)
+	c.setState(ConnectionStateSuspended, err, c.opts.suspendedRetryTimeout())
+	// (RTN14f)
+	lg.Debug("Reached SUSPENDED state while opening connection")
+	return c.connectAfterSuspension(arg)
 }
 
 func (c *Connection) connectWith(arg connArgs) (result, error) {
 	c.mtx.Lock()
-	// set ably connection state to connecting, connecting state exists regardless of whether raw connection is successful or not
-	if !c.isActive() { // check if already in connecting state
+	if !c.isActive() {
 		c.lockSetState(ConnectionStateConnecting, nil, 0)
 	}
 	c.mtx.Unlock()
+
 	u, err := url.Parse(c.opts.realtimeURL())
 	if err != nil {
 		return nil, err
@@ -368,12 +393,6 @@ func (c *Connection) connectWith(arg connArgs) (result, error) {
 	}
 	u.RawQuery = query.Encode()
 	proto := c.opts.protocol()
-
-	if c.State() == ConnectionStateClosed { // RTN12d - if connection is closed by client, don't try to reconnect
-		return nopResult, nil
-	}
-
-	// if err is nil, raw connection with server is successful
 	conn, err := c.dial(proto, u)
 	if err != nil {
 		return nil, err
@@ -381,15 +400,11 @@ func (c *Connection) connectWith(arg connArgs) (result, error) {
 
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
-
 	if c.logger().Is(LogVerbose) {
 		c.setConn(verboseConn{conn: conn, logger: c.logger()})
 	} else {
 		c.setConn(conn)
 	}
-	// Start eventloop
-	go c.eventloop()
-
 	c.reconnecting = arg.mode == recoveryMode || arg.mode == resumeMode
 	c.arg = arg
 	return res, nil
@@ -455,7 +470,6 @@ func (c *Connection) Key() string {
 //
 // Ping returns non-nil error without any attempt of communication with Ably
 // if the connection state is ConnectionStateClosed or ConnectionStateFailed.
-// RTN13
 //func (c *Connection) Ping() (ping, pong time.Duration, err error) {
 //	return 0, 0, errors.New("TODO")
 //}
@@ -655,6 +669,7 @@ func (c *Connection) setConn(conn proto.Conn) {
 	c.connMtx.Lock()
 	c.conn = conn
 	c.connMtx.Unlock()
+	go c.eventloop()
 }
 
 func (c *Connection) logger() *LoggerOptions {
@@ -692,16 +707,11 @@ func (c *Connection) eventloop() {
 		c.connMtx.Unlock()
 		if err != nil {
 			c.mtx.Lock()
-			if c.state == ConnectionStateClosing {
-				// RTN12b, RTN12c
-				c.lockSetState(ConnectionStateClosed, err, 0)
-				c.mtx.Unlock()
-				return
-			}
 			if c.state == ConnectionStateClosed {
 				c.mtx.Unlock()
 				return
 			}
+
 			// RTN23a
 			c.lockSetState(ConnectionStateDisconnected, err, 0)
 			c.mtx.Unlock()
@@ -793,7 +803,7 @@ func (c *Connection) eventloop() {
 				// RTN12f
 				c.sendClose()
 				c.mtx.Unlock()
-				continue
+				return
 			}
 
 			c.mtx.Unlock()
@@ -940,16 +950,10 @@ func (c *Connection) lockSetState(state ConnectionState, err error, retryIn time
 	return c.errorReason.unwrapNil()
 }
 
-// ctxCancelOnStateTransition returns a context that is canceled when the
-// connection transitions to any state.
-//
-// This is useful for stopping timers when
-// another event has caused the connection to transition, thus invalidating the
-// original connection state at the time the timer was set.
 func (c *Connection) ctxCancelOnStateTransition() (context.Context, context.CancelFunc) {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	off := c.internalEmitter.OnceAll(func(change ConnectionStateChange) {
+	off := c.internalEmitter.OnceAll(func(ConnectionStateChange) {
 		cancel()
 	})
 


Test patch:
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index 25fea3c..323af2b 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -1,18 +1,12 @@
 package ably_test
 
 import (
-	"bytes"
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
-	"log"
-	"net"
 	"net/http"
-	"net/http/httptest"
 	"net/url"
-	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -183,426 +177,6 @@ func connectAndWait(t *testing.T, realtime *ably.Realtime) {
 	}
 }
 
-type connectionStateChanges chan ably.ConnectionStateChange
-
-func (c connectionStateChanges) Receive(change ably.ConnectionStateChange) {
-	c <- change
-}
-
-func TestRealtimeConn_RTN12_Connection_Close(t *testing.T) {
-
-	setUpWithEOF := func() (app *ablytest.Sandbox, client *ably.Realtime, doEOF chan struct{}) {
-		doEOF = make(chan struct{}, 1)
-
-		app, client = ablytest.NewRealtime(
-			ably.WithAutoConnect(false),
-			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
-				c, err := ablyutil.DialWebsocket(protocol, u, timeout)
-				return protoConnWithFakeEOF{Conn: c, doEOF: doEOF}, err
-			}))
-
-		err := ablytest.Wait(ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		connectionState := client.Connection.State()
-		if connectionState != ably.ConnectionStateConnected {
-			t.Fatalf("expected %v; got %v", ably.ConnectionStateConnected, connectionState)
-		}
-		return
-	}
-
-	setUpWithConnectingInterrupt := func() (app *ablytest.Sandbox, client *ably.Realtime, dialErr chan error, waitTillDial chan error) {
-		dialErr = make(chan error, 1)
-		waitTillDial = make(chan error)
-		app, client = ablytest.NewRealtime(
-			ably.WithAutoConnect(false),
-			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
-				waitTillDial <- nil
-				if err := <-dialErr; err != nil {
-					return nil, err
-				}
-				return ablyutil.DialWebsocket(protocol, u, timeout)
-			}))
-		return
-	}
-
-	t.Run("RTN12a: transition to closed on connection close", func(t *testing.T) {
-		t.Parallel()
-		app, client, _ := setUpWithEOF()
-		defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
-		stateChange := make(connectionStateChanges, 2)
-		client.Connection.OnAll(stateChange.Receive)
-
-		client.Close()
-
-		var change ably.ConnectionStateChange
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		if expected, got := ably.ConnectionStateClosing, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		if expected, got := ably.ConnectionStateClosed, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-		ablytest.Instantly.NoRecv(t, nil, stateChange, t.Fatalf)
-
-	})
-
-	t.Run("RTN12b: transition to closed on close request timeout", func(t *testing.T) {
-		t.Parallel()
-		connDetails := proto.ConnectionDetails{
-			ConnectionKey:      "foo",
-			ConnectionStateTTL: proto.DurationFromMsecs(time.Minute * 20),
-			MaxIdleInterval:    proto.DurationFromMsecs(time.Minute * 5),
-		}
-
-		afterCalls := make(chan ablytest.AfterCall)
-		now, after := ablytest.TimeFuncs(afterCalls)
-
-		var in chan *proto.ProtocolMessage
-		realtimeRequestTimeout := time.Minute
-		c, _ := ably.NewRealtime(
-			ably.WithAutoConnect(false),
-			ably.WithToken("fake:token"),
-			ably.WithRealtimeRequestTimeout(realtimeRequestTimeout),
-			ably.WithNow(now),
-			ably.WithAfter(after),
-			ably.WithDial(func(p string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
-				in = make(chan *proto.ProtocolMessage, 1)
-				out := make(chan *proto.ProtocolMessage, 16)
-				in <- &proto.ProtocolMessage{
-					Action:            proto.ActionConnected,
-					ConnectionID:      "connection",
-					ConnectionDetails: &connDetails,
-				}
-				return ablytest.MessagePipe(in, out,
-					ablytest.MessagePipeWithNowFunc(now),
-					ablytest.MessagePipeWithAfterFunc(after),
-				)(p, u, timeout)
-			}))
-
-		err := ablytest.Wait(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		stateChange := make(connectionStateChanges, 2)
-		c.Connection.OnAll(stateChange.Receive)
-
-		maxIdleInterval := time.Duration(connDetails.MaxIdleInterval)
-		receiveTimeout := realtimeRequestTimeout + maxIdleInterval
-
-		c.Close()
-		// RTN23a The connection should be closed due to lack of activity past
-		// receiveTimeout
-		var change ably.ConnectionStateChange
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		if expected, got := ably.ConnectionStateClosing, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		// Expect timer for a message receive.
-		var timer ablytest.AfterCall
-
-		ablytest.Instantly.Recv(t, &timer, afterCalls, t.Fatalf)
-		if expected, got := receiveTimeout, timer.D; expected != got {
-			t.Fatalf("expected %v, got %v", expected, got)
-		}
-
-		// Let the deadline pass without a message; expect a disconnection.
-		timer.Fire()
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateClosed, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		errReason := change.Reason
-		// make sure the reason is timeout
-		if !strings.Contains(errReason.Error(), "timeout") {
-			t.Errorf("expected %q to contain timeout", errReason.Error())
-		}
-
-		ablytest.Instantly.NoRecv(t, nil, stateChange, t.Fatalf)
-	})
-
-	t.Run("RTN12c: transition to closed on transport error", func(t *testing.T) {
-		t.Parallel()
-		app, client, doEOF := setUpWithEOF()
-		defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
-		stateChange := make(connectionStateChanges, 2)
-		client.Connection.OnAll(stateChange.Receive)
-
-		client.Close()
-
-		var change ably.ConnectionStateChange
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		if expected, got := ably.ConnectionStateClosing, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		doEOF <- struct{}{}
-
-		select {
-		case change = <-stateChange:
-		case <-time.After(50 * time.Millisecond):
-			t.Fatal("didn't transition on EOF")
-		}
-
-		if expected, got := ably.ConnectionStateClosed, change; expected != got.Current {
-			t.Fatalf("expected transition to %v, got %v", expected, got)
-		}
-
-		ablytest.Instantly.NoRecv(t, nil, stateChange, t.Fatalf)
-	})
-
-	t.Run("RTN12d : should abort reconnection timer while disconnected on closed", func(t *testing.T) {
-		t.Parallel()
-		ttl := 400 * time.Millisecond
-		disconnectTTl := 2 * ttl
-		connDetails := proto.ConnectionDetails{
-			ConnectionKey:      "foo",
-			ConnectionStateTTL: proto.DurationFromMsecs(time.Minute * 20),
-			MaxIdleInterval:    proto.DurationFromMsecs(time.Minute * 5),
-		}
-
-		afterCalls := make(chan ablytest.AfterCall)
-		now, after := ablytest.TimeFuncs(afterCalls)
-
-		var in chan *proto.ProtocolMessage
-
-		dialErr := make(chan error, 1)
-
-		realtimeRequestTimeout := time.Minute
-		c, _ := ably.NewRealtime(
-			ably.WithAutoConnect(false),
-			ably.WithToken("fake:token"),
-			ably.WithRealtimeRequestTimeout(realtimeRequestTimeout),
-			ably.WithConnectionStateTTL(ttl),
-			ably.WithDisconnectedRetryTimeout(disconnectTTl),
-			ably.WithNow(now),
-			ably.WithAfter(ablyutil.After),
-			ably.WithDial(func(p string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
-				if err := <-dialErr; err != nil {
-					return nil, err
-				}
-				in = make(chan *proto.ProtocolMessage, 1)
-				out := make(chan *proto.ProtocolMessage, 16)
-				in <- &proto.ProtocolMessage{
-					Action:            proto.ActionConnected,
-					ConnectionID:      "connection",
-					ConnectionDetails: &connDetails,
-				}
-				return ablytest.MessagePipe(in, out,
-					ablytest.MessagePipeWithNowFunc(now),
-					ablytest.MessagePipeWithAfterFunc(after),
-				)(p, u, timeout)
-			}))
-
-		dialErr <- nil
-		err := ablytest.Wait(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		stateChange := make(connectionStateChanges, 2)
-		c.Connection.OnAll(stateChange.Receive)
-
-		maxIdleInterval := time.Duration(connDetails.MaxIdleInterval)
-		receiveTimeout := realtimeRequestTimeout + maxIdleInterval
-
-		// Expect timer for a message receive.
-		var timer ablytest.AfterCall
-
-		ablytest.Instantly.Recv(t, &timer, afterCalls, t.Fatalf)
-		if expected, got := receiveTimeout, timer.D; expected != got {
-			t.Fatalf("expected %v, got %v", expected, got)
-		}
-
-		// Let the deadline pass without a message; expect a disconnection.
-		timer.Fire()
-		dialErr <- errors.New("can't reconnect once disconnected")
-
-		var change ably.ConnectionStateChange
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		// received disconnect on first receive
-		if expected, got := ably.ConnectionStateDisconnected, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		// retry connection with ably server
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateConnecting, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		dialErr <- errors.New("can't reconnect once disconnected")
-		// first reconnect failed (raw connection can't be established, outside retry loop), so returned disconnect
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateDisconnected, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		// Went inside retry loop for connecting with server again
-		c.Close()
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateClosed, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-		ablytest.Before(time.Second).NoRecv(t, nil, stateChange, t.Fatalf)
-	})
-
-	t.Run("RTN12d: should abort reconnection timer while suspended on closed", func(t *testing.T) {
-		t.Parallel()
-		ttl := 400 * time.Millisecond
-		disconnectTTl := 2 * ttl
-		suspendTTL := time.Second
-
-		connDetails := proto.ConnectionDetails{
-			ConnectionKey:      "foo",
-			ConnectionStateTTL: proto.DurationFromMsecs(time.Minute * 20),
-			MaxIdleInterval:    proto.DurationFromMsecs(time.Minute * 5),
-		}
-
-		afterCalls := make(chan ablytest.AfterCall)
-		now, after := ablytest.TimeFuncs(afterCalls)
-
-		var in chan *proto.ProtocolMessage
-
-		dialErr := make(chan error, 1)
-		realtimeRequestTimeout := time.Minute
-		c, _ := ably.NewRealtime(
-			ably.WithAutoConnect(false),
-			ably.WithToken("fake:token"),
-			ably.WithRealtimeRequestTimeout(realtimeRequestTimeout),
-			ably.WithNow(now),
-			ably.WithConnectionStateTTL(ttl),
-			ably.WithDisconnectedRetryTimeout(disconnectTTl),
-			ably.WithSuspendedRetryTimeout(suspendTTL),
-			ably.WithAfter(ablyutil.After),
-			ably.WithDial(func(p string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
-				if err := <-dialErr; err != nil {
-					return nil, err
-				}
-				in = make(chan *proto.ProtocolMessage, 1)
-				out := make(chan *proto.ProtocolMessage, 16)
-				in <- &proto.ProtocolMessage{
-					Action:            proto.ActionConnected,
-					ConnectionID:      "connection",
-					ConnectionDetails: &connDetails,
-				}
-				return ablytest.MessagePipe(in, out,
-					ablytest.MessagePipeWithNowFunc(now),
-					ablytest.MessagePipeWithAfterFunc(after),
-				)(p, u, timeout)
-			}))
-
-		dialErr <- nil
-		err := ablytest.Wait(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		stateChange := make(connectionStateChanges, 2)
-		c.Connection.OnAll(stateChange.Receive)
-
-		maxIdleInterval := time.Duration(connDetails.MaxIdleInterval)
-		receiveTimeout := realtimeRequestTimeout + maxIdleInterval
-
-		// Expect timer for a message receive.
-		var timer ablytest.AfterCall
-
-		ablytest.Instantly.Recv(t, &timer, afterCalls, t.Fatalf)
-		if expected, got := receiveTimeout, timer.D; expected != got {
-			t.Fatalf("expected %v, got %v", expected, got)
-		}
-
-		// Let the deadline pass without a message; expect a disconnection.
-		timer.Fire()
-		dialErr <- errors.New("can't reconnect once disconnected")
-
-		var change ably.ConnectionStateChange
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-
-		// received disconnect on first receive
-		if expected, got := ably.ConnectionStateDisconnected, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-		// retry connection with ably server
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateConnecting, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		dialErr <- errors.New("can't reconnect once disconnected")
-		// first connect failed (raw connection can't be established), so returned disconnect
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateDisconnected, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		// after suspended state goes into suspend retry loop indefinitely
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateSuspended, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		c.Close()
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateClosed, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change.Current)
-		}
-
-		ablytest.Before(time.Second).NoRecv(t, nil, stateChange, t.Fatalf)
-	})
-
-	t.Run("RTN12f: transition to closed when close is called intermittently", func(t *testing.T) {
-		t.Parallel()
-		app, client, interrupt, waitTillConnecting := setUpWithConnectingInterrupt()
-		defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
-		stateChange := make(ably.ConnStateChanges, 10)
-		off := client.Connection.OnAll(stateChange.Receive)
-		defer off()
-
-		client.Connect()
-
-		var change ably.ConnectionStateChange
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateConnecting, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
-		}
-
-		<-waitTillConnecting
-
-		client.Close()
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateClosing, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
-		}
-
-		interrupt <- nil // continue to connected
-
-		ablytest.Soon.Recv(t, &change, stateChange, t.Fatalf)
-		if expected, got := ably.ConnectionStateClosed, change.Current; expected != got {
-			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
-		}
-
-	})
-}
-
 func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 	t.Parallel()
 
@@ -710,6 +284,7 @@ func (c protoConnWithFakeEOF) Receive(deadline time.Time) (*proto.ProtocolMessag
 	go func() {
 		msg, err := c.Conn.Receive(deadline)
 		received <- result{msg: msg, err: err}
+
 	}()
 
 	select {
@@ -2251,69 +1826,61 @@ func (w *writerLogger) Printf(level ably.LogLevel, format string, v ...interface
 	fmt.Fprintf(w.w, format, v...)
 }
 
-func TestRealtimeConn_RTN14c(t *testing.T) {
+func TestRealtimeConn_RTN14c_ConnectedTimeout(t *testing.T) {
 	t.Parallel()
-	ts := httptest.NewUnstartedServer(nil)
-	reqTimeout := 5 * time.Millisecond
-	ts.Config = &http.Server{
-		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			time.Sleep(reqTimeout + 5*time.Millisecond)
-		}),
-	}
-	ts.TLS = &tls.Config{
-		GetConfigForClient: func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
-			time.Sleep(reqTimeout * 2)
-			return nil, context.DeadlineExceeded
-		},
-	}
-	ts.StartTLS()
-	defer ts.Close()
 
-	host, port, err := net.SplitHostPort(strings.TrimPrefix(ts.URL, "https://"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	p, _ := strconv.Atoi(port)
-	var buf bytes.Buffer
-	lg := &writerLogger{w: &buf}
-	client, err := ably.NewRealtime(
-		ably.WithKey("xxx:xxx"),
-		ably.WithRealtimeHost(host),
-		ably.WithTLSPort(p),
-		ably.WithRealtimeRequestTimeout(reqTimeout),
-		ably.WithLogHandler(lg),
-		ably.WithLogLevel(ably.LogDebug),
+	afterCalls := make(chan ablytest.AfterCall)
+	now, after := ablytest.TimeFuncs(afterCalls)
+
+	in := make(chan *proto.ProtocolMessage, 10)
+	out := make(chan *proto.ProtocolMessage, 10)
+
+	c, err := ably.NewRealtime(
 		ably.WithAutoConnect(false),
+		ably.WithToken("fake:token"),
+		ably.WithNow(now),
+		ably.WithAfter(after),
+		ably.WithDial(ablytest.MessagePipe(in, out,
+			ablytest.MessagePipeWithNowFunc(now),
+			ablytest.MessagePipeWithAfterFunc(after),
+		)),
 	)
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer client.Close()
-	ok := make(chan struct{}, 1)
-	client.Connection.On(ably.ConnectionEventDisconnected, func(csc ably.ConnectionStateChange) {
-		ok <- struct{}{}
-	})
-	client.Connect()
-	ablytest.Soon.Recv(t, nil, ok, t.Fatalf)
-	sub := "Dial Failed in "
-	x := buf.String()
-	idx := strings.Index(x, sub)
-	if idx == -1 {
-		t.Fatal("Missing Dial i/o timeout error")
-	}
-	x = x[idx+len(sub):]
-	x = strings.TrimSpace(x)
-	x = strings.Split(x, " ")[0]
-
-	d, err := time.ParseDuration(x)
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer c.Close()
+
+	stateChanges := make(ably.ConnStateChanges, 10)
+	off := c.Connection.OnAll(stateChanges.Receive)
+	defer off()
+
+	c.Connect()
+
+	var change ably.ConnectionStateChange
+	ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
+	if expected, got := connecting, change.Current; expected != got {
+		t.Fatalf("expected %v; got %v", expected, got)
+	}
+
+	// Once dialed, expect a timer for realtimeRequestTimeout.
+
+	const realtimeRequestTimeout = 10 * time.Second // DF1b
 
-	// We can't be precise here so just an estimate. Manual tests ranges between 5ms -10ms
-	// TODO: Find a proper way to record Dial i/o timeouts duration.
-	if d < reqTimeout || d > reqTimeout+(2*time.Millisecond) {
-		t.Errorf("expected i/o timeout to be %v got %v", reqTimeout, d)
+	var receiveTimer ablytest.AfterCall
+	ablytest.Instantly.Recv(t, &receiveTimer, afterCalls, t.Fatalf)
+
+	if expected, got := realtimeRequestTimeout, receiveTimer.D; expected != got {
+		t.Fatalf("expected %v; got %v", expected, got)
+	}
+
+	// Expect a state change to DISCONNECTED when the timer expires.
+	ablytest.Instantly.NoRecv(t, nil, stateChanges, t.Fatalf)
+
+	receiveTimer.Fire()
+
+	ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
+	if expected, got := disconnected, change.Current; expected != got {
+		t.Fatalf("expected %v; got %v", expected, got)
 	}
 }
 
diff --git a/ably/transitioner_test.go b/ably/transitioner_test.go
index 783bed8..cd6f134 100644
--- a/ably/transitioner_test.go
+++ b/ably/transitioner_test.go
@@ -160,7 +160,7 @@ func (c ConnTransitioner) finishConnecting(err error) connTransitionFunc {
 
 			return connNextStates{
 					connecting: c.recover(reconnect),
-					suspended:  c.fireSuspend(reconnect, suspend),
+					suspended:  c.fireSuspend(suspend),
 				}, func() {
 					c.Realtime.Close() // cancel timers; don't reconnect
 				}
@@ -220,10 +220,9 @@ func (c ConnTransitioner) recover(timer ablytest.AfterCall) connTransitionFunc {
 	}
 }
 
-func (c ConnTransitioner) fireSuspend(reconnectTimer ablytest.AfterCall, suspendTimer ablytest.AfterCall) connTransitionFunc {
+func (c ConnTransitioner) fireSuspend(timer ablytest.AfterCall) connTransitionFunc {
 	return func() (connNextStates, func()) {
-		suspendTimer.Fire()
-		reconnectTimer.Fire()
+		timer.Fire()
 		return c.suspend()
 	}
 }