Repository: ably/ably-java
Issue: `ConcurrentModificationException` when `unsubscribe` then `detach` channel presence listener

A customer is experiencing a crash with `ably-java` [version 1.2.10](https://github.com/ably/ably-java/releases/tag/v1.2.10), therefore my understanding is that this is pure JVM (i.e. a server type application - I think it's some kind of a bot), not Android. They have described the scenario to us:

> When the UserA encounters a new user join to chatroom he was in , it registers the user presence listener.

```java
presenceChannel.presence.subscribe(member -> {
```

> When UserA no longer needs to track the users presence (not a member of any active chat rooms), the listener is removed:

```java
presenceChannel.presence.unsubscribe();
try {
    presenceChannel.detach();
} catch (AblyException e) {
    throw new BotException(e.toString());
}
```

> The error is encountered mostly when a user goes offline....
> The following is the exception stack:

```
2022-01-25T15:37:29,129|INFO|SomeClient|WebSocketConnectReadThread-132|{"l":"DEBUG","msg":"receive presence event, action: leave, data: {\\"userId\\":\\"SomeUserIdValue\\",\\"status\\":\\"Online\\",\\"customStatus\\":\\"Available From SomeNamedTool\\"}","fnc":"SomeClient.registerUserPresence"} 
2022-01-25T15:37:29,131|DEBUG|UserPresenceManager|WebSocketConnectReadThread-132|updateUserPresence: userId=PATESTFICA138, source=MESSAGING, state=OFFLINE 
(ERROR): io.ably.lib.transport.WebSocketTransport: Unexpected exception processing received binary message 

io.ably.lib.types.AblyException: java.util.ConcurrentModificationException 
    at io.ably.lib.types.AblyException.fromThrowable(AblyException.java:54) 
    at io.ably.lib.transport.ConnectionManager.onMessage(ConnectionManager.java:1080) 
    at io.ably.lib.transport.WebSocketTransport$WsClient.onMessage(WebSocketTransport.java:161) 
    at org.java_websocket.client.WebSocketClient.onWebsocketMessage(WebSocketClient.java:505) 
    at org.java_websocket.drafts.Draft_6455.processFrameBinary(Draft_6455.java:835) 
    at org.java_websocket.drafts.Draft_6455.processFrame(Draft_6455.java:794) 
    at org.java_websocket.WebSocketImpl.decodeFrames(WebSocketImpl.java:381) 
    at org.java_websocket.WebSocketImpl.decode(WebSocketImpl.java:218) 
    at org.java_websocket.client.WebSocketClient.run(WebSocketClient.java:425) 
    at java.base/java.lang.Thread.run(Thread.java:829) 

Caused by: java.util.ConcurrentModificationException 
    at java.base/java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1043) 
    at java.base/java.util.ArrayList$Itr.next(ArrayList.java:997) 
    at io.ably.lib.realtime.Presence$Multicaster.onPresenceMessage(Presence.java:359) 
    at io.ably.lib.realtime.Presence.broadcastPresence(Presence.java:345) 
    at io.ably.lib.realtime.Presence.setPresence(Presence.java:334) 
    at io.ably.lib.realtime.ChannelBase.onPresence(ChannelBase.java:769) 
    at io.ably.lib.realtime.ChannelBase.onChannelMessage(ChannelBase.java:1158) 
    at io.ably.lib.realtime.AblyRealtime$InternalChannels.onMessage(AblyRealtime.java:187) 
    at io.ably.lib.transport.ConnectionManager.onChannelMessage(ConnectionManager.java:1090) 
    at io.ably.lib.transport.ConnectionManager.onMessage(ConnectionManager.java:1075) 
    ... 8 more
```

(some details redacted above)

Hubspot (internal) threads conversing with this customer are [here](https://app.hubspot.com/live-messages/6939709/inbox/2125159045#email-reply-editor) and, then, [here](https://app.hubspot.com/live-messages/6939709/inbox/2126124055#email-reply-editor).



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


Changed files:
lib/src/main/java/io/ably/lib/realtime/ChannelBase.java, lib/src/main/java/io/ably/lib/realtime/ConnectionStateListener.java, lib/src/main/java/io/ably/lib/util/Multicaster.java, lib/src/main/java/io/ably/lib/realtime/Presence.java, lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java, lib/src/main/java/io/ably/lib/realtime/CompletionListener.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 f5a8cc0d..c32b6769 100644
--- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
+++ b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java
@@ -782,7 +782,7 @@ public abstract class ChannelBase extends EventEmitter<ChannelEvent, ChannelStat
     private static class MessageMulticaster extends io.ably.lib.util.Multicaster<MessageListener> implements MessageListener {
         @Override
         public void onMessage(Message message) {
-            for(MessageListener member : members)
+            for (final MessageListener member : getMembers())
                 try {
                     member.onMessage(message);
                 } catch (Throwable t) {
diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java b/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java
index 963703eb..5de088c5 100644
--- a/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java
+++ b/lib/src/main/java/io/ably/lib/realtime/ChannelStateListener.java
@@ -56,7 +56,7 @@ public interface ChannelStateListener {
     class Multicaster extends io.ably.lib.util.Multicaster<ChannelStateListener> implements ChannelStateListener {
         @Override
         public void onChannelStateChanged(ChannelStateChange stateChange) {
-            for(ChannelStateListener member : members)
+            for (final ChannelStateListener member : getMembers())
                 try {
                     member.onChannelStateChanged(stateChange);
                 } catch(Throwable t) {}
diff --git a/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java b/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java
index 431cb39f..38dc923a 100644
--- a/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java
+++ b/lib/src/main/java/io/ably/lib/realtime/CompletionListener.java
@@ -29,7 +29,7 @@ public interface CompletionListener {
 
         @Override
         public void onSuccess() {
-            for(CompletionListener member : members)
+            for (final CompletionListener member : getMembers())
                 try {
                     member.onSuccess();
                 } catch(Throwable t) {}
@@ -37,7 +37,7 @@ public interface CompletionListener {
 
         @Override
         public void onError(ErrorInfo reason) {
-            for(CompletionListener member : members)
+            for (final CompletionListener member : getMembers())
                 try {
                     member.onError(reason);
                 } catch(Throwable t) {}
diff --git a/lib/src/main/java/io/ably/lib/realtime/ConnectionStateListener.java b/lib/src/main/java/io/ably/lib/realtime/ConnectionStateListener.java
index bcc444b9..10a9c403 100644
--- a/lib/src/main/java/io/ably/lib/realtime/ConnectionStateListener.java
+++ b/lib/src/main/java/io/ably/lib/realtime/ConnectionStateListener.java
@@ -38,7 +38,7 @@ public interface ConnectionStateListener {
     class Multicaster extends io.ably.lib.util.Multicaster<ConnectionStateListener> implements ConnectionStateListener {
         @Override
         public void onConnectionStateChanged(ConnectionStateChange state) {
-            for(ConnectionStateListener member : members)
+            for (final ConnectionStateListener member : getMembers())
                 try {
                     member.onConnectionStateChanged(state);
                 } catch(Throwable t) {}
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 529a8d90..044f7d79 100644
--- a/lib/src/main/java/io/ably/lib/realtime/Presence.java
+++ b/lib/src/main/java/io/ably/lib/realtime/Presence.java
@@ -357,7 +357,7 @@ public class Presence {
     private static class Multicaster extends io.ably.lib.util.Multicaster<PresenceListener> implements PresenceListener {
         @Override
         public void onPresenceMessage(PresenceMessage message) {
-            for(PresenceListener member : members)
+            for (final PresenceListener member : getMembers())
                 try {
                     member.onPresenceMessage(message);
                 } catch(Throwable t) {}
diff --git a/lib/src/main/java/io/ably/lib/util/Multicaster.java b/lib/src/main/java/io/ably/lib/util/Multicaster.java
index be9fb046..9cd3f713 100644
--- a/lib/src/main/java/io/ably/lib/util/Multicaster.java
+++ b/lib/src/main/java/io/ably/lib/util/Multicaster.java
@@ -1,19 +1,27 @@
 package io.ably.lib.util;
 
 import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.List;
 
+/**
+ * Collection of members who are listeners, with methods that are safe to be called from any thread.
+ * @param <T> The type of elements being added to this multicaster - the listeners.
+ */
 public abstract class Multicaster<T> {
-
-    protected final List<T> members = new ArrayList<T>();
+    private final List<T> members = new ArrayList<>();
 
     public Multicaster(T... members) { for(T m : members) this.members.add(m); }
-    
-    public void add(T member) { members.add(member); }
-    public void remove(T member) { members.remove(member); }
-    public void clear() { members.clear(); }
-    public boolean isEmpty() { return members.isEmpty(); }
-    public int size() { return members.size(); }
-    public Iterator<T>  iterator() { return members.iterator(); }
+
+    public synchronized void add(T member) { members.add(member); }
+    public synchronized void remove(T member) { members.remove(member); }
+    public synchronized void clear() { members.clear(); }
+    public synchronized boolean isEmpty() { return members.isEmpty(); }
+    public synchronized int size() { return members.size(); }
+
+    /**
+     * Returns a snapshot of the members of this multicaster instance.
+     */
+    protected synchronized List<T> getMembers() {
+        return new ArrayList<>(members);
+    }
 }