tickerFlow implementation

kotlinx-coroutines-core/jvm/src/flow/TickerFlow.kt

@@ -0,0 +1,33 @@
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.channels.awaitClose
+import java.util.*
+import kotlin.concurrent.schedule
+
+/**
+ * Creates a flow that produces the first item after the given initial delay and subsequent items with the
+ * given delay between them.
+ *
+ * The resulting flow is a callback flow, which basically listens @see [Timer.schedule]
+ *
+ * This Flow stops producing elements immediately after [Job.cancel] invocation.
+ *
+ * @param period period between each element in milliseconds.
+ * @param initialDelay delay after which the first element will be produced (it is equal to [period] by default) in milliseconds.
+ */
+public fun tickerFlow(
+        period: Long,
+        initialDelay: Long = period
+): Flow<Unit> = callbackFlow {
+    require(period > 0)
+    require(initialDelay > -1)
+
+    val timer = Timer()
+    timer.schedule(initialDelay, period) {
+        offer(Unit)
+    }
+
+    awaitClose { timer.cancel() }
+}
+

kotlinx-coroutines-core/jvm/test/flow/TickerFlowTest.kt

@@ -0,0 +1,111 @@
+package flow
+
+import kotlinx.coroutines.TestBase
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.tickerFlow
+import java.util.concurrent.CancellationException
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+
+class TickerFlowTest : TestBase() {
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testNegativePeriod() = runTest {
+        // WHEN
+        tickerFlow(-1).launchIn(this)
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testZeroPeriod() = runTest {
+        // WHEN
+        tickerFlow(0).launchIn(this)
+    }
+
+    @Test(expected = IllegalArgumentException::class)
+    fun testNegativeInitialDelay() = runTest {
+        // WHEN
+        tickerFlow(100, -1).launchIn(this)
+    }
+
+    @Test
+    fun testInitialDelay() = runTest {
+        // GIVEN
+        val inbox = mutableListOf<Unit>()
+
+        // WHEN
+        val periodicTicker =
+                tickerFlow(100, 100).onEach {
+                    inbox.add(Unit)
+                }.launchIn(this)
+
+        delay(500)
+
+        // THEN
+        assertEquals(4, inbox.size)
+
+        periodicTicker.cancelAndJoin()
+    }
+
+    @Test
+    fun testZeroInitialDelay() = runTest {
+        // GIVEN
+        val inbox = mutableListOf<Unit>()
+
+        // WHEN
+        val periodicTicker =
+                tickerFlow(100, 0).onEach {
+                    inbox.add(Unit)
+                }.launchIn(this)
+
+        delay(500)
+
+        // THEN
+        assertEquals(5, inbox.size)
+
+        periodicTicker.cancelAndJoin()
+    }
+
+
+    @Test
+    fun testReceive() = runTest {
+        // GIVEN
+        val inbox = mutableListOf<Unit>()
+
+        // WHEN
+        val periodicTicker =
+                tickerFlow(100).onEach {
+                    inbox.add(Unit)
+                }.launchIn(this)
+
+        delay(500)
+
+        // THEN
+        assertEquals(4, inbox.size)
+
+        periodicTicker.cancelAndJoin()
+    }
+
+    @Test
+    fun testDoNotReceiveAfterCancel() = runTest {
+        // GIVEN
+        val inbox = mutableListOf<Unit>()
+
+        // WHEN
+        val periodicTicker =
+                tickerFlow(100).onEach {
+                    inbox.add(Unit)
+                }.launchIn(this)
+
+        delay(50)
+        periodicTicker.cancel(CancellationException())
+
+        // THEN
+        assertEquals(0, inbox.size)
+    }
+
+
+}