Add cycleBetweenNonNull

dachlatten-flow/src/main/kotlin/de/sipgate/dachlatten/flow/FlowCycler.kt

@@ -4,6 +4,7 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onStart
 import kotlin.time.Duration
 
 fun tickEvery(time: Duration) = flow {
@@ -19,17 +20,34 @@ fun <T> cycleBetween(
     flow2: Flow<T>,
 ): Flow<T> = cycleBetween(tickEvery(tickerInterval), flow1, flow2)
 
-fun <I,T> cycleBetween(
-    ticker: Flow<I>,
+fun <I, T> cycleBetween(
+    ticker: Flow<I?>,
     flow1: Flow<T>,
     flow2: Flow<T>,
 ): Flow<T> {
     var first = false
-    return combine(ticker, flow1, flow2) { _, a, b ->
+    return combine(ticker.onStart { emit(null) }, flow1, flow2) { _, a, b ->
         first = !first
         return@combine when {
             first -> a
             else -> b
         }
     }
 }
+
+fun <T> cycleBetweenNonNull(
+    tickerInterval: Duration,
+    flow1: Flow<T>,
+    flow2: Flow<T>,
+): Flow<T> = cycleBetween(tickEvery(tickerInterval), flow1, flow2)
+
+fun <I, T> cycleBetweenNonNull(ticker: Flow<I?>, flow1: Flow<T>, flow2: Flow<T>): Flow<T> {
+    var first = false
+    return combine(ticker.onStart { emit(null) }, flow1, flow2) { _, a, b ->
+        first = !first
+        when {
+            first -> a ?: b
+            else -> b ?: a
+        }
+    }
+}

dachlatten-flow/src/test/kotlin/de/sipgate/dachlatten/flow/CycleBetweenNonNullTest.kt

@@ -0,0 +1,100 @@
+package de.sipgate.dachlatten.flow
+
+import app.cash.turbine.test
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.onEmpty
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Nested
+import org.junit.jupiter.api.Test
+import kotlin.time.Duration.Companion.seconds
+
+class CycleBetweenNonNullTest {
+    @Test
+    fun cyclerReturnsAlwaysAWhenBIsNull() = runTest {
+        cycleBetweenNonNull(ticker.toFlow(), flowA, nullFlow).test {
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    @Test
+    fun cyclerReturnsAlwaysBWhenAIsNull() = runTest {
+        cycleBetweenNonNull(ticker.toFlow(), nullFlow, flowB).test {
+            assertEquals(contentB, awaitItem())
+
+            tick()
+            assertEquals(contentB, awaitItem())
+
+            tick()
+            assertEquals(contentB, awaitItem())
+        }
+    }
+
+    @Test
+    fun cyclerReturnsNullWhenBothAAndBAreNull() = runTest {
+        cycleBetweenNonNull(ticker.toFlow(), nullFlow, nullFlow).test {
+            Assertions.assertNull(awaitItem())
+
+            tick()
+            Assertions.assertNull(awaitItem())
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun cycleBetweenWithDurationWorks() = runTest {
+        cycleBetweenNonNull(5.seconds, flowA, flowB).test {
+            assertEquals(contentA, awaitItem())
+
+            expectNoEvents()
+            advanceTimeBy(5.seconds)
+            assertEquals(contentB, awaitItem())
+
+            expectNoEvents()
+            advanceTimeBy(5.seconds)
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    @Test
+    fun cycleBetweenWithDurationAndNullWorks() = runTest {
+        cycleBetweenNonNull(ticker.toFlow(), flowA, nullFlow).test {
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    private val contentA = "contentA"
+    private val flowA = flowOf(contentA)
+
+    private val contentB = "contentB"
+    private val flowB = flowOf(contentB)
+
+    private val nullFlow = flowOf<Unit?>(null)
+
+    private val ticker = Channel<Unit?>()
+
+    private suspend fun tick() {
+        ticker.send(null)
+    }
+
+    private fun <T> Channel<T>.toFlow() = receiveAsFlow().onEmpty<T?> { emit(null) }
+}

dachlatten-flow/src/test/kotlin/de/sipgate/dachlatten/flow/CycleBetweenTest.kt

@@ -0,0 +1,73 @@
+package de.sipgate.dachlatten.flow
+
+import app.cash.turbine.test
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.onEmpty
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import kotlin.time.Duration.Companion.seconds
+
+class CycleBetweenTest {
+    @Test
+    fun cycleBetweenSwitchesTheFlowBeingEmitted() = runTest {
+        cycleBetween(ticker.toFlow(), flowA, flowB).test {
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(contentB, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    @Test
+    fun cycleBetweenWorksWithNullValues() = runTest {
+        cycleBetween(ticker.toFlow(), flowA, nullFlow).test {
+            assertEquals(contentA, awaitItem())
+
+            tick()
+            assertEquals(null, awaitItem())
+
+            tick()
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun cycleBetweenWithDurationWorks() = runTest {
+        cycleBetween(5.seconds, flowA, flowB).test {
+            assertEquals(contentA, awaitItem())
+
+            expectNoEvents()
+            advanceTimeBy(5.seconds)
+            assertEquals(contentB, awaitItem())
+
+            expectNoEvents()
+            advanceTimeBy(5.seconds)
+            assertEquals(contentA, awaitItem())
+        }
+    }
+
+    private val contentA = "contentA"
+    private val flowA = flowOf(contentA)
+
+    private val contentB = "contentB"
+    private val flowB = flowOf(contentB)
+
+    private val nullFlow = flowOf<Unit?>(null)
+
+    private val ticker = Channel<String?>()
+
+    private suspend fun tick() {
+        ticker.send(null)
+    }
+
+    private fun <T> Channel<T>.toFlow() = consumeAsFlow().onEmpty<T?> { emit(null) }
+}