Repository: ably/ably-go
Pull request #159 (2020-04-27 19:06:13)

add implementation for  RTN15b-c spec
Covers

- [x] [(RTN15b) ](https://docs.ably.io/client-lib-development-guide/features/#RTN15b)
  - [x] [(RTN15b1) ](https://docs.ably.io/client-lib-development-guide/features/#RTN15b1)
  - [x] [(RTN15b2) ](https://docs.ably.io/client-lib-development-guide/features/#RTN15b2)
- [ ] [(RTN15c) ]()
  - [x] [(RTN15c1) ](https://docs.ably.io/client-lib-development-guide/features/#RTN15c1)
  - [x] [(RTN15c2) ](https://docs.ably.io/client-lib-development-guide/features/#RTN15c2)
  - [x] [(RTN15c3) ]()
  - [x] [(RTN15c4) ]()
  - [ ] [(RTN15c5) ]()


Patch:
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index be66b14..a167730 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -156,10 +156,16 @@ var attachResultStates = []StateEnum{
 }
 
 func (c *RealtimeChannel) attach(result bool) (Result, error) {
+	return c.mayAttach(result, true)
+}
+
+func (c *RealtimeChannel) mayAttach(result, checkActive bool) (Result, error) {
 	c.state.Lock()
 	defer c.state.Unlock()
-	if c.isActive() {
-		return nopResult, nil
+	if checkActive {
+		if c.isActive() {
+			return nopResult, nil
+		}
 	}
 	if !c.client.Connection.lockIsActive() {
 		return nil, c.state.set(StateChanFailed, errAttach)
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 0a7d755..8e3e0ae 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -1,6 +1,7 @@
 package ably
 
 import (
+	"net/http"
 	"time"
 
 	"github.com/ably/ably-go/ably/proto"
@@ -30,7 +31,9 @@ func NewRealtimeClient(opts *ClientOptions) (*RealtimeClient, error) {
 	c.rest = rest
 	c.Auth = rest.Auth
 	c.Channels = newChannels(c)
-	conn, err := newConn(c.opts(), rest.Auth, c.onChannelMsg, c.onConnStateChange)
+	conn, err := newConn(c.opts(), rest.Auth, connCallbacks{
+		c.onChannelMsg, c.onReconnectMsg, c.onConnStateChange,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -59,6 +62,34 @@ func (c *RealtimeClient) onChannelMsg(msg *proto.ProtocolMessage) {
 	c.Channels.Get(msg.Channel).notify(msg)
 }
 
+func (c *RealtimeClient) onReconnectMsg(msg *proto.ProtocolMessage) {
+	switch msg.Action {
+	case proto.ActionConnected:
+		if msg.Error != nil {
+			// (RTN15c3)
+			for _, ch := range c.Channels.All() {
+				switch ch.State() {
+				case StateConnSuspended:
+					ch.attach(false)
+				case StateChanAttaching, StateChanAttached:
+					ch.mayAttach(false, false)
+				}
+			}
+		}
+
+	case proto.ActionError:
+		// (RTN15c4)
+
+		for _, ch := range c.Channels.All() {
+			ch.state.syncSet(StateChanFailed, msg.Error)
+		}
+	}
+}
+
+func tokenError(err *proto.ErrorInfo) bool {
+	return err.StatusCode == http.StatusUnauthorized && (40140 <= err.Code && err.Code < 40150)
+}
+
 func (c *RealtimeClient) onConnStateChange(state State) {
 	// TODO: Replace with EventEmitter https://github.com/ably/ably-go/pull/144
 	c.Channels.broadcastConnStateChange(state)
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 2178731..a0e3775 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -18,30 +18,42 @@ var (
 // Conn represents a single connection RealtimeClient instantiates for
 // communication with Ably servers.
 type Conn struct {
-	details       proto.ConnectionDetails
-	id            string
-	serial        int64
-	msgSerial     int64
-	err           error
-	conn          proto.Conn
-	opts          *ClientOptions
-	state         *stateEmitter
-	stateCh       chan State
-	pending       pendingEmitter
-	queue         *msgQueue
-	auth          *Auth
-	onChannelMsg  func(*proto.ProtocolMessage)
-	onStateChange func(State)
-}
-
-func newConn(opts *ClientOptions, auth *Auth, onChannelMsg func(*proto.ProtocolMessage), onStateChange func(State)) (*Conn, error) {
+	details      proto.ConnectionDetails
+	id           string
+	serial       int64
+	msgSerial    int64
+	err          error
+	conn         proto.Conn
+	opts         *ClientOptions
+	state        *stateEmitter
+	stateCh      chan State
+	pending      pendingEmitter
+	queue        *msgQueue
+	auth         *Auth
+	callbacks    connCallbacks
+	reconnecting bool
+}
+
+type connCallbacks struct {
+	onChannelMsg func(*proto.ProtocolMessage)
+	// onReconnectMsg is called when we get a response from reconnect request. We
+	// move this up because some implementation details for (RTN15c) requires
+	// access to Channels and we dont have it here so we let RealtimeClient do the
+	// work.
+	onReconnectMsg func(*proto.ProtocolMessage)
+	onStateChange  func(State)
+	// reconnecting tracks if we have issued a reconnection request. If we receive any message
+	// with this set to true then its the first message/response after issuing the
+	// reconnection request.
+}
+
+func newConn(opts *ClientOptions, auth *Auth, callbacks connCallbacks) (*Conn, error) {
 	c := &Conn{
-		opts:          opts,
-		state:         newStateEmitter(StateConn, StateConnInitialized, "", auth.logger()),
-		pending:       newPendingEmitter(auth.logger()),
-		auth:          auth,
-		onChannelMsg:  onChannelMsg,
-		onStateChange: onStateChange,
+		opts:      opts,
+		state:     newStateEmitter(StateConn, StateConnInitialized, "", auth.logger()),
+		pending:   newPendingEmitter(auth.logger()),
+		auth:      auth,
+		callbacks: callbacks,
 	}
 	c.queue = newMsgQueue(c)
 	if opts.Listener != nil {
@@ -90,7 +102,17 @@ func (c *Conn) reconnect(result bool) (Result, error) {
 	connKey := c.details.ConnectionKey
 	connSerial := c.serial
 	c.state.Unlock()
-	return c.connectWithRecovery(result, connKey, connSerial)
+	r, err := c.connectWithRecovery(result, connKey, connSerial)
+	if err != nil {
+		return nil, err
+	}
+	// We have successfully dialed reconnection request. We need to set this so
+	// when the next message arrives it will be treated as the response to
+	// reconnection request.
+	c.state.Lock()
+	c.reconnecting = true
+	c.state.Unlock()
+	return r, nil
 }
 
 func (c *Conn) connectWithRecovery(result bool, connKey string, connSerial int64) (Result, error) {
@@ -390,28 +412,65 @@ func (c *Conn) eventloop() {
 			c.state.Unlock()
 		case proto.ActionError:
 			if msg.Channel != "" {
-				c.onChannelMsg(msg)
+				c.callbacks.onChannelMsg(msg)
 				break
 			}
 			c.state.Lock()
+			if c.reconnecting {
+				c.reconnecting = false
+				if tokenError(msg.Error) {
+					// (RTN15c5)
+					// TODO: (gernest) implement (RTN15h) This can be done as a separate task?
+				} else {
+					// (RTN15c4)
+					c.callbacks.onReconnectMsg(msg)
+				}
+			}
 			c.setState(StateConnFailed, newErrorProto(msg.Error))
 			c.state.Unlock()
 			c.queue.Fail(newErrorProto(msg.Error))
 		case proto.ActionConnected:
 			c.auth.updateClientID(msg.ConnectionDetails.ClientID)
-			c.state.Lock()
-			c.id = msg.ConnectionID
 			if msg.ConnectionDetails != nil {
+				c.state.Lock()
 				c.details = *msg.ConnectionDetails
+				c.state.Unlock()
+
 				// Spec RSA7b3, RSA7b4, RSA12a
 				c.auth.updateClientID(c.details.ClientID)
 
 				maxIdleInterval := time.Duration(msg.ConnectionDetails.MaxIdleInterval) * time.Millisecond
 				receiveTimeout = c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
 			}
+			c.state.Lock()
+			reconnecting := c.reconnecting
+			if reconnecting {
+				c.reconnecting = false
+			}
+			c.state.Unlock()
+			if reconnecting {
+				// (RTN15c1) (RTN15c2)
+				c.state.Lock()
+				c.setState(StateConnConnected, msg.Error)
+				id := c.id
+				c.state.Unlock()
+				if id != msg.ConnectionID {
+					// (RTN15c3)
+					// we are calling this outside of locks to avoid deadlock because in the
+					// RealtimeClient client where this callback is implemented we do some ops
+					// with this Conn where we re acquire Conn.state.Lock again.
+					c.callbacks.onReconnectMsg(msg)
+				}
+			} else {
+				// preserve old behavior.
+				c.state.Lock()
+				c.setState(StateConnConnected, nil)
+				c.state.Unlock()
+			}
+			c.state.Lock()
+			c.id = msg.ConnectionID
 			c.serial = -1
 			c.msgSerial = 0
-			c.setState(StateConnConnected, nil)
 			c.state.Unlock()
 			c.queue.Flush()
 		case proto.ActionDisconnected:
@@ -425,7 +484,7 @@ func (c *Conn) eventloop() {
 			c.setState(StateConnClosed, nil)
 			c.state.Unlock()
 		default:
-			c.onChannelMsg(msg)
+			c.callbacks.onChannelMsg(msg)
 		}
 	}
 }
@@ -437,7 +496,7 @@ func (c *Conn) setState(state StateEnum, err error) error {
 	// EventEmitter at https://github.com/ably/ably-go/pull/144.
 	ch := make(chan State, 1)
 	c.state.once(ch)
-	go func() { c.onStateChange(<-ch) }()
+	go func() { c.callbacks.onStateChange(<-ch) }()
 
 	return c.state.set(state, err)
 }


Test patch:
diff --git a/ably/ably_test.go b/ably/ably_test.go
index 96a1ca1..ef8ae55 100644
--- a/ably/ably_test.go
+++ b/ably/ably_test.go
@@ -17,6 +17,23 @@ func nonil(err ...error) error {
 	return nil
 }
 
+type closeClient struct {
+	io.Closer
+	skip []int
+}
+
+func (c *closeClient) Close() error {
+	e := c.Closer.Close()
+	if a, ok := e.(*ably.Error); ok {
+		for _, v := range c.skip {
+			if a.StatusCode == v {
+				return nil
+			}
+		}
+	}
+	return e
+}
+
 func safeclose(t *testing.T, closers ...io.Closer) {
 	type failed struct {
 		i   int
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index c476868..f182efe 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -1,7 +1,9 @@
 package ably_test
 
 import (
+	"fmt"
 	"io"
+	"net/http"
 	"net/url"
 	"testing"
 	"time"
@@ -102,7 +104,8 @@ func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 
 type protoConnWithFakeEOF struct {
 	proto.Conn
-	doEOF <-chan struct{}
+	doEOF     <-chan struct{}
+	onMessage func(msg *proto.ProtocolMessage)
 }
 
 func (c protoConnWithFakeEOF) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
@@ -121,8 +124,684 @@ func (c protoConnWithFakeEOF) Receive(deadline time.Time) (*proto.ProtocolMessag
 
 	select {
 	case r := <-received:
+		if c.onMessage != nil {
+			c.onMessage(r.msg)
+		}
 		return r.msg, r.err
 	case <-c.doEOF:
 		return nil, io.EOF
 	}
 }
+
+func TestRealtimeConn_RTN15b(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	gotDial := make(chan chan struct{})
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				m.messages = append(m.messages, msg)
+			}}, 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)
+	}
+
+	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)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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)
+	}
+	if len(metaList) != 2 {
+		t.Fatalf("expected 2 connection dialing got %d", len(metaList))
+	}
+
+	{ //(RTN15b1)
+		u := metaList[1].dial
+		resume := u.Query().Get("resume")
+		connKey := recent(metaList[0].messages, proto.ActionConnected).ConnectionDetails.ConnectionKey
+		if resume == "" {
+			t.Fatal("expected resume query param to be set")
+		}
+		if resume != connKey {
+			t.Errorf("resume: expected %q got %q", connKey, resume)
+		}
+	}
+
+	{ //(RTN15b2)
+		u := metaList[1].dial
+		serial := u.Query().Get("connectionSerial")
+		connSerial := fmt.Sprint(metaList[0].messages[0].ConnectionSerial)
+		if serial == "" {
+			t.Fatal("expected connectionSerial query param to be set")
+		}
+		if serial != connSerial {
+			t.Errorf("connectionSerial: expected %q got %q", connSerial, serial)
+		}
+	}
+}
+
+func recent(msgs []*proto.ProtocolMessage, action proto.Action) *proto.ProtocolMessage {
+	for i := len(msgs) - 1; i >= 0; i-- {
+		if msgs[i].Action == action {
+			return msgs[i]
+		}
+	}
+	return nil
+}
+
+func TestRealtimeConn_RTN15c1(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	gotDial := make(chan chan struct{})
+
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				m.messages = append(m.messages, msg)
+			}}, 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)
+	}
+
+	chanStateChanges := make(chan ably.State)
+	channel.On(chanStateChanges)
+
+	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)
+	}
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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 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")
+	}
+
+	select {
+	case state = <-chanStateChanges:
+		t.Fatal("expected no channel state changes")
+	case <-time.After(50 * time.Millisecond):
+		// (RTN15c1)
+		//
+		// - current connectionId == resume message connectionId
+		// - resume message has no error
+		// - no channel state changes happened.
+		if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
+			t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
+		}
+		if metaList[1].messages[0].Error != nil {
+			t.Error("expected resume error to be nil")
+		}
+	}
+
+}
+func TestRealtimeConn_RTN15c2(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	errInfo := &proto.ErrorInfo{
+		StatusCode: 401,
+	}
+
+	gotDial := make(chan chan struct{})
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				if len(metaList) == 2 && len(m.messages) == 0 {
+					msg.Error = errInfo
+				}
+				m.messages = append(m.messages, msg)
+			}}, 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)
+	}
+	chanStateChanges := make(chan ably.State)
+	channel.On(chanStateChanges)
+
+	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)
+	}
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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 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")
+	}
+
+	<-stateChanges
+	select {
+	case state = <-chanStateChanges:
+		t.Fatal("expected no channel state changes")
+	case <-time.After(50 * time.Millisecond):
+		// (RTN15c2)
+		//
+		// - current connectionId == resume message connectionId
+		// - resume message has an error
+		// - Conn.Reqson == message resume error
+		// - no channel state changes happened.
+		if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
+			t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
+		}
+		if metaList[1].messages[0].Error == nil {
+			t.Error("expected resume error")
+		}
+		err = client.Connection.Reason()
+		if err == nil {
+			t.Fatal("expected reason to be set")
+		}
+		reason := err.(*proto.ErrorInfo)
+		if reason.StatusCode != errInfo.StatusCode {
+			t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
+		}
+	}
+}
+func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	errInfo := &proto.ErrorInfo{
+		StatusCode: 401,
+	}
+	connID := "new-conn-id"
+	gotDial := make(chan chan struct{})
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				if len(metaList) == 2 && len(m.messages) == 0 {
+					msg.Error = errInfo
+					msg.ConnectionID = connID
+				}
+				m.messages = append(m.messages, msg)
+			}}, 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)
+	}
+	chanStateChanges := make(chan ably.State, 18)
+	channel.On(chanStateChanges, ably.StateChanAttaching)
+
+	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)
+	}
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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)
+	}
+
+	<-stateChanges
+	select {
+	case state = <-chanStateChanges:
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("didn't change state")
+	}
+	// we are testing to make sure we have initiated a new attach for channels
+	// in ATTACHED state.
+	if expected, got := ably.StateChanAttaching, state; expected != got.State {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+	err = client.Connection.Reason()
+	if err == nil {
+		t.Fatal("expected reason to be set")
+	}
+	reason := err.(*proto.ErrorInfo)
+	if reason.StatusCode != errInfo.StatusCode {
+		t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
+	}
+}
+
+func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	errInfo := &proto.ErrorInfo{
+		StatusCode: 401,
+	}
+	connID := "new-conn-id"
+	gotDial := make(chan chan struct{})
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				if len(metaList) == 2 && len(m.messages) == 0 {
+					msg.Error = errInfo
+					msg.ConnectionID = connID
+				}
+				if msg.Action == proto.ActionAttached {
+					msg.Action = proto.ActionHeartbeat
+				}
+				m.messages = append(m.messages, msg)
+			}}, 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 := channel.Attach(); 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)
+	}
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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)
+	}
+
+	<-stateChanges
+
+	// we are testing to make sure we have initiated a new attach for channels
+	// in ATTACHING  state.
+	if expected, got := ably.StateChanAttaching, channel.State(); expected != got {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+	err = client.Connection.Reason()
+	if err == nil {
+		t.Fatal("expected reason to be set")
+	}
+	reason := err.(*proto.ErrorInfo)
+	if reason.StatusCode != errInfo.StatusCode {
+		t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
+	}
+}
+
+func TestRealtimeConn_RTN15c4(t *testing.T) {
+	t.Parallel()
+
+	doEOF := make(chan struct{}, 1)
+
+	type meta struct {
+		dial     *url.URL
+		messages []*proto.ProtocolMessage
+	}
+
+	var metaList []*meta
+	errInfo := &proto.ErrorInfo{
+		StatusCode: http.StatusBadRequest,
+	}
+	gotDial := make(chan chan struct{})
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		NoConnect: true,
+		Dial: func(protocol string, u *url.URL) (proto.Conn, error) {
+			m := &meta{dial: u}
+			metaList = append(metaList, m)
+			if len(metaList) > 1 {
+				goOn := make(chan struct{})
+				gotDial <- goOn
+				<-goOn
+			}
+			c, err := ablyutil.DialWebsocket(protocol, u)
+			return protoConnWithFakeEOF{Conn: c, doEOF: doEOF, onMessage: func(msg *proto.ProtocolMessage) {
+				if len(metaList) == 2 && len(m.messages) == 0 {
+					msg.Action = proto.ActionError
+					msg.Error = errInfo
+				}
+				m.messages = append(m.messages, msg)
+			}}, err
+		},
+	})
+	defer safeclose(t, &closeClient{Closer: client, skip: []int{http.StatusBadRequest}}, 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)
+	}
+	chanStateChanges := make(chan ably.State, 1)
+	channel.On(chanStateChanges, ably.StateChanFailed)
+
+	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)
+	}
+	rest, err := ably.NewRestClient(app.Options())
+	if err != nil {
+		t.Fatal(err)
+	}
+	goOn := <-gotDial
+	err = rest.Channels.Get("channel", nil).Publish("name", "data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(goOn)
+
+	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)
+	}
+	<-stateChanges
+	select {
+	case state = <-chanStateChanges:
+	case <-time.After(50 * time.Millisecond):
+		t.Fatal("didn't change state")
+	}
+	if expected, got := ably.StateChanFailed, state; expected != got.State {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+	err = client.Connection.Reason()
+	if err == nil {
+		t.Fatal("expected reason to be set")
+	}
+	reason := err.(*ably.Error)
+	if reason.StatusCode != errInfo.StatusCode {
+		t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
+	}
+	// The client should transition to the FAILED state
+	if expected, got := ably.StateConnFailed, client.Connection.State(); expected != got {
+		t.Fatalf("expected transition to %v, got %v", expected, got)
+	}
+}