From 77387849ea742adbd15e4544693a82d92060474a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Petrus=20Nguy=E1=BB=85n=20Th=C3=A1i=20H=E1=BB=8Dc?= Date: Wed, 6 Jul 2022 12:32:55 +0700 Subject: [PATCH 1/3] feat(pairwise): add `pairwise` operator --- README.md | 5 + api/FlowExt.api | 4 + .../kotlin/com/hoc081098/flowext/pairwise.kt | 58 +++++++++ .../com/hoc081098/flowext/PairwiseTest.kt | 113 ++++++++++++++++++ 4 files changed, 180 insertions(+) create mode 100644 src/commonMain/kotlin/com/hoc081098/flowext/pairwise.kt create mode 100644 src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt diff --git a/README.md b/README.md index c35f0739..4a979c94 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,7 @@ - [`dematerialize`](#dematerialize) - [`raceWith`](#racewith--ambwith) - [`ambWith`](#racewith--ambwith) + - [`pairwise`](#pairwise) - `retryWhenWithDelayStrategy` - `retryWhenWithExponentialBackoff` - `retryWithExponentialBackoff` @@ -173,6 +174,10 @@ . - Similar to [RxJS raceWith](https://rxjs.dev/api/operators/raceWith) +#### pairwise + +- Similar to [RxJS pairwise](https://rxjs.dev/api/operators/pairwise) + #### skipUntil / dropUntil - ReactiveX docs: https://reactivex.io/documentation/operators/skipuntil.html diff --git a/api/FlowExt.api b/api/FlowExt.api index 025d8d5f..f675ef8c 100644 --- a/api/FlowExt.api +++ b/api/FlowExt.api @@ -137,6 +137,10 @@ public final class com/hoc081098/flowext/NeverFlowKt { public static final fun neverFlow ()Lkotlinx/coroutines/flow/Flow; } +public final class com/hoc081098/flowext/PairwiseKt { + public static final fun pairwise (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow; +} + public final class com/hoc081098/flowext/RaceKt { public static final fun amb (Ljava/lang/Iterable;)Lkotlinx/coroutines/flow/Flow; public static final fun amb (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;[Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow; diff --git a/src/commonMain/kotlin/com/hoc081098/flowext/pairwise.kt b/src/commonMain/kotlin/com/hoc081098/flowext/pairwise.kt new file mode 100644 index 00000000..3f65da8b --- /dev/null +++ b/src/commonMain/kotlin/com/hoc081098/flowext/pairwise.kt @@ -0,0 +1,58 @@ +/* + * MIT License + * + * Copyright (c) 2021-2022 Petrus Nguyễn Thái Học + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.hoc081098.flowext + +import com.hoc081098.flowext.utils.NULL_VALUE +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow + +/** + * Groups pairs of consecutive emissions together and emits them as a pair. + * + * Emits the `(n)th` and `(n-1)th` events as a pair. + * The first value won't be emitted until the second one arrives. + * + * This operator is more optimizer than [bufferCount] version: + * ```kotlin + * val flow: Flow + * + * val result: Flow> = flow + * .bufferCount(bufferSize = 2, startBufferEvery = 1) + * .mapNotNull { + * if (it.size < 2) null + * else it[0] to it[1] + * } + * ``` + */ +public fun Flow.pairwise(): Flow> = flow { + var last: Any? = null + + collect { + if (last !== null) { + emit(Pair(NULL_VALUE.unbox(last), it)) + } + last = it ?: NULL_VALUE + } +} diff --git a/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt b/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt new file mode 100644 index 00000000..e377cb3f --- /dev/null +++ b/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt @@ -0,0 +1,113 @@ +/* + * MIT License + * + * Copyright (c) 2021-2022 Petrus Nguyễn Thái Học + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.hoc081098.flowext + +import com.hoc081098.flowext.utils.BaseStepTest +import com.hoc081098.flowext.utils.TestException +import com.hoc081098.flowext.utils.assertFailsWith +import com.hoc081098.flowext.utils.test +import kotlin.test.Test +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.mapNotNull +import kotlinx.coroutines.flow.take + +@ExperimentalCoroutinesApi +class PairwiseTest : BaseStepTest() { + @Test + fun testPairwise() = runTest { + range(0, 4) + .pairwise() + .test( + listOf( + Event.Value(0 to 1), + Event.Value(1 to 2), + Event.Value(2 to 3), + Event.Complete + ) + ) + + range(0, 4) + .bufferCount(bufferSize = 2, startBufferEvery = 1) + .mapNotNull { + if (it.size < 2) null + else it[0] to it[1] + } + .test( + listOf( + Event.Value(0 to 1), + Event.Value(1 to 2), + Event.Value(2 to 3), + Event.Complete + ) + ) + } + + @Test + fun testPairwiseEmpty() = runTest { + emptyFlow() + .pairwise() + .test( + listOf( + Event.Complete + ) + ) + } + + @Test + fun testPairwiseSingle() = runTest { + flowOf(1) + .pairwise() + .test( + listOf( + Event.Complete + ) + ) + } + + @Test + fun testPairwiseFailureUpstream() = runTest { + assertFailsWith( + flow { throw TestException() } + .pairwise() + ) + } + + @Test + fun testPairwiseCancellation() = runTest { + range(1, 100) + .pairwise() + .take(2) + .test( + listOf( + Event.Value(1 to 2), + Event.Value(2 to 3), + Event.Complete + ) + ) + } +} From 6f29088fbc95218dda311a91d8bf9403d1de54fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Petrus=20Nguy=E1=BB=85n=20Th=C3=A1i=20H=E1=BB=8Dc?= Date: Wed, 6 Jul 2022 12:36:12 +0700 Subject: [PATCH 2/3] feat(pairwise): add `pairwise` operator CHANGELOG.md --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 079f25f7..fed206e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ - `flowFromSuspend`. - `mapEager`, `flatMapConcatEager`, `flattenConcatEager`. - `skipUntil`, `dropUntil`. + - `pairwise`. - Support for Apple Silicon targets - `iosSimulatorArm64`. From 93efe7495d688313982408f52d89e16d8775145f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Petrus=20Nguy=E1=BB=85n=20Th=C3=A1i=20H=E1=BB=8Dc?= Date: Wed, 6 Jul 2022 12:48:18 +0700 Subject: [PATCH 3/3] feat(pairwise): tests --- .../com/hoc081098/flowext/PairwiseTest.kt | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt b/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt index e377cb3f..f6d6d578 100644 --- a/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt +++ b/src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt @@ -33,6 +33,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.mapNotNull import kotlinx.coroutines.flow.take @@ -67,6 +68,23 @@ class PairwiseTest : BaseStepTest() { ) } + @Test + fun testPairwiseNullable() = runTest { + // 0 - null - 2 - null + + range(0, 4) + .map { it.takeIf { it % 2 == 0 } } + .pairwise() + .test( + listOf( + Event.Value(0 to null), + Event.Value(null to 2), + Event.Value(2 to null), + Event.Complete + ) + ) + } + @Test fun testPairwiseEmpty() = runTest { emptyFlow()