Repository: ably/ably-java
Issue: Presence messages superseded whilst channel in attaching state

presence.enter and presence.update (and the others) call presence.updatePresence.

updatePresence will put messages on a queue to be sent later, in the event that the channel is currently attaching. This queue is a HashMap that is keyed by clientid. This means if a presence.enter and presence.update occur in quick succession before the channel attaches, the former is cancelled out by the latter (and any listeners are not called).

There are two possible options:

1. Queue up all presence messages and then send in-order on channel attach.
2. Only send the latest message, but call all event listeners with the result.

The spec needs updating to provide more clarification on what the behaviour should be (see: https://github.com/ably/specification/issues/130), but the current implementation in ably-js is to just send all of the pending messages.


Changed files:
lib/src/main/java/io/ably/lib/realtime/Presence.java, lib/src/main/java/io/ably/lib/realtime/ChannelBase.java, lib/src/test/java/io/ably/lib/test/common/Helpers.java, lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java

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 3d219b33..fb28d966 100644
--- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
+++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
@@ -196,8 +196,7 @@ public abstract class ChannelBase extends EventEmitter<ChannelEvent, ChannelStat
                 PresenceMessage[] presenceMessages = queuedMessage.msg.presence;
                 if (presenceMessages != null && presenceMessages.length > 0) {
                     for (PresenceMessage presenceMessage : presenceMessages) {
-                        this.presence.addPendingPresence(presenceMessage.clientId, presenceMessage,
-                            queuedMessage.listener);
+                        this.presence.addPendingPresence(presenceMessage, queuedMessage.listener);
                     }
                 }
             }
diff --git a/lib/src/main/java/io/ably/lib/realtime/Presence.java b/lib/src/main/java/io/ably/lib/realtime/Presence.java
index e34b8a49..aeaf2a8b 100644
--- a/lib/src/main/java/io/ably/lib/realtime/Presence.java
+++ b/lib/src/main/java/io/ably/lib/realtime/Presence.java
@@ -117,9 +117,11 @@ public class Presence {
         return get(new Param(GET_WAITFORSYNC, String.valueOf(wait)), new Param(GET_CLIENTID, clientId));
     }
 
-    synchronized void addPendingPresence(String clientId, PresenceMessage presenceMessage, CompletionListener listener) {
-        final QueuedPresence queuedPresence = new QueuedPresence(presenceMessage,listener);
-        pendingPresence.put(clientId,queuedPresence);
+    void addPendingPresence(PresenceMessage presenceMessage, CompletionListener listener) {
+        synchronized(channel) {
+            final QueuedPresence queuedPresence = new QueuedPresence(presenceMessage,listener);
+            pendingPresence.add(queuedPresence);
+        }
     }
 
     /**
@@ -739,13 +741,12 @@ public class Presence {
      * @throws AblyException
      */
     public void updatePresence(PresenceMessage msg, CompletionListener listener) throws AblyException {
-        Log.v(TAG, "update(); channel = " + channel.name);
+        Log.v(TAG, "updatePresence(); channel = " + channel.name);
 
         AblyRealtime ably = channel.ably;
         boolean connected = (ably.connection.state == ConnectionState.connected);
-        String clientId;
         try {
-            clientId = ably.auth.checkClientId(msg, false, connected);
+            ably.auth.checkClientId(msg, false, connected);
         } catch(AblyException e) {
             if(listener != null) {
                 listener.onError(e.errorInfo);
@@ -759,8 +760,7 @@ public class Presence {
             case initialized:
                 channel.attach();
             case attaching:
-                QueuedPresence queued = new QueuedPresence(msg, listener);
-                pendingPresence.put(clientId, queued);
+                pendingPresence.add(new QueuedPresence(msg, listener));
                 break;
             case attached:
                 ProtocolMessage message = new ProtocolMessage(ProtocolMessage.Action.presence, channel.name);
@@ -852,7 +852,7 @@ public class Presence {
         QueuedPresence(PresenceMessage msg, CompletionListener listener) { this.msg = msg; this.listener = listener; }
     }
 
-    private final Map<String, QueuedPresence> pendingPresence = new HashMap<String, QueuedPresence>();
+    private final List<QueuedPresence> pendingPresence = new ArrayList<QueuedPresence>();
 
     private void sendQueuedMessages() {
         Log.v(TAG, "sendQueuedMessages()");
@@ -864,7 +864,7 @@ public class Presence {
             return;
 
         ProtocolMessage message = new ProtocolMessage(ProtocolMessage.Action.presence, channel.name);
-        Iterator<QueuedPresence> allQueued = pendingPresence.values().iterator();
+        Iterator<QueuedPresence> allQueued = pendingPresence.iterator();
         PresenceMessage[] presenceMessages = message.presence = new PresenceMessage[count];
         CompletionListener listener;
 
@@ -883,7 +883,9 @@ public class Presence {
             }
             listener = mListener.isEmpty() ? null : mListener;
         }
+
         pendingPresence.clear();
+
         try {
             connectionManager.send(message, queueMessages, listener);
         } catch(AblyException e) {
@@ -895,7 +897,7 @@ public class Presence {
 
     private void failQueuedMessages(ErrorInfo reason) {
         Log.v(TAG, "failQueuedMessages()");
-        for(QueuedPresence msg : pendingPresence.values())
+        for(QueuedPresence msg : pendingPresence)
             if(msg.listener != null)
                 try {
                     msg.listener.onError(reason);
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 788a81bb..bc6cc8fe 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
@@ -164,15 +164,35 @@ public class Helpers {
             error = null;
         }
 
-        public synchronized ErrorInfo waitFor(int count) {
+        /**
+         * Wait for a specified amount of time, or until success occurs.
+         */
+        public synchronized ErrorInfo waitFor(int count, long timeoutInMillis) {
+            long timeoutAt = System.currentTimeMillis() + timeoutInMillis;
             while(successCount<count && error == null)
-                try { wait(); } catch(InterruptedException e) {}
+                try {
+                    if (System.currentTimeMillis() > timeoutAt) {
+                        break;
+                    }
+                    
+                     wait(); 
+                } catch(InterruptedException e) {}
             success = successCount >= count;
             return error;
         }
 
+        /**
+         * Wait for a specified number of successes, with an arbitrarily long timeout.
+         */
+        public synchronized ErrorInfo waitFor(int count) {
+            return waitFor(count, 600000);
+        }
+
+        /**
+         * Wait for a single success with an arbitrarily long timeout.
+         */
         public synchronized ErrorInfo waitFor() {
-            return waitFor(1);
+            return waitFor(1, 600000);
         }
 
         /**
diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java
index 4342b580..7148b9e3 100644
--- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java
+++ b/lib/src/test/java/io/ably/lib/test/realtime/RealtimePresenceTest.java
@@ -1649,6 +1649,97 @@ public class RealtimePresenceTest extends ParameterizedTest {
         }
     }
 
+    /**
+     * <p>
+     * Validates a client sending multiple presence updates when the channel is in the attaching
+     * state will have all messages sent once the channel attaches, and all listeners will be called.
+     * </p>
+     *
+     * @throws AblyException
+     */
+    @Test
+    public void realtime_presence_update_multiple_queued_messages() throws AblyException {
+        /* Ably instance that will emit presence events */
+        AblyRealtime ably1 = null;
+        /* Ably instance that will receive presence events */
+        AblyRealtime ably2 = null;
+
+        String channelName = "test.presence.subscribe.update_multiple_queued_messages" + System.currentTimeMillis();
+        EnumSet<PresenceMessage.Action> actions = EnumSet.of(Action.update, Action.enter);
+
+        try {
+            ClientOptions option1 = createOptions(testVars.keys[0].keyStr);
+            option1.clientId = "emitter client";
+            ClientOptions option2 = createOptions(testVars.keys[0].keyStr);
+            option2.clientId = "receiver client";
+
+            ably1 = new AblyRealtime(option1);
+            ably2 = new AblyRealtime(option2);
+
+            Channel channel1 = ably1.channels.get(channelName);
+
+            Channel channel2 = ably2.channels.get(channelName);
+            channel2.attach();
+            (new ChannelWaiter(channel2)).waitFor(ChannelState.attached);
+
+            CompletionWaiter messageCompletionListener = new CompletionWaiter();
+
+            final ArrayList<PresenceMessage> receivedMessageStack = new ArrayList<>();
+            channel2.presence.subscribe(actions, new Presence.PresenceListener() {
+                @Override
+                public void onPresenceMessage(PresenceMessage message) {
+                    synchronized (receivedMessageStack) {
+                        receivedMessageStack.add(message);
+                        receivedMessageStack.notify();
+                    }
+                }
+            });
+
+            /*
+                Start emitting channel with ably client 1 (emitter)
+
+                This is synchronized against the channel so that channel.setState cant mark
+                the channel as attached until we're done queueing up events.
+            */
+            synchronized (channel1) {
+                channel1.presence.enter("Hello, #2!", messageCompletionListener);
+                channel1.presence.update("Lorem ipsum", messageCompletionListener);
+                channel1.presence.update("Dolor sit!", messageCompletionListener);
+            }
+
+            /* Wait until receiver client (ably2) observes {@code Action.leave}
+             * is emitted from emitter client (ably1)
+             */
+            try {
+                synchronized (receivedMessageStack) {
+                    while (receivedMessageStack.size() == 0 ||
+                            !receivedMessageStack.get(receivedMessageStack.size()-1).clientId.equals(ably1.options.clientId) ||
+                            !receivedMessageStack.get(receivedMessageStack.size()-1).data.equals("Dolor sit!"))
+                                receivedMessageStack.wait();
+                }
+            } catch(InterruptedException e) {}
+
+            /* Validate that,
+             *- we received specific actions
+             */
+            assertThat(receivedMessageStack.size(), is(equalTo(3)));
+            for (PresenceMessage message : receivedMessageStack) {
+                assertTrue(actions.contains(message.action));
+            }
+
+            /*
+             * Validate that
+             * - our listeners are called within 10 seconds
+             */
+            messageCompletionListener.waitFor(3, 10000);
+            assertTrue(messageCompletionListener.success);
+
+        } finally {
+            if (ably1 != null) ably1.close();
+            if (ably2 != null) ably2.close();
+        }
+    }
+
     /**
      * <p>
      * Validates a client can observe presence messages of other client,