feat(pairwise): add pairwise operator

CHANGELOG.md

@@ -9,6 +9,7 @@
   - `flowFromSuspend`.
   - `mapEager`, `flatMapConcatEager`, `flattenConcatEager`.
   - `skipUntil`, `dropUntil`.
+  - `pairwise`.
 
 - Support for Apple Silicon targets
   - `iosSimulatorArm64`.

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

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;

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<T>
+ *
+ * val result: Flow<Pair<T, T>> = flow
+ *   .bufferCount(bufferSize = 2, startBufferEvery = 1)
+ *   .mapNotNull {
+ *     if (it.size < 2) null
+ *     else it[0] to it[1]
+ *   }
+ * ```
+ */
+public fun <T> Flow<T>.pairwise(): Flow<Pair<T, T>> = flow {
+  var last: Any? = null
+
+  collect {
+    if (last !== null) {
+      emit(Pair(NULL_VALUE.unbox(last), it))
+    }
+    last = it ?: NULL_VALUE
+  }
+}

src/commonTest/kotlin/com/hoc081098/flowext/PairwiseTest.kt

@@ -0,0 +1,131 @@
+/*
+ * 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.map
+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 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<Int>()
+      .pairwise()
+      .test(
+        listOf(
+          Event.Complete
+        )
+      )
+  }
+
+  @Test
+  fun testPairwiseSingle() = runTest {
+    flowOf(1)
+      .pairwise()
+      .test(
+        listOf(
+          Event.Complete
+        )
+      )
+  }
+
+  @Test
+  fun testPairwiseFailureUpstream() = runTest {
+    assertFailsWith<TestException>(
+      flow<Int> { 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
+        )
+      )
+  }
+}