Manage bindings in concurrent data structures. (#26)

* Use concurrenthshmap and concurrentlinkedqueue for managing bindings

* Change bindings to a val (from a var)

Author: dustinconrad

src/main/kotlin/org/phoenixframework/PhxChannel.kt

@@ -1,6 +1,8 @@
 package org.phoenixframework
 
 import java.lang.IllegalStateException
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedQueue
 
 
 class PhxChannel(
@@ -44,7 +46,7 @@ class PhxChannel(
 
 
     var state: PhxChannel.PhxState
-    var bindings: MutableList<Triple<String, Int, (PhxMessage) -> Unit>>
+    val bindings: ConcurrentHashMap<String, ConcurrentLinkedQueue<Pair<Int, (PhxMessage) -> Unit>>>
     var bindingRef: Int
     var timeout: Long
     var joinedOnce: Boolean
@@ -58,7 +60,7 @@ class PhxChannel(
 
     init {
         this.state = PhxChannel.PhxState.CLOSED
-        this.bindings = ArrayList()
+        this.bindings = ConcurrentHashMap()
         this.bindingRef = 0
         this.timeout = socket.timeout
         this.joinedOnce = false
@@ -197,7 +199,9 @@ class PhxChannel(
         val ref = bindingRef
         this.bindingRef = ref + 1
 
-        this.bindings.add(Triple(event, ref, callback))
+        this.bindings.getOrPut(event) { ConcurrentLinkedQueue() }
+                .add(ref to callback)
+
         return ref
     }
 
@@ -219,9 +223,11 @@ class PhxChannel(
     public fun off(event: String, ref: Int? = null) {
         // Remove any subscriptions that match the given event and ref ID. If no ref
         // ID is given, then remove all subscriptions for an event.
-        this.bindings = bindings
-                .filterNot { it.first == event && (ref == null || ref == it.second) }
-                .toMutableList()
+        if (ref != null) {
+            this.bindings[event]?.removeIf{ ref == it.first }
+        } else {
+            this.bindings.remove(event)
+        }
     }
 
     /**
@@ -333,9 +339,7 @@ class PhxChannel(
      */
     fun trigger(message: PhxMessage) {
         val handledMessage = onMessage(message)
-        this.bindings
-                .filter { it.first == message.event  }
-                .forEach { it.third(handledMessage) }
+        this.bindings[message.event]?.forEach { it.second(handledMessage) }
     }
 
     /**

src/test/kotlin/org/phoenixframework/PhxChannelTest.kt

@@ -6,6 +6,8 @@ import org.junit.Test
 import org.mockito.Mockito
 import org.mockito.MockitoAnnotations
 import org.mockito.Spy
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
 
 class PhxChannelTest {
 
@@ -124,5 +126,28 @@ class PhxChannelTest {
         assertThat(aCalled).isFalse()
         assertThat(bCalled).isTrue()
     }
+
+    @Test
+    fun `Issue 22`() {
+        // This reproduces a concurrent modification exception.  The original cause is most likely as follows:
+        // 1.  Push (And receive) messages very quickly
+        // 2.  PhxChannel.push, calls PhxPush.send()
+        // 3.  PhxPush calls startTimeout().
+        // 4.  PhxPush.startTimeout() calls this.channel.on(refEvent) - This modifies the bindings list
+        // 5.  any trigger (possibly from a timeout) can be iterating through the binding list that was modified in step 4.
+
+        val f1 = CompletableFuture.runAsync {
+            for (i in 0..1000) {
+                channel.on("event-$i") { /** do nothing **/ }
+            }
+        }
+        val f3 = CompletableFuture.runAsync {
+            for (i in 0..1000) {
+                channel.trigger(PhxMessage(event = "event-$i", ref = defaultRef))
+            }
+        }
+
+        CompletableFuture.allOf(f1, f3).get(10, TimeUnit.SECONDS)
+    }
 }