Updated ManualDispatchQueue to run items scheduled during a tick

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

@@ -17,7 +17,7 @@ import org.junit.jupiter.api.Test
 import org.mockito.Mock
 import org.mockito.MockitoAnnotations
 import org.mockito.stubbing.Answer
-import org.phoenixframework.utilities.ManualDispatchQueue
+import org.phoenixframework.queue.ManualDispatchQueue
 import org.phoenixframework.utilities.getBindings
 
 class ChannelTest {
@@ -302,7 +302,7 @@ class ChannelTest {
       fakeClock.tick(6_000)
       whenever(socket.isConnected).thenReturn(true)
 
-      fakeClock.tick(5_000)
+      fakeClock.tick(4_000)
       joinPush.trigger("ok", kEmptyPayload)
 
       fakeClock.tick(2_000)
@@ -334,7 +334,7 @@ class ChannelTest {
     }
 
     private fun receivesTimeout(joinPush: Push) {
-      fakeClock.tick(joinPush.timeout * 2)
+      fakeClock.tick(joinPush.timeout)
     }
 
     private fun receivesError(joinPush: Push) {

src/test/kotlin/org/phoenixframework/queue/ManualDispatchQueue.kt

@@ -0,0 +1,75 @@
+package org.phoenixframework.queue
+
+import org.phoenixframework.DispatchQueue
+import org.phoenixframework.DispatchWorkItem
+import java.util.concurrent.TimeUnit
+
+class ManualDispatchQueue : DispatchQueue {
+
+  var tickTime: Long = 0
+  private val tickTimeUnit: TimeUnit = TimeUnit.MILLISECONDS
+  var workItems: MutableList<ManualDispatchWorkItem> = mutableListOf()
+
+  fun reset() {
+    this.tickTime = 0
+    this.workItems = mutableListOf()
+  }
+
+  fun tick(duration: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
+    val durationInMs = tickTimeUnit.convert(duration, unit)
+
+    // calculate what time to advance to
+    val advanceTo = tickTime + durationInMs
+
+    // Filter all work items that are due to be fired and have not been
+    // cancelled. Return early if there are no items to fire
+    var pastDueWorkItems = workItems.filter { it.isPastDue(advanceTo) && !it.isCancelled }
+
+    // Keep looping until there are no more work items that are passed the advance to time
+    while (pastDueWorkItems.isNotEmpty()) {
+
+      // Perform all work items that are due
+      pastDueWorkItems.forEach {
+        tickTime = it.deadline
+        it.perform()
+      }
+
+      // Remove all work items that are past due or canceled
+      workItems.removeAll { it.isPastDue(tickTime) || it.isCancelled }
+      pastDueWorkItems = workItems.filter { it.isPastDue(advanceTo) && !it.isCancelled }
+    }
+
+    // Now that all work has been performed, advance the clock
+    this.tickTime = advanceTo
+
+  }
+
+  override fun queue(delay: Long, unit: TimeUnit, runnable: () -> Unit): DispatchWorkItem {
+    // Converts the given unit and delay to the unit used by this class
+    val delayInMs = tickTimeUnit.convert(delay, unit)
+    val deadline = tickTime + delayInMs
+
+    val workItem = ManualDispatchWorkItem(runnable, deadline)
+    workItems.add(workItem)
+
+    return workItem
+  }
+
+  override fun queueAtFixedRate(
+    delay: Long,
+    period: Long,
+    unit: TimeUnit,
+    runnable: () -> Unit
+  ): DispatchWorkItem {
+
+    val delayInMs = tickTimeUnit.convert(delay, unit)
+    val periodInMs = tickTimeUnit.convert(period, unit)
+    val deadline = tickTime + delayInMs
+
+    val workItem =
+        ManualDispatchWorkItem(runnable, deadline, periodInMs)
+    workItems.add(workItem)
+
+    return workItem
+  }
+}

src/test/kotlin/org/phoenixframework/queue/ManualDispatchQueueTest.kt

@@ -0,0 +1,161 @@
+package org.phoenixframework.queue
+
+import com.google.common.truth.Truth.assertThat
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import java.util.concurrent.TimeUnit
+
+internal class ManualDispatchQueueTest {
+
+
+  private lateinit var queue: ManualDispatchQueue
+
+  @BeforeEach
+  internal fun setUp() {
+    queue = ManualDispatchQueue()
+  }
+
+  @AfterEach
+  internal fun tearDown() {
+    queue.reset()
+  }
+
+  @Test
+  internal fun `reset the queue`() {
+    var task100Called = false
+    var task200Called = false
+    var task300Called = false
+
+    queue.queue(100, TimeUnit.MILLISECONDS) {
+      task100Called = true
+    }
+    queue.queue(200, TimeUnit.MILLISECONDS) {
+      task200Called = true
+    }
+    queue.queue(300, TimeUnit.MILLISECONDS) {
+      task300Called = true
+    }
+
+    queue.tick(250)
+
+    assertThat(queue.tickTime).isEqualTo(250)
+    assertThat(queue.workItems).hasSize(1)
+
+    queue.reset()
+    assertThat(queue.tickTime).isEqualTo(0)
+    assertThat(queue.workItems).isEmpty()
+  }
+
+  @Test
+  internal fun `triggers work that is passed due`() {
+    var task100Called = false
+    var task200Called = false
+    var task300Called = false
+
+    queue.queue(100, TimeUnit.MILLISECONDS) {
+      task100Called = true
+    }
+    queue.queue(200, TimeUnit.MILLISECONDS) {
+      task200Called = true
+    }
+    queue.queue(300, TimeUnit.MILLISECONDS) {
+      task300Called = true
+    }
+
+    queue.tick(100)
+    assertThat(task100Called).isTrue()
+
+    queue.tick(100)
+    assertThat(task200Called).isTrue()
+
+    queue.tick(50)
+    assertThat(task300Called).isFalse()
+  }
+
+  @Test
+  internal fun `triggers all work that is passed due`() {
+    var task100Called = false
+    var task200Called = false
+    var task300Called = false
+
+    queue.queue(100, TimeUnit.MILLISECONDS) {
+      task100Called = true
+    }
+    queue.queue(200, TimeUnit.MILLISECONDS) {
+      task200Called = true
+    }
+    queue.queue(300, TimeUnit.MILLISECONDS) {
+      task300Called = true
+    }
+
+    queue.tick(250)
+    assertThat(task100Called).isTrue()
+    assertThat(task200Called).isTrue()
+    assertThat(task300Called).isFalse()
+  }
+
+  @Test
+  internal fun `triggers work that is scheduled for a time that is after tick`() {
+    var task100Called = false
+    var task200Called = false
+    var task300Called = false
+
+    queue.queue(100, TimeUnit.MILLISECONDS) {
+      task100Called = true
+
+      queue.queue(100, TimeUnit.MILLISECONDS) {
+        task200Called = true
+      }
+
+    }
+
+    queue.queue(300, TimeUnit.MILLISECONDS) {
+      task300Called = true
+    }
+
+    queue.tick(250)
+    assertThat(task100Called).isTrue()
+    assertThat(task200Called).isTrue()
+    assertThat(task300Called).isFalse()
+
+    assertThat(queue.tickTime).isEqualTo(250)
+  }
+
+
+  @Test
+  internal fun `does not triggers nested work that is scheduled outside of the tick`() {
+    var task100Called = false
+    var task200Called = false
+    var task300Called = false
+
+    queue.queue(100, TimeUnit.MILLISECONDS) {
+      task100Called = true
+
+      queue.queue(100, TimeUnit.MILLISECONDS) {
+        task200Called = true
+
+        queue.queue(100, TimeUnit.MILLISECONDS) {
+          task300Called = true
+        }
+      }
+    }
+
+    queue.tick(250)
+    assertThat(task100Called).isTrue()
+    assertThat(task200Called).isTrue()
+    assertThat(task300Called).isFalse()
+  }
+
+  @Test
+  internal fun `queueAtFixedRate repeats work`() {
+    var repeatTaskCallCount = 0
+
+    queue.queueAtFixedRate(100, 100, TimeUnit.MILLISECONDS) {
+      repeatTaskCallCount += 1
+    }
+
+    queue.tick(500)
+    assertThat(repeatTaskCallCount).isEqualTo(5)
+  }
+}

src/test/kotlin/org/phoenixframework/queue/ManualDispatchWorkItem.kt

@@ -0,0 +1,33 @@
+package org.phoenixframework.queue
+
+import org.phoenixframework.DispatchWorkItem
+
+class ManualDispatchWorkItem(
+  private val runnable: () -> Unit,
+  var deadline: Long,
+  private val period: Long = 0
+) : DispatchWorkItem {
+
+  private var performCount = 0
+
+  // Test
+  fun isPastDue(tickTime: Long): Boolean {
+    return this.deadline <= tickTime
+  }
+
+  fun perform() {
+    if (isCancelled) return
+    runnable.invoke()
+    performCount += 1
+
+    // If the task is repeatable, then schedule the next deadline after the given period
+    deadline += period
+  }
+
+  // DispatchWorkItem
+  override var isCancelled: Boolean = false
+
+  override fun cancel() {
+    this.isCancelled = true
+  }
+}