Repository: ably/ably-java
Issue: Lib is not re-sending pending messages on new transport after a resume

Log posted by a customer [here](https://ably-real-time.slack.com/archives/C030C5YLY/p1558623828295800) (cf [conversation](https://app.intercom.io/a/apps/ua39m1ld/inbox/inbox/all/conversations/22156922233)) shows several issues:

1. At one point it disconnects the transport for inactivity at 14:29:08.086Z, 16s after getting a detached from realtime at 14:28:52.151. That shouldn't happen.
1. After it reconnects and resumes, it does not seem to be re-sending messages on the new transport that were pending on the previous transport, as required by RTN15f
1. Related to the previous one: messages called back with a failure due to getting an ack for a higher msgSerial than the pending queue are called back with "Unknown error / 50000 / 500", which could be more informative



┆Issue is synchronized with this [Jira Bug](https://ably.atlassian.net/browse/SDK-430) by [Unito](https://www.unito.io)


Changed files:
lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java, lib/src/test/java/io/ably/lib/test/util/MockWebsocketFactory.java, lib/src/main/java/io/ably/lib/realtime/ChannelBase.java, lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java, lib/src/main/java/io/ably/lib/transport/ITransport.java, lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java, lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

diff --git a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
index fc441118..4a1f9b02 100644
--- a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
+++ b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
@@ -231,6 +231,20 @@ public class AblyRealtime extends AblyRest {
             }
         }
 
+        /**
+         * By spec RTN15c3
+         */
+        @Override
+        public void reattachOnResumeFailure() {
+            for (Map.Entry<String, Channel> channelEntry : map.entrySet()) {
+                Channel channel = channelEntry.getValue();
+                if (channel.state == ChannelState.attaching || channel.state == ChannelState.attached || channel.state == ChannelState.suspended) {
+                    Log.d(TAG, "reAttach(); channel = " + channel.name);
+                    channel.attach(true, null);
+                }
+            }
+        }
+
         private void clear() {
             map.clear();
         }
diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
index c36cfabf..9c26cb3b 100644
--- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
+++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
@@ -181,7 +181,7 @@ public abstract class ChannelBase extends EventEmitter<ChannelEvent, ChannelStat
         this.attach(false, listener);
     }
 
-    private void attach(boolean forceReattach, CompletionListener listener) {
+    void attach(boolean forceReattach, CompletionListener listener) {
         clearAttachTimers();
         attachWithTimeout(forceReattach, listener);
     }
diff --git a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
index a3e171a2..9286e280 100644
--- a/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
+++ b/lib/src/main/java/io/ably/lib/transport/ConnectionManager.java
@@ -81,6 +81,8 @@ public class ConnectionManager implements ConnectListener {
         void onMessage(ProtocolMessage msg);
         void suspendAll(ErrorInfo error, boolean notifyStateChange);
         Iterable<Channel> values();
+
+        void reattachOnResumeFailure();
     }
 
     /***********************************
@@ -1183,44 +1185,48 @@ public class ConnectionManager implements ConnectListener {
     }
 
     private synchronized void onConnected(ProtocolMessage message) {
-        /* if the returned connection id differs from
-         * the existing connection id, then this means
-         * we need to suspend all existing attachments to
-         * the old connection.
-         * If realtime did not reply with an error, it
-         * signifies that this was a result of an earlier
-         * connection being invalidated due to being stale.
-         *
-         * Suspend all channels attached to the previous id;
-         * this will be reattached in setConnection() */
-        ErrorInfo error = message.error;
-        if(connection.id != null && !message.connectionId.equals(connection.id)) {
-            /* we need to suspend the original connection */
-            if(error == null) {
-                error = REASON_SUSPENDED;
+        final ErrorInfo error = message.error;
+        connection.reason = error;
+        if (connection.id != null) { // there was a previous connection, so this is a resume and RTN15c applies
+            Log.d(TAG, "There was a connection resume");
+            if(message.connectionId.equals(connection.id)) {
+                // resume succeeded
+                if(message.error == null) {
+                    // RTN15c1: no action required wrt channel state
+                    Log.d(TAG, "connection has reconnected and resumed successfully");
+                } else {
+                    // RTN15c2: no action required wrt channel state
+                    Log.d(TAG, "connection resume success with non-fatal error: " + error.message);
+                }
+                // Add pending messages to the front of queued messages to be sent later
+                addPendingMessagesToQueuedMessages(false);
+            } else {
+                // RTN15c3: resume failed
+                if (error != null){
+                    Log.d(TAG, "connection resume failed with error: " + error.message);
+                }else { // This shouldn't happen but, putting it here for safety
+                    Log.d(TAG, "connection resume failed without error" );
+                }
+
+                channels.reattachOnResumeFailure();
+                // Add any messages still pending from the previous transport (RTN19a) to the front of queued messages
+                // however, this time the pending messages have to have newly assigned `
+                // msgSerial`s. They can't simply be replayed, as they are in the successful resume case
+                addPendingMessagesToQueuedMessages(true);
             }
-            channels.suspendAll(error, false);
         }
 
-        /* set the new connection id */
-        ConnectionDetails connectionDetails = message.connectionDetails;
-        connection.key = connectionDetails.connectionKey;
-        if (!message.connectionId.equals(connection.id)) {
-            /* The connection id has changed. Reset the message serial and the
-             * pending message queue (which fails the messages currently in
-             * there). */
-            pendingMessages.reset(msgSerial,
-                    new ErrorInfo("Connection resume failed", 500, 50000));
-            msgSerial = 0;
-        }
         connection.id = message.connectionId;
+
         if(message.connectionSerial != null) {
-            connection.serial = message.connectionSerial.longValue();
+            connection.serial = message.connectionSerial;
             if (connection.key != null)
                 connection.recoveryKey = connection.key + ":" + message.connectionSerial;
         }
 
+        ConnectionDetails connectionDetails = message.connectionDetails;
         /* Get any parameters from connectionDetails. */
+        connection.key = connectionDetails.connectionKey; //RTN16d
         maxIdleInterval = connectionDetails.maxIdleInterval;
         connectionStateTtl = connectionDetails.connectionStateTtl;
 
@@ -1232,12 +1238,39 @@ public class ConnectionManager implements ConnectListener {
             requestState(transport, new StateIndication(ConnectionState.failed, e.errorInfo));
             return;
         }
-
         /* indicated connected currentState */
         setSuspendTime();
         requestState(new StateIndication(ConnectionState.connected, error));
     }
 
+    /**
+     * Add all pending queued messages to the front of QueuedMessages for them to be sent later
+     * Spec: RTN19a
+     * @param resetMessageSerial whether to reset message serial, this will determine whether to reset message serials
+     * on pending queue, for example when a connection resume failed
+     */
+    private void addPendingMessagesToQueuedMessages(boolean resetMessageSerial) {
+        // Add messages from pending messages to front of queuedMessages in order to retry them
+        queuedMessages.addAll(0, pendingMessages.queue);
+        //rewind start serial back to the first serial since we are clearing the queue
+        if (!pendingMessages.queue.isEmpty()){
+            //Reset current serial to the first pending message on previous queue as we are going to clear the queue now
+            msgSerial =  pendingMessages.queue.get(0).msg.msgSerial;
+            pendingMessages.resetStartSerial((int) (msgSerial));
+            pendingMessages.clearQueue();
+        }
+
+        //RTN19a
+        if (resetMessageSerial){
+            pendingMessages.resetStartSerial(0);
+            msgSerial = 0; //msgSerial will increase in sendImpl when messages are sent
+        }
+    }
+
+    public List<QueuedMessage> getPendingMessages() {
+        return pendingMessages.queue;
+    }
+
     private synchronized void onDisconnected(ProtocolMessage message) {
         ErrorInfo reason = message.error;
         if(reason != null && isTokenError(reason)) {
@@ -1718,6 +1751,13 @@ public class ConnectionManager implements ConnectListener {
             startSerial = 0;
         }
 
+        public void resetStartSerial(int from) {
+             startSerial = from;
+        }
+
+        synchronized void clearQueue() {
+            queue.clear();
+        }
     }
 
     /***********************
diff --git a/lib/src/main/java/io/ably/lib/transport/ITransport.java b/lib/src/main/java/io/ably/lib/transport/ITransport.java
index 364c03ab..93b426f3 100644
--- a/lib/src/main/java/io/ably/lib/transport/ITransport.java
+++ b/lib/src/main/java/io/ably/lib/transport/ITransport.java
@@ -122,6 +122,8 @@ public interface ITransport {
      */
     void send(ProtocolMessage msg) throws AblyException;
 
+    void receive(ProtocolMessage msg) throws AblyException;
+
     /**
      * Get connection URL
      * @return
diff --git a/lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java b/lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java
index 55b23d68..85284b17 100644
--- a/lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java
+++ b/lib/src/main/java/io/ably/lib/transport/WebSocketTransport.java
@@ -70,7 +70,7 @@ public class WebSocketTransport implements ITransport {
 
             Log.d(TAG, "connect(); wsUri = " + wsUri);
             synchronized(this) {
-                wsConnection = new WsClient(URI.create(wsUri));
+                wsConnection = new WsClient(URI.create(wsUri), this::receive);
                 if(isTls) {
                     SSLContext sslContext = SSLContext.getInstance("TLS");
                     sslContext.init( null, null, null );
@@ -99,6 +99,11 @@ public class WebSocketTransport implements ITransport {
         }
     }
 
+    @Override
+    public void receive(ProtocolMessage msg) throws AblyException {
+        connectionManager.onMessage(this, msg);
+    }
+
     @Override
     public void send(ProtocolMessage msg) throws AblyException {
         Log.d(TAG, "send(); action = " + msg.action);
@@ -137,14 +142,21 @@ public class WebSocketTransport implements ITransport {
         //Gives the chance to child classes to do message pre-processing
     }
 
+    //interface to transfer Protocol message from websocket
+    interface WebSocketReceiver {
+        void onMessage(ProtocolMessage protocolMessage) throws AblyException;
+    }
+
     /**************************
      * WebSocketHandler methods
      **************************/
 
-    class WsClient extends WebSocketClient {
+     class WsClient extends WebSocketClient {
+       private final WebSocketReceiver receiver;
 
-        WsClient(URI serverUri) {
+        WsClient(URI serverUri, WebSocketReceiver receiver) {
             super(serverUri);
+            this.receiver = receiver;
         }
 
         @Override
@@ -180,7 +192,7 @@ public class WebSocketTransport implements ITransport {
                 ProtocolMessage msg = ProtocolSerializer.readMsgpack(blob.array());
                 Log.d(TAG, "onMessage(): msg (binary) = " + msg);
                 WebSocketTransport.this.preProcessReceivedMessage(msg);
-                connectionManager.onMessage(WebSocketTransport.this, msg);
+                receiver.onMessage(msg);
             } catch (AblyException e) {
                 String msg = "Unexpected exception processing received binary message";
                 Log.e(TAG, msg, e);
@@ -194,7 +206,7 @@ public class WebSocketTransport implements ITransport {
                 ProtocolMessage msg = ProtocolSerializer.fromJSON(string);
                 Log.d(TAG, "onMessage(): msg (text) = " + msg);
                 WebSocketTransport.this.preProcessReceivedMessage(msg);
-                connectionManager.onMessage(WebSocketTransport.this, msg);
+                receiver.onMessage(msg);
             } catch (AblyException e) {
                 String msg = "Unexpected exception processing received text message";
                 Log.e(TAG, msg, e);
diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java
index 3277c902..eae88550 100644
--- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java
+++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeResumeTest.java
@@ -4,12 +4,15 @@ import io.ably.lib.debug.DebugOptions;
 import io.ably.lib.realtime.AblyRealtime;
 import io.ably.lib.realtime.Channel;
 import io.ably.lib.realtime.ChannelState;
+import io.ably.lib.realtime.ConnectionEvent;
 import io.ably.lib.realtime.ConnectionState;
+import io.ably.lib.realtime.ConnectionStateListener;
 import io.ably.lib.test.common.Helpers.ChannelWaiter;
 import io.ably.lib.test.common.Helpers.CompletionSet;
 import io.ably.lib.test.common.Helpers.ConnectionWaiter;
 import io.ably.lib.test.common.Helpers.MessageWaiter;
 import io.ably.lib.test.common.ParameterizedTest;
+import io.ably.lib.test.util.MockWebsocketFactory;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ErrorInfo;
@@ -20,11 +23,14 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
 
+import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.List;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -82,7 +88,7 @@ public class RealtimeResumeTest extends ParameterizedTest {
 
         } catch (AblyException e) {
             e.printStackTrace();
-            fail("init0: Unexpected exception instantiating library");
+            fail("Unexpected exception: "+e.getMessage());
         } finally {
             if(ably != null) {
                 ably.close();
@@ -572,7 +578,6 @@ public class RealtimeResumeTest extends ParameterizedTest {
      * round of messages which should be queued and published after
      * we reconnect the sender.
      */
-    @Ignore("FIXME: fix exception")
     @Test
     public void resume_publish_queue() {
         AblyRealtime receiver = null;
@@ -684,6 +689,243 @@ public class RealtimeResumeTest extends ParameterizedTest {
         }
     }
 
+    /**
+     * In case of resume success, verify that pending messages are resent. By blocking ack/nacks before sending the
+     * message while connected and then disconnect, add some more messages
+     * */
+    @Test
+    public void resume_publish_resend_pending_messages_when_resume_is_successful() {
+        final long delay = 200;
+        final String channelName = "resume_publish_queue";
+        AblyRealtime sender = null;
+        try {
+            final MockWebsocketFactory mockWebsocketFactory = new MockWebsocketFactory();
+            String keyStr = testVars.keys[0].keyStr;
+            DebugOptions senderOptions = createOptions(keyStr);
+            senderOptions.queueMessages = true;
+            senderOptions.transportFactory = mockWebsocketFactory;
+            sender = new AblyRealtime(senderOptions);
+            final Channel senderChannel = sender.channels.get(channelName);
+            senderChannel.attach();
+            (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached);
+            assertEquals(
+                "The sender's channel should be attached",
+                senderChannel.state, ChannelState.attached
+            );
+
+            MockWebsocketFactory.MockWebsocketTransport transport = mockWebsocketFactory.getCreatedTransport();
+
+            final CompletionSet senderCompletion = new CompletionSet();
+            //send 3 successful messages
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("non_pending messages" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+
+            /* wait for the publish callback to be called.*/
+            ErrorInfo[] errors = senderCompletion.waitFor();
+            assertTrue(
+                "First completion has errors",
+                errors.length == 0
+            );
+
+            //assert that messages sent till now are sent with correct size and serials
+            assertEquals("First round of messages has incorrect size", 3, transport.getPublishedMessages().size());
+
+            for (int i = 0; i < transport.getPublishedMessages().size(); i++) {
+                ProtocolMessage protocolMessage = transport.getPublishedMessages().get(i);
+                assertEquals("Sent serial incorrect", Long.valueOf(i), protocolMessage.msgSerial);
+            }
+
+            //now clear published messages
+            transport.clearPublishedMessages();
+
+            //block ack/nack messages to simulate pending message
+            mockWebsocketFactory.blockReceive(message -> message.action == ProtocolMessage.Action.ack ||
+                message.action == ProtocolMessage.Action.nack);
+
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("pending_queued_message_" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+            assertEquals(sender.connection.connectionManager.getPendingMessages().size(),3);
+
+            final String connectionId = sender.connection.id;
+
+            //now let's disconnect
+            sender.connection.connectionManager.requestState(ConnectionState.disconnected);
+            (new ConnectionWaiter(sender.connection)).waitFor(ConnectionState.disconnected);
+
+            //send 3 more messages while disconnected
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("queued_message_" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+            //now let's unblock the ack nacks and reconnect
+            mockWebsocketFactory.blockReceive(message -> false);
+            /* reconnect the sender */
+            sender.connection.connect();
+            (new ConnectionWaiter(sender.connection)).waitFor(ConnectionState.connected);
+            assertEquals("Connection must be connected", ConnectionState.connected, sender.connection.state);
+            //make sure connection id is a resume success
+            assertEquals("Connection id has changed", connectionId, sender.connection.id);
+
+            //replace mock transport
+            transport = mockWebsocketFactory.getCreatedTransport();
+
+            /* wait for the publish callback to be called.*/
+            ErrorInfo[] senderErrors = senderCompletion.waitFor();
+            assertTrue(
+                "Second round of send has errors",
+                senderErrors.length == 0
+            );
+
+            assertEquals("Second round of messages has incorrect size", 6, transport.getPublishedMessages().size());
+            //make sure they were sent with correct serials
+            for (int i = 0; i < transport.getPublishedMessages().size(); i++) {
+                ProtocolMessage protocolMessage = transport.getPublishedMessages().get(i);
+                assertEquals("Sent serial incorrect", Long.valueOf(i+3), protocolMessage.msgSerial);
+            }
+
+            //make sure that pending queue is cleared
+            assertEquals("There are still pending messages in the queue",
+                sender.connection.connectionManager.getPendingMessages().size(),
+                0);
+
+        } catch (AblyException e) {
+            fail("Unexpected exception: "+e.getMessage());
+        }  finally {
+            if (sender != null) {
+                sender.close();
+            }
+        }
+    }
+
+    /**
+     * In case of resume failure verify that messages are being resent
+     * */
+    @Test
+    public void resume_publish_resend_pending_messages_when_resume_failed() throws AblyException {
+        final long delay = 200;
+        final String channelName = "sender_channel";
+        final MockWebsocketFactory mockWebsocketFactory = new MockWebsocketFactory();
+        final DebugOptions options = createOptions(testVars.keys[0].keyStr);
+        options.realtimeRequestTimeout = 2000L;
+        options.transportFactory = mockWebsocketFactory;
+        try(AblyRealtime ably = new AblyRealtime(options)) {
+            final long newTtl = 1000L;
+            final long newIdleInterval = 1000L;
+            /* We want this greater than newTtl + newIdleInterval */
+            final long waitInDisconnectedState = 3000L;
+
+            ably.connection.on(ConnectionEvent.connected, new ConnectionStateListener() {
+                @Override
+                public void onConnectionStateChanged(ConnectionStateChange state) {
+                    try {
+                        Field connectionStateField = ably.connection.connectionManager.getClass().getDeclaredField("connectionStateTtl");
+                        connectionStateField.setAccessible(true);
+                        connectionStateField.setLong(ably.connection.connectionManager, newTtl);
+                        Field maxIdleField = ably.connection.connectionManager.getClass().getDeclaredField("maxIdleInterval");
+                        maxIdleField.setAccessible(true);
+                        maxIdleField.setLong(ably.connection.connectionManager, newIdleInterval);
+                    } catch (NoSuchFieldException | IllegalAccessException e) {
+                        fail("Unexpected exception in checking connectionStateTtl");
+                    }
+                }
+            });
+
+            ConnectionWaiter connectionWaiter = new ConnectionWaiter(ably.connection);
+            connectionWaiter.waitFor(ConnectionState.connected);
+
+            final Channel senderChannel = ably.channels.get(channelName);
+            senderChannel.attach();
+            (new ChannelWaiter(senderChannel)).waitFor(ChannelState.attached);
+            assertEquals(
+                "The sender's channel should be attached",
+                senderChannel.state, ChannelState.attached
+            );
+
+            MockWebsocketFactory.MockWebsocketTransport transport = mockWebsocketFactory.getCreatedTransport();
+            CompletionSet senderCompletion = new CompletionSet();
+            //send 3 successful messages
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("non_pending messages" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+
+            /* wait for the publish callback to be called.*/
+            ErrorInfo[] errors = senderCompletion.waitFor();
+            assertTrue(
+                "First completion has errors",
+                errors.length == 0
+            );
+
+            //assert that messages sent till now are sent with correct size and serials
+            assertEquals("First round of messages has incorrect size", 3, transport.getPublishedMessages().size());
+            for (int i = 0; i < transport.getPublishedMessages().size(); i++) {
+                ProtocolMessage protocolMessage = transport.getPublishedMessages().get(i);
+                assertEquals("Sent serial incorrect", Long.valueOf(i), protocolMessage.msgSerial);
+            }
+
+            //block acks nacks before send
+            mockWebsocketFactory.blockReceive(message -> message.action == ProtocolMessage.Action.ack ||
+                message.action == ProtocolMessage.Action.nack);
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("pending_queued_message_" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+
+            final String firstConnectionId = ably.connection.id;
+
+            /* suppress automatic retries by the connection manager and disconnect */
+            try {
+                Method method = ably.connection.connectionManager.getClass().getDeclaredMethod("disconnectAndSuppressRetries");
+                method.setAccessible(true);
+                method.invoke(ably.connection.connectionManager);
+            } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+                fail("Unexpected exception in suppressing retries");
+            }
+            connectionWaiter.waitFor(ConnectionState.disconnected);
+            assertEquals("Disconnected state was not reached", ConnectionState.disconnected, ably.connection.state);
+
+            //send some more messages while disconnected
+            for (int i = 0; i < 3; i++) {
+                senderChannel.publish("queued_message_" + i, "Test pending queued messages " + i,
+                    senderCompletion.add());
+            }
+            //now let's unblock the ack nacks and reconnect
+            mockWebsocketFactory.blockReceive(message -> false);
+            /* Wait for the connection to go stale, then reconnect */
+            try {
+                Thread.sleep(waitInDisconnectedState);
+            } catch (InterruptedException e) {
+            }
+            ably.connection.connect();
+            connectionWaiter.waitFor(ConnectionState.connected);
+            assertEquals("Connected state was not reached", ConnectionState.connected, ably.connection.state);
+            //replace transport
+            transport = mockWebsocketFactory.getCreatedTransport();
+            /* Verify the connection is new */
+            assertNotNull(ably.connection.id);
+            assertNotEquals("Connection has the same id", firstConnectionId, ably.connection.id);
+
+            /* wait for the publish callback to be called.*/
+
+            ErrorInfo[] resendErrors = senderCompletion.waitFor();
+            assertTrue(
+                "Second round of messages (queued) has errors",
+                resendErrors.length == 0
+            );
+
+            assertEquals("Second round of messages has incorrect size", 6, transport.getPublishedMessages().size());
+            //make sure they were sent with reset serials
+            for (int i = 0; i < transport.getPublishedMessages().size(); i++) {
+                ProtocolMessage protocolMessage = transport.getPublishedMessages().get(i);
+                assertEquals("Sent serial incorrect", Long.valueOf(i), protocolMessage.msgSerial);
+            }
+        }
+    }
+
     //RTL4j2
     @Test
     public void resume_rewind_1 ()
diff --git a/lib/src/test/java/io/ably/lib/test/util/MockWebsocketFactory.java b/lib/src/test/java/io/ably/lib/test/util/MockWebsocketFactory.java
index 3d0b0c44..102b3227 100644
--- a/lib/src/test/java/io/ably/lib/test/util/MockWebsocketFactory.java
+++ b/lib/src/test/java/io/ably/lib/test/util/MockWebsocketFactory.java
@@ -1,5 +1,8 @@
 package io.ably.lib.test.util;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.transport.ITransport;
 import io.ably.lib.transport.WebSocketTransport;
@@ -17,6 +20,13 @@ public class MockWebsocketFactory implements ITransport.Factory {
         block,
         fail
     }
+
+    enum ReceiveBehaviour {
+        allow,
+        block,
+        fail
+    }
+
     enum ConnectBehaviour {
         allow,
         fail
@@ -35,8 +45,10 @@ public class MockWebsocketFactory implements ITransport.Factory {
     }
 
     SendBehaviour sendBehaviour = SendBehaviour.allow;
+    ReceiveBehaviour receiveBehaviour = ReceiveBehaviour.allow;
     ConnectBehaviour connectBehaviour = ConnectBehaviour.allow;
-    MessageFilter messageFilter = null;
+    MessageFilter sendMessageFilter = null;
+    MessageFilter receiveMessageFilter = null;
     HostFilter hostFilter = null;
     HostTransform hostTransform = null;
 
@@ -63,26 +75,38 @@ public class MockWebsocketFactory implements ITransport.Factory {
         return lastCreatedTransport;
     }
 
+    //only use this when you know when transport is created - just for tests
+    public MockWebsocketTransport getCreatedTransport() {
+        return (MockWebsocketTransport) lastCreatedTransport;
+    }
+
     public void blockSend(MessageFilter filter) {
-        messageFilter = filter;
+        sendMessageFilter = filter;
         sendBehaviour = SendBehaviour.block;
     }
+
+    //use the same filters temporarily
+    public void blockReceive(MessageFilter filter) {
+        receiveMessageFilter = filter;
+        receiveBehaviour = ReceiveBehaviour.block;
+    }
+
     public void blockSend() { blockSend(null); }
 
     public void allowSend(MessageFilter filter) {
-        messageFilter = filter;
+        sendMessageFilter = filter;
         sendBehaviour = SendBehaviour.allow;
     }
     public void allowSend() { allowSend(null);}
 
     public void failSend(MessageFilter filter) {
-        messageFilter = filter;
+        sendMessageFilter = filter;
         sendBehaviour = SendBehaviour.fail;
     }
     public void failSend() { failSend(null); }
 
-    public void setMessageFilter(MessageFilter filter) {
-        messageFilter = filter;
+    public void setSendMessageFilter(MessageFilter filter) {
+        sendMessageFilter = filter;
     }
 
     public void failConnect(HostFilter filter) {
@@ -98,9 +122,10 @@ public class MockWebsocketFactory implements ITransport.Factory {
     /*
      * Special transport class that allows blocking send() and other operations
      */
-    private class MockWebsocketTransport extends WebSocketTransport {
+    public class MockWebsocketTransport extends WebSocketTransport {
         private final TransportParams givenTransportParams;
         private final TransportParams transformedTransportParams;
+        private final List<ProtocolMessage> publishedMessages = new ArrayList<>();
 
         private MockWebsocketTransport(TransportParams givenTransportParams, TransportParams transformedTransportParams, ConnectionManager connectionManager) {
             super(transformedTransportParams, connectionManager);
@@ -108,23 +133,34 @@ public class MockWebsocketFactory implements ITransport.Factory {
             this.transformedTransportParams = transformedTransportParams;
         }
 
+        public List<ProtocolMessage> getPublishedMessages() {
+            return publishedMessages;
+        }
+
+        public void clearPublishedMessages() {
+            publishedMessages.clear();
+        }
+
         @Override
         public void send(ProtocolMessage msg) throws AblyException {
+            if (msg.action == ProtocolMessage.Action.message){
+                publishedMessages.add(msg);
+            }
             switch (sendBehaviour) {
                 case allow:
-                    if (messageFilter == null || messageFilter.matches(msg)) {
+                    if (sendMessageFilter == null || sendMessageFilter.matches(msg)) {
                         super.send(msg);
                     }
                     break;
                 case block:
-                    if (messageFilter == null || messageFilter.matches(msg)) {
+                    if (sendMessageFilter == null || sendMessageFilter.matches(msg)) {
                         /* do nothing */
                     } else {
                         super.send(msg);
                     }
                     break;
                 case fail:
-                    if (messageFilter == null || messageFilter.matches(msg)) {
+                    if (sendMessageFilter == null || sendMessageFilter.matches(msg)) {
                         throw AblyException.fromErrorInfo(new ErrorInfo("Mock", 40000));
                     } else {
                         super.send(msg);
@@ -133,6 +169,32 @@ public class MockWebsocketFactory implements ITransport.Factory {
             }
         }
 
+        @Override
+        public void receive(ProtocolMessage msg) throws AblyException {
+            switch (receiveBehaviour) {
+                case allow:
+                    if (receiveMessageFilter == null || receiveMessageFilter.matches(msg)) {
+                        super.receive(msg);
+                    }
+                    break;
+                case block:
+                    if (receiveMessageFilter == null || receiveMessageFilter.matches(msg)) {
+                        /* do nothing */
+                    } else {
+                        super.receive(msg);
+                    }
+                    break;
+                case fail:
+                    if (receiveMessageFilter == null || receiveMessageFilter.matches(msg)) {
+                        throw AblyException.fromErrorInfo(new ErrorInfo("Mock", 40000));
+                    } else {
+                        super.receive(msg);
+                    }
+                    break;
+            }
+        }
+
+
         @Override
         public void connect(ConnectListener connectListener) {
             String host = givenTransportParams.getHost();