Repository: ably/ably-go
Pull request #170 (2020-07-18 08:01:59)

add RTN19 - Transport state side effect
-  [x] RTN19a
- [x] RTN19b)

lmars: @gernest what is the status of this PR? This came up in https://github.com/ably/ably-go/pull/208#discussion_r519712866, would be good to get an ETA on when we expect this to be delivered.

Patch:
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index e3f3618..9c1945c 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -217,11 +217,7 @@ func (c *RealtimeChannel) lockAttach(err error) (Result, error) {
 		Action:  proto.ActionAttach,
 		Channel: c.Name,
 	}
-	err = c.client.Connection.send(msg, nil)
-	if err != nil {
-		return nil, c.setState(ChannelStateFailed, err)
-	}
-
+	c.client.Connection.send(msg, nil)
 	return res, nil
 }
 
@@ -242,12 +238,22 @@ func (c *RealtimeChannel) Detach(ctx context.Context) error {
 func (c *RealtimeChannel) detach() (Result, error) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
-	switch {
-	case c.state == ChannelStateFailed:
-		return nil, channelStateError(ChannelStateFailed, errDetach)
-	case !c.isActive():
+	if !c.isActive() {
 		return nopResult, nil
 	}
+	return c.detachUnsafe()
+}
+
+func (c *RealtimeChannel) detachSkipVerifyActive() (Result, error) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.detachUnsafe()
+}
+
+func (c *RealtimeChannel) detachUnsafe() (Result, error) {
+	if c.state == ChannelStateFailed {
+		return nil, channelStateError(ChannelStateFailed, errDetach)
+	}
 	if !c.client.Connection.lockIsActive() {
 		return nil, c.lockSetState(ChannelStateFailed, errDetach)
 	}
@@ -257,10 +263,7 @@ func (c *RealtimeChannel) detach() (Result, error) {
 		Action:  proto.ActionDetach,
 		Channel: c.Name,
 	}
-	err := c.client.Connection.send(msg, nil)
-	if err != nil {
-		return nil, c.lockSetState(ChannelStateFailed, err)
-	}
+	c.client.Connection.send(msg, nil)
 	return res, nil
 }
 
@@ -433,9 +436,7 @@ func (c *RealtimeChannel) send(msg *proto.ProtocolMessage) (Result, error) {
 	}
 
 	res, listen := newErrResult()
-	if err := c.client.Connection.send(msg, listen); err != nil {
-		return nil, err
-	}
+	c.client.Connection.send(msg, listen)
 	return res, nil
 }
 
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 1baeb76..715897c 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -73,6 +73,8 @@ func (c *Realtime) onReconnected(isNewID bool) {
 		for _, ch := range c.Channels.All() {
 			ch.queue.Flush()
 		}
+		//RTN19a
+		c.Connection.resendPending()
 		return
 	}
 
@@ -81,8 +83,12 @@ func (c *Realtime) onReconnected(isNewID bool) {
 		// TODO: SUSPENDED
 		case ChannelStateAttaching, ChannelStateAttached:
 			ch.mayAttach(false)
+		case ChannelStateDetaching:
+			ch.detachSkipVerifyActive()
 		}
 	}
+	//RTN19a
+	c.Connection.resendPending()
 }
 
 func (c *Realtime) onReconnectionFailed(err *proto.ErrorInfo) {
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index e4dd14c..5ed47f6 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -369,9 +369,12 @@ loop:
 }
 
 func (c *Connection) connectWith(arg connArgs) (Result, error) {
+	c.mtx.Lock()
 	if !c.isActive() {
-		c.setState(ConnectionStateConnecting, nil, 0)
+		c.lockSetState(ConnectionStateConnecting, nil, 0)
 	}
+	c.mtx.Unlock()
+
 	u, err := url.Parse(c.opts.realtimeURL())
 	if err != nil {
 		return nil, err
@@ -434,7 +437,6 @@ func (c *Connection) close() {
 
 func (c *Connection) sendClose() {
 	msg := &proto.ProtocolMessage{Action: proto.ActionClose}
-	c.updateSerial(msg, nil)
 
 	// TODO: handle error. If you can't send a message, the fail-fast way to
 	// deal with it is to discard the WebSocket and perform a normal
@@ -564,37 +566,51 @@ func (em ConnectionEventEmitter) OffAll() {
 	em.emitter.OffAll()
 }
 
-func (c *Connection) updateSerial(msg *proto.ProtocolMessage, listen chan<- error) {
+func (c *Connection) advanceSerial() {
 	const maxint64 = 1<<63 - 1
-	msg.MsgSerial = c.msgSerial
 	c.msgSerial = (c.msgSerial + 1) % maxint64
-	if listen != nil {
-		c.pending.Enqueue(msg.MsgSerial, listen)
-	}
 }
 
-func (c *Connection) send(msg *proto.ProtocolMessage, listen chan<- error) error {
+func (c *Connection) send(msg *proto.ProtocolMessage, listen chan<- error) {
 	c.mtx.Lock()
 	switch state := c.state; state {
+	default:
+		c.mtx.Unlock()
+		listen <- connStateError(state, nil)
+
 	case ConnectionStateInitialized, ConnectionStateConnecting, ConnectionStateDisconnected:
 		c.mtx.Unlock()
 		if c.opts.NoQueueing {
-			return connStateError(state, errQueueing)
+			listen <- connStateError(state, errQueueing)
 		}
 		c.queue.Enqueue(msg, listen)
-		return nil
+
 	case ConnectionStateConnected:
-	default:
-		c.mtx.Unlock()
-		return connStateError(state, nil)
-	}
-	if err := c.verifyAndUpdateMessages(msg); err != nil {
-		c.mtx.Unlock()
-		return err
+		if err := c.verifyAndUpdateMessages(msg); err != nil {
+			c.mtx.Unlock()
+			listen <- err
+			return
+		}
+		msg.MsgSerial = c.msgSerial
+		err := c.conn.Send(msg)
+		if err != nil {
+			// An error here means there has been some transport-level failure in the
+			// connection. The connection itself is probably discarded, which causes the
+			// concurrent Receive in eventloop to fail, which in turn starts the
+			// reconnection logic. But in case it isn't, force that by closing the
+			// connection. Otherwise, the message we enqueue here may be in the queue
+			// indefinitely.
+			c.conn.Close()
+			c.mtx.Unlock()
+			c.queue.Enqueue(msg, listen)
+		} else {
+			c.advanceSerial()
+			if listen != nil {
+				c.pending.Enqueue(msg, listen)
+			}
+			c.mtx.Unlock()
+		}
 	}
-	c.updateSerial(msg, listen)
-	defer c.mtx.Unlock()
-	return c.conn.Send(msg)
 }
 
 // verifyAndUpdateMessages ensures the ClientID sent with published messages or
@@ -664,6 +680,19 @@ func (c *Connection) setSerial(serial int64) {
 	c.serial = serial
 }
 
+func (c *Connection) resendPending() {
+	lg := c.logger().sugar()
+	c.mtx.Lock()
+	cx := make([]msgCh, len(c.pending.queue))
+	copy(cx, c.pending.queue)
+	c.pending.queue = []msgCh{}
+	c.mtx.Unlock()
+	lg.Debugf("resending %d messages waiting for ACK/NACK", len(cx))
+	for _, v := range cx {
+		c.send(v.msg, v.ch)
+	}
+}
+
 func (c *Connection) eventloop() {
 	var lastActivityAt time.Time
 	var connDetails *proto.ConnectionDetails
@@ -703,12 +732,12 @@ func (c *Connection) eventloop() {
 		case proto.ActionHeartbeat:
 		case proto.ActionAck:
 			c.mtx.Lock()
-			c.pending.Ack(msg.MsgSerial, msg.Count, newErrorFromProto(msg.Error))
+			c.pending.Ack(msg, newErrorFromProto(msg.Error))
 			c.setSerial(c.serial + 1)
 			c.mtx.Unlock()
 		case proto.ActionNack:
 			c.mtx.Lock()
-			c.pending.Nack(msg.MsgSerial, msg.Count, newErrorFromProto(msg.Error))
+			c.pending.Nack(msg, newErrorFromProto(msg.Error))
 			c.mtx.Unlock()
 		case proto.ActionError:
 
diff --git a/ably/state.go b/ably/state.go
index 4cc7ca6..ea4f963 100644
--- a/ably/state.go
+++ b/ably/state.go
@@ -103,7 +103,7 @@ func channelStateError(state ChannelState, err error) *ErrorInfo {
 
 // queuedEmitter emits confirmation events triggered by ACK or NACK messages.
 type pendingEmitter struct {
-	queue  []serialCh
+	queue  []msgCh
 	logger *LoggerOptions
 }
 
@@ -113,9 +113,9 @@ func newPendingEmitter(log *LoggerOptions) pendingEmitter {
 	}
 }
 
-type serialCh struct {
-	serial int64
-	ch     chan<- error
+type msgCh struct {
+	msg *proto.ProtocolMessage
+	ch  chan<- error
 }
 
 func (q pendingEmitter) Len() int {
@@ -123,77 +123,77 @@ func (q pendingEmitter) Len() int {
 }
 
 func (q pendingEmitter) Less(i, j int) bool {
-	return q.queue[i].serial < q.queue[j].serial
+	return q.queue[i].msg.MsgSerial < q.queue[j].msg.MsgSerial
 }
 
 func (q pendingEmitter) Swap(i, j int) {
 	q.queue[i], q.queue[j] = q.queue[j], q.queue[i]
 }
 
-func (q pendingEmitter) Search(serial int64) int {
-	return sort.Search(q.Len(), func(i int) bool { return q.queue[i].serial >= serial })
+func (q pendingEmitter) Search(msg *proto.ProtocolMessage) int {
+	return sort.Search(q.Len(), func(i int) bool { return q.queue[i].msg.MsgSerial >= msg.MsgSerial })
 }
 
-func (q *pendingEmitter) Enqueue(serial int64, ch chan<- error) {
-	switch i := q.Search(serial); {
+func (q *pendingEmitter) Enqueue(msg *proto.ProtocolMessage, ch chan<- error) {
+	switch i := q.Search(msg); {
 	case i == q.Len():
-		q.queue = append(q.queue, serialCh{serial, ch})
-	case q.queue[i].serial == serial:
-		q.logger.Printf(LogWarning, "duplicated message serial: %d", serial)
+		q.queue = append(q.queue, msgCh{msg, ch})
+	case q.queue[i].msg.MsgSerial == msg.MsgSerial:
+		q.logger.Printf(LogWarning, "duplicated message serial: %d", msg.MsgSerial)
 	default:
-		q.queue = append(q.queue, serialCh{})
+		q.queue = append(q.queue, msgCh{})
 		copy(q.queue[i+1:], q.queue[i:])
-		q.queue[i] = serialCh{serial, ch}
+		q.queue[i] = msgCh{msg, ch}
 	}
 }
 
-func (q *pendingEmitter) Ack(serial int64, count int, errInfo *ErrorInfo) {
+func (q *pendingEmitter) Ack(msg *proto.ProtocolMessage, errInfo *ErrorInfo) {
 	if q.Len() == 0 {
 		return
 	}
 	ack, nack := 0, 0
 	// Ensure range [serial,serial+count] fits inside q.
-	switch i := q.Search(serial); {
+	switch i := q.Search(msg); {
 	case i == q.Len():
 		nack = q.Len()
-	case q.queue[i].serial == serial:
+	case q.queue[i].msg.MsgSerial == msg.MsgSerial:
 		nack = i
-		ack = min(i+count, q.Len())
+		ack = min(i+msg.Count, q.Len())
 	default:
 		nack = i + 1
-		ack = min(i+1+count, q.Len())
+		ack = min(i+1+msg.Count, q.Len())
 	}
 	err := errInfo.unwrapNil()
 	if err == nil && nack > 0 {
 		err = errSerialSkipped
 	}
 	for _, sch := range q.queue[:nack] {
-		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.serial)
+		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.msg.MsgSerial)
 		sch.ch <- err
 	}
 	for _, sch := range q.queue[nack:ack] {
-		q.logger.Printf(LogVerbose, "received ACK for message serial %d", sch.serial)
+		q.logger.Printf(LogVerbose, "received ACK for message serial %d", sch.msg.MsgSerial)
 		sch.ch <- nil
 	}
 	q.queue = q.queue[ack:]
 }
 
-func (q *pendingEmitter) Nack(serial int64, count int, errInfo *ErrorInfo) {
+func (q *pendingEmitter) Nack(msg *proto.ProtocolMessage, errInfo *ErrorInfo) {
 	if q.Len() == 0 {
 		return
 	}
 	nack := 0
-	switch i := q.Search(serial); {
+	switch i := q.Search(msg); {
 	case i == q.Len():
 		nack = q.Len()
-	case q.queue[i].serial == serial:
-		nack = min(i+count, q.Len())
+	case q.queue[i].msg.MsgSerial == msg.MsgSerial:
+		nack = min(i+msg.Count, q.Len())
 	default:
-		nack = min(i+1+count, q.Len())
+		nack = min(i+1+msg.Count, q.Len())
 	}
 	err := errInfo.unwrapNil()
 	for _, sch := range q.queue[:nack] {
-		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.serial)
+		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.msg.MsgSerial)
 		sch.ch <- err
 	}
 	q.queue = q.queue[nack:]
@@ -226,11 +226,7 @@ func (q *msgQueue) Enqueue(msg *proto.ProtocolMessage, listen chan<- error) {
 func (q *msgQueue) Flush() {
 	q.mtx.Lock()
 	for _, msgch := range q.queue {
-		err := q.conn.send(msgch.msg, msgch.ch)
-		if err != nil {
-			q.logger().Printf(LogError, "failure sending message (serial=%d): %v", msgch.msg.MsgSerial, err)
-			msgch.ch <- newError(90000, err)
-		}
+		q.conn.send(msgch.msg, msgch.ch)
 	}
 	q.queue = nil
 	q.mtx.Unlock()


Test patch:
diff --git a/ably/ably_test.go b/ably/ably_test.go
index f054b81..8ea6159 100644
--- a/ably/ably_test.go
+++ b/ably/ably_test.go
@@ -5,9 +5,11 @@ import (
 	"io"
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/ably/ably-go/ably"
 	"github.com/ably/ably-go/ably/ablytest"
+	"github.com/ably/ably-go/ably/proto"
 )
 
 func nonil(err ...error) error {
@@ -87,3 +89,21 @@ type messages chan *ably.Message
 func (ms messages) Receive(m *ably.Message) {
 	ms <- m
 }
+
+type connMock struct {
+	SendFunc    func(*proto.ProtocolMessage) error
+	ReceiveFunc func(deadline time.Time) (*proto.ProtocolMessage, error)
+	CloseFunc   func() error
+}
+
+func (r connMock) Send(a0 *proto.ProtocolMessage) error {
+	return r.SendFunc(a0)
+}
+
+func (r connMock) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	return r.ReceiveFunc(deadline)
+}
+
+func (r connMock) Close() error {
+	return r.CloseFunc()
+}
diff --git a/ably/ablytest/timeout.go b/ably/ablytest/timeout.go
index 2c8e9eb..4bfa459 100644
--- a/ably/ablytest/timeout.go
+++ b/ably/ablytest/timeout.go
@@ -28,7 +28,7 @@ type WithTimeout struct {
 //
 // It returns the second, boolean value returned by the receive operation,
 // or false if the operation times out.
-func (wt WithTimeout) Recv(t *testing.T, into interface{}, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
+func (wt WithTimeout) Recv(t *testing.T, into, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
 	t.Helper()
 	ok, timeout := wt.recv(into, from)
 	if timeout {
@@ -38,7 +38,7 @@ func (wt WithTimeout) Recv(t *testing.T, into interface{}, from interface{}, fai
 }
 
 // NoRecv is like Recv, except it asserts no value is received.
-func (wt WithTimeout) NoRecv(t *testing.T, into interface{}, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
+func (wt WithTimeout) NoRecv(t *testing.T, into, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
 	t.Helper()
 	if into == nil {
 		into = &into
@@ -50,7 +50,16 @@ func (wt WithTimeout) NoRecv(t *testing.T, into interface{}, from interface{}, f
 	return ok
 }
 
-func (wt WithTimeout) recv(into interface{}, from interface{}) (ok, timeout bool) {
+// Send is like Recv, except it sends.
+func (wt WithTimeout) Send(t *testing.T, ch, v interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
+	t.Helper()
+	if timeout := wt.send(ch, v); timeout {
+		fail("timed out waiting for channel send")
+	}
+	return ok
+}
+
+func (wt WithTimeout) recv(into, from interface{}) (ok, timeout bool) {
 	chosen, recv, ok := reflect.Select([]reflect.SelectCase{{
 		Dir:  reflect.SelectRecv,
 		Chan: reflect.ValueOf(from),
@@ -64,6 +73,18 @@ func (wt WithTimeout) recv(into interface{}, from interface{}) (ok, timeout bool
 	return ok, chosen == 1
 }
 
+func (wt WithTimeout) send(ch, v interface{}) (timeout bool) {
+	chosen, _, _ := reflect.Select([]reflect.SelectCase{{
+		Dir:  reflect.SelectSend,
+		Chan: reflect.ValueOf(ch),
+		Send: reflect.ValueOf(v),
+	}, {
+		Dir:  reflect.SelectRecv,
+		Chan: reflect.ValueOf(time.After(time.Duration(wt.before))),
+	}})
+	return chosen == 1
+}
+
 func (wt WithTimeout) IsTrue(pred func() bool) bool {
 	ticker := time.NewTicker(10 * time.Millisecond)
 	defer ticker.Stop()
diff --git a/ably/export_test.go b/ably/export_test.go
index 028bbd5..e77308a 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -124,3 +124,10 @@ const ConnectionStateTTLErrFmt = connectionStateTTLErrFmt
 func DefaultFallbackHosts() []string {
 	return defaultFallbackHosts()
 }
+
+// PendingItems returns the number of messages waiting for Ack/Nack
+func (c *Connection) PendingItems() int {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.pending.Len()
+}
diff --git a/ably/realtime_channel_spec_test.go b/ably/realtime_channel_spec_test.go
index be5ebf1..a7efa66 100644
--- a/ably/realtime_channel_spec_test.go
+++ b/ably/realtime_channel_spec_test.go
@@ -452,7 +452,7 @@ func TestRealtimeChannel_RTL6c5_NoImplicitAttach(t *testing.T) {
 	t.Parallel()
 
 	app, c := ablytest.NewRealtime()
-	defer safeclose(t, app, ablytest.FullRealtimeCloser(c))
+	defer safeclose(t, ablytest.FullRealtimeCloser(c), app)
 
 	if err := ablytest.Wait(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil); err != nil {
 		t.Fatal(err)
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index b376741..3f32257 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -149,5 +149,5 @@ func TestRealtime_multiple(t *testing.T) {
 func TestRealtime_DontCrashOnCloseWhenEchoOff(t *testing.T) {
 	t.Parallel()
 	app, client := ablytest.NewRealtime(ably.WithAutoConnect(false))
-	defer safeclose(t, app, ablytest.FullRealtimeCloser(client))
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 }
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index e5c324b..0fb9efa 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -2144,3 +2144,216 @@ func TestRealtimeConn_RTN2g(t *testing.T) {
 		t.Errorf("expected %q got %q", proto.LibraryString, lib)
 	}
 }
+
+func TestRealtimeConn_RTN19b(t *testing.T) {
+	t.Parallel()
+	connIDs := make(chan string)
+	var breakConn func()
+	var out, in chan *proto.ProtocolMessage
+	c, err := ably.NewRealtime(
+		ably.WithAutoConnect(false),
+		ably.WithKey("fake:key"),
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+			in = make(chan *proto.ProtocolMessage, 1)
+			in <- &proto.ProtocolMessage{
+				Action:       proto.ActionConnected,
+				ConnectionID: <-connIDs,
+				ConnectionDetails: &proto.ConnectionDetails{
+					ConnectionKey: "key",
+				},
+			}
+			out = make(chan *proto.ProtocolMessage, 16)
+			breakConn = func() { close(in) }
+			return ablytest.MessagePipe(in, out)(protocol, u, timeout)
+		}),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	changes := make(ably.ConnStateChanges, 2)
+	off := c.Connection.On(ably.ConnectionEventConnected, changes.Receive)
+	defer off()
+	c.Connect()
+	connIDs <- "1"
+	ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	attaching := c.Channels.Get("attaching")
+	_ = ablytest.ResultFunc.Go(func(ctx context.Context) error { return attaching.Attach(ctx) })
+
+	attachChanges := make(ably.ChannelStateChanges, 1)
+	off = attaching.OnAll(attachChanges.Receive)
+	defer off()
+	ablytest.Soon.Recv(t, nil, attachChanges, t.Fatalf)
+
+	detaching := c.Channels.Get("detaching")
+	detachChange := make(ably.ChannelStateChanges, 1)
+	off = detaching.OnAll(detachChange.Receive)
+	defer off()
+
+	wait := ablytest.ResultFunc.Go(func(ctx context.Context) error { return detaching.Attach(ctx) })
+	var state ably.ChannelStateChange
+	ablytest.Soon.Recv(t, &state, detachChange, t.Fatalf)
+	if expect, got := ably.ChannelStateAttaching, state.Current; expect != got {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	in <- &proto.ProtocolMessage{
+		Action:  proto.ActionAttached,
+		Channel: "detaching",
+	}
+	ablytest.Wait(wait, nil)
+	state = ably.ChannelStateChange{}
+	ablytest.Soon.Recv(t, &state, detachChange, t.Fatalf)
+	if expect, got := ably.ChannelStateAttached, state.Current; expect != got {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+
+	_ = ablytest.ResultFunc.Go(func(ctx context.Context) error { return detaching.Detach(ctx) })
+	state = ably.ChannelStateChange{}
+	ablytest.Soon.Recv(t, &state, detachChange, t.Fatalf)
+	if expect, got := ably.ChannelStateDetaching, state.Current; expect != got {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+
+	msgs := []proto.ProtocolMessage{
+		{
+			Channel: "attaching",
+			Action:  proto.ActionAttach,
+		},
+		{
+			Channel: "detaching",
+			Action:  proto.ActionAttach,
+		},
+		{
+			Channel: "detaching",
+			Action:  proto.ActionDetach,
+		},
+	}
+	for _, expect := range msgs {
+		var got *proto.ProtocolMessage
+		ablytest.Instantly.Recv(t, &got, out, t.Fatalf)
+		if expect.Action != got.Action {
+			t.Errorf("expected %v got %v", expect.Action, got.Action)
+		}
+		if expect.Channel != got.Channel {
+			t.Errorf("expected %v got %v", expect.Channel, got.Channel)
+		}
+	}
+	breakConn()
+	connIDs <- "2"
+	ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	if c.Connection.ID() != "2" {
+		t.Fatal("expected new connection")
+	}
+	msgs = []proto.ProtocolMessage{
+		{
+			Channel: "attaching",
+			Action:  proto.ActionAttach,
+		},
+		{
+			Channel: "detaching",
+			Action:  proto.ActionDetach,
+		},
+	}
+	for _, expect := range msgs {
+		var got *proto.ProtocolMessage
+		ablytest.Instantly.Recv(t, &got, out, t.Fatalf)
+		if expect.Action != got.Action {
+			t.Errorf("expected %v got %v", expect.Action, got.Action)
+		}
+		if expect.Channel != got.Channel {
+			t.Errorf("expected %v got %v", expect.Channel, got.Channel)
+		}
+	}
+}
+
+func TestRealtimeConn_RTN19a(t *testing.T) {
+	t.Parallel()
+	connIDs := make(chan string)
+	var breakConn func()
+	var out, in chan *proto.ProtocolMessage
+	c, err := ably.NewRealtime(
+		ably.WithAutoConnect(false),
+		ably.WithKey("fake:key"),
+		ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (proto.Conn, error) {
+			in = make(chan *proto.ProtocolMessage, 1)
+			in <- &proto.ProtocolMessage{
+				Action:       proto.ActionConnected,
+				ConnectionID: <-connIDs,
+				ConnectionDetails: &proto.ConnectionDetails{
+					ConnectionKey: "key",
+				},
+			}
+			out = make(chan *proto.ProtocolMessage, 16)
+			breakConn = func() { close(in) }
+			return ablytest.MessagePipe(in, out)(protocol, u, timeout)
+		}),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	changes := make(ably.ConnStateChanges, 2)
+	off := c.Connection.On(ably.ConnectionEventConnected, changes.Receive)
+	defer off()
+	c.Connect()
+	connIDs <- "1"
+	ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+
+	name := "channel"
+	channel := c.Channels.Get(name)
+	chanChange := make(ably.ChannelStateChanges, 1)
+	off = channel.OnAll(chanChange.Receive)
+	defer off()
+
+	wait := ablytest.ResultFunc.Go(func(ctx context.Context) error { return channel.Attach(ctx) })
+	var state ably.ChannelStateChange
+	ablytest.Soon.Recv(t, &state, chanChange, t.Fatalf)
+	if expect, got := ably.ChannelStateAttaching, state.Current; got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	in <- &proto.ProtocolMessage{
+		Action:  proto.ActionAttached,
+		Channel: name,
+	}
+	ablytest.Wait(wait, nil)
+	state = ably.ChannelStateChange{}
+	ablytest.Soon.Recv(t, &state, chanChange, t.Fatalf)
+	if expect, got := ably.ChannelStateAttached, state.Current; got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	ctx, _ := context.WithTimeout(context.TODO(), time.Millisecond)
+	err = channel.Publish(ctx, "ack", "ack")
+	if err != nil {
+		if err != context.DeadlineExceeded {
+			t.Fatal(err)
+		}
+	}
+	ablytest.Soon.Recv(t, nil, out, t.Fatalf) // attach
+
+	var msg *proto.ProtocolMessage
+	ablytest.Soon.Recv(t, &msg, out, t.Fatalf)
+	if expect, got := proto.ActionMessage, msg.Action; got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+
+	if expect, got := 1, c.Connection.PendingItems(); got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	breakConn()
+	connIDs <- "2"
+	ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	if expect, got := "2", c.Connection.ID(); got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+
+	ablytest.Instantly.Recv(t, nil, out, t.Fatalf) // attach
+	msg = nil
+	ablytest.Instantly.Recv(t, &msg, out, t.Fatalf)
+	if expect, got := proto.ActionMessage, msg.Action; got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	if expect, got := name, msg.Channel; got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+	if expect, got := 1, c.Connection.PendingItems(); got != expect {
+		t.Fatalf("expected %v got %v", expect, got)
+	}
+}
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index b45666a..8d21449 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -1,11 +1,15 @@
 package ably_test
 
 import (
+	"context"
+	"errors"
+	"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"
 )
 
@@ -201,7 +205,7 @@ func TestRealtimeConn_BreakConnLoopOnInactiveState(t *testing.T) {
 			app, client := ablytest.NewRealtime(
 				ably.WithDial(ablytest.MessagePipe(in, out)),
 			)
-			defer safeclose(t, app, ablytest.FullRealtimeCloser(client))
+			defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 			connected := &proto.ProtocolMessage{
 				Action:            proto.ActionConnected,
@@ -230,3 +234,67 @@ func TestRealtimeConn_BreakConnLoopOnInactiveState(t *testing.T) {
 		})
 	}
 }
+
+func TestRealtimeConn_SendErrorReconnects(t *testing.T) {
+	sendErr := make(chan error, 1)
+	closed := make(chan struct{}, 1)
+	allowDial := make(chan struct{})
+
+	dial := ablytest.DialFunc(func(p string, url *url.URL, timeout time.Duration) (proto.Conn, error) {
+		<-allowDial
+		ws, err := ablyutil.DialWebsocket(p, url, timeout)
+		if err != nil {
+			return nil, err
+		}
+		return connMock{
+			SendFunc: func(m *proto.ProtocolMessage) error {
+				select {
+				case err := <-sendErr:
+					return err
+				default:
+					return ws.Send(m)
+				}
+			},
+			ReceiveFunc: ws.Receive,
+			CloseFunc: func() error {
+				closed <- struct{}{}
+				return ws.Close()
+			},
+		}, nil
+	})
+
+	app, c := ablytest.NewRealtime(ably.WithDial(dial))
+	defer safeclose(t, ablytest.FullRealtimeCloser(c), app)
+
+	allowDial <- struct{}{}
+
+	if err := ablytest.Wait(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil); err != nil {
+		t.Fatal(err)
+	}
+
+	// Cause a send error; expect message to be enqueued and transport to be
+	// closed.
+
+	sendErr <- errors.New("fail")
+
+	publishErr := make(chan error, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		e := c.Channels.Get("test").Publish(ctx, "test", nil)
+		publishErr <- e
+	}()
+
+	ablytest.Instantly.Recv(t, nil, closed, t.Fatalf)
+	ablytest.Instantly.NoRecv(t, nil, publishErr, t.Fatalf)
+
+	// Reconnect should happen instantly as a result of transport closure.
+	ablytest.Instantly.Send(t, allowDial, struct{}{}, t.Fatalf)
+
+	// After reconnection, message should be published.
+	var err error
+	ablytest.Soon.Recv(t, &err, publishErr, t.Fatalf)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
diff --git a/ably/state_test.go b/ably/state_test.go
index 59d6b1d..47d3f11 100644
--- a/ably/state_test.go
+++ b/ably/state_test.go
@@ -3,6 +3,8 @@ package ably
 import (
 	"errors"
 	"testing"
+
+	"github.com/ably/ably-go/ably/proto"
 )
 
 var errNotEmitted = errors.New("not emitted")
@@ -29,9 +31,9 @@ func receive(ch ...chan error) []error {
 	return errs
 }
 
-func emit(serial int64, count int, fn func(*pendingEmitter, int64, int, *ErrorInfo), err *ErrorInfo) func(*pendingEmitter) {
+func emit(msg *proto.ProtocolMessage, fn func(*pendingEmitter, *proto.ProtocolMessage, *ErrorInfo), err *ErrorInfo) func(*pendingEmitter) {
 	return func(q *pendingEmitter) {
-		fn(q, serial, count, err)
+		fn(q, msg, err)
 	}
 }
 
@@ -43,7 +45,7 @@ func testQueuedEmitter(t *testing.T, serials, ack, nack []int64, emit func(*pend
 	}
 	q := &pendingEmitter{logger: &LoggerOptions{}}
 	for serial, i := range index {
-		q.Enqueue(serial, ch[i])
+		q.Enqueue(&proto.ProtocolMessage{MsgSerial: serial}, ch[i])
 	}
 	emit(q)
 	errs := receive(ch...)
@@ -70,40 +72,43 @@ func testQueuedEmitter(t *testing.T, serials, ack, nack []int64, emit func(*pend
 }
 
 func TestQueuedEmitter(t *testing.T) {
+	msg := func(serial int64, count int) *proto.ProtocolMessage {
+		return &proto.ProtocolMessage{MsgSerial: serial, Count: count}
+	}
 	cases := [...]struct {
 		serial, ack, nack []int64
 		emit              func(*pendingEmitter)
 	}{{ // 5 pending messages, ack first two
 		serial: []int64{1, 2, 3, 4, 5},
 		ack:    []int64{1, 2},
-		emit:   emit(1, 2, (*pendingEmitter).Ack, nil),
+		emit:   emit(msg(1, 2), (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, ack for second and third, first should got nack
 		serial: []int64{1, 2, 3, 4, 5},
 		ack:    []int64{2, 3},
 		nack:   []int64{1},
-		emit:   emit(2, 2, (*pendingEmitter).Ack, nil),
+		emit:   emit(msg(2, 2), (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, 1 and 2 get nack, rest ack
 		serial: []int64{5, 2, 4, 1, 3},
 		ack:    []int64{3, 4, 5},
 		nack:   []int64{1, 2},
-		emit:   emit(3, 3, (*pendingEmitter).Ack, nil),
+		emit:   emit(msg(3, 3), (*pendingEmitter).Ack, nil),
 	}, { // 10 pending messages, first 5 get nacks, next 4 ack
 		serial: []int64{9, 4, 2, 6, 10, 3, 1, 7, 5, 8},
 		ack:    []int64{6, 7, 8, 9},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(6, 4, (*pendingEmitter).Ack, nil),
+		emit:   emit(msg(6, 4), (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, nack first 3
 		serial: []int64{3, 5, 1, 4, 2},
 		nack:   []int64{1, 2, 3},
-		emit:   emit(1, 3, (*pendingEmitter).Nack, errNack),
+		emit:   emit(msg(1, 3), (*pendingEmitter).Nack, errNack),
 	}, { // 5 pending messages, nack all
 		serial: []int64{2, 1, 3, 5, 4},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(1, 5, (*pendingEmitter).Nack, errNack),
+		emit:   emit(msg(1, 5), (*pendingEmitter).Nack, errNack),
 	}, { // 5 pending messages, nack all
 		serial: []int64{4, 5, 2, 3, 1},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(2, 10, (*pendingEmitter).Nack, errNack),
+		emit:   emit(msg(2, 10), (*pendingEmitter).Nack, errNack),
 	}}
 	for _, cas := range cases {
 		testQueuedEmitter(t, cas.serial, cas.ack, cas.nack, cas.emit)