Repository: ably/ably-java
Issue: Released channel re-added to the channel map after DETACHED message

If you call channels.release(), that removes the channel from the channel map and detaches it. However, the protocol messaged for DETACHED returning to us re-adds the channel to the channel map (via a call to channels.get()), therefore undoing the release action.

One way to fix this would be to not re-add the channel when processing DETACHED if its not there.



Changed files:
lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java, lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java, lib/src/test/java/io/ably/lib/test/common/Helpers.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 666521af..ffb33d42 100644
--- a/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
+++ b/lib/src/main/java/io/ably/lib/realtime/AblyRealtime.java
@@ -217,8 +217,12 @@ public class AblyRealtime extends AblyRest {
         @Override
         public void onMessage(ProtocolMessage msg) {
             String channelName = msg.channel;
-            Channel channel;
-            synchronized(this) { channel = channels.get(channelName); }
+            Channel channel = null;
+            synchronized(this) {
+                if (channels.containsKey(channelName)) {
+                    channel = channels.get(channelName);
+                }
+            }
             if(channel == null) {
                 Log.e(TAG, "Received channel message for non-existent channel");
                 return;
diff --git a/lib/src/test/java/io/ably/lib/test/common/Helpers.java b/lib/src/test/java/io/ably/lib/test/common/Helpers.java
index bc6cc8fe..2dce9b10 100644
--- a/lib/src/test/java/io/ably/lib/test/common/Helpers.java
+++ b/lib/src/test/java/io/ably/lib/test/common/Helpers.java
@@ -632,24 +632,56 @@ public class Helpers {
          * Wait for a given number of messages
          */
         public void waitForRecv() {
-            waitForRecv(1);
+            waitForRecv(1, 6000000);
         }
         public void waitForSend() {
-            waitForSend(1);
+            waitForSend(1, 6000000);
+        }
+        public void waitForRecv(int count) {
+            waitForRecv(count, 6000000);
+        }
+        public void waitForSend(int count) {
+            waitForSend(count, 6000000);
         }
 
         /**
          * Wait for a given number of messages
          * @param count
          */
-        public synchronized void waitForRecv(int count) {
+        public synchronized void waitForRecv(int count, long timeoutInMillis) {
+            long timeoutAt = System.currentTimeMillis() + timeoutInMillis;
             while(receivedMessages.size() < count) {
-                try { wait(); } catch(InterruptedException e) {}
+                synchronized (this) {
+                    try {
+                        if (System.currentTimeMillis() > timeoutAt || receivedMessages.size() >= count) {
+                            break;
+                        }
+
+                        wait();
+                    } catch(InterruptedException e) {}
+                }
+            }
+
+            if (receivedMessages.size() < count) {
+                throw new AssertionError("Did not receive expected number of messages");
             }
         }
-        public synchronized void waitForSend(int count) {
+        public synchronized void waitForSend(int count, long timeoutInMillis) {
+            long timeoutAt = System.currentTimeMillis() + timeoutInMillis;
             while(sentMessages.size() < count) {
-                try { wait(); } catch(InterruptedException e) {}
+                synchronized (this) {
+                    try {
+                        if (System.currentTimeMillis() > timeoutAt || sentMessages.size() >= count) {
+                            break;
+                        }
+
+                        wait();
+                    } catch(InterruptedException e) {}
+                }
+            }
+
+            if (sentMessages.size() < count) {
+                throw new AssertionError("Did not send expected number of messages");
             }
         }
 
diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java
index b7391033..46ddaaac 100644
--- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java
+++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java
@@ -2007,6 +2007,41 @@ public class RealtimeChannelTest extends ParameterizedTest {
         }
     }
 
+    /*
+     * Checks that the DETACHED message sent by the server when a channel is released is dropped.
+     */
+    @Test
+    public void detach_message_to_released_channel_is_dropped() throws AblyException {
+        AblyRealtime ably = null;
+        long oldRealtimeTimeout = Defaults.realtimeRequestTimeout;
+        final String channelName = "detach_message_to_released_channel_is_dropped";
+
+        try {
+            DebugOptions opts = createOptions(testVars.keys[0].keyStr);
+            Helpers.RawProtocolMonitor monitor = Helpers.RawProtocolMonitor.createReceiver(ProtocolMessage.Action.detached);
+            opts.protocolListener = monitor;
+
+            /* Make test faster */
+            Defaults.realtimeRequestTimeout = 1000;
+            opts.channelRetryTimeout = 1000;
+
+            ably = new AblyRealtime(opts);
+            Channel channel = ably.channels.get(channelName);
+            channel.attach();
+            (new ChannelWaiter(channel)).waitFor(ChannelState.attached);
+
+            // Listen for detach messages and release the channel
+            ably.channels.release(channelName);
+            monitor.waitForRecv(1, 10000);
+
+            assertFalse(ably.channels.containsKey("messages_to_non_existent_channels_are_dropped"));
+        } finally {
+            if (ably != null)
+                ably.close();
+            Defaults.realtimeRequestTimeout = oldRealtimeTimeout;
+        }
+    }
+
     class DetachingProtocolListener implements DebugOptions.RawProtocolListener {
 
         public Channel theChannel;