Add opt-in dropping behavior to ObservableSource.asFlow

Author: mareklangiewicz

reactive/kotlinx-coroutines-rx2/src/RxConvert.kt

@@ -87,17 +87,19 @@ public fun <T : Any> ReceiveChannel<T>.asObservable(context: CoroutineContext):
  * A channel with the [default][Channel.BUFFERED] buffer size is used. Use the [buffer] operator on the
  * resulting flow to specify a user-defined value and to control what happens when data is produced faster
  * than consumed, i.e. to control the back-pressure behavior. Check [callbackFlow] for more details.
+ *
+ * @param dropWhenFull Set to true to drop items when buffer is full instead of blocking the [ObservableSource].
  */
-@Suppress("NOTHING_TO_INLINE")
+@Suppress("BlockingMethodInNonBlockingContext")
 @ExperimentalCoroutinesApi
-public fun <T: Any> ObservableSource<T>.asFlow(): Flow<T> = callbackFlow {
+public fun <T: Any> ObservableSource<T>.asFlow(dropWhenFull: Boolean = false): Flow<T> = callbackFlow {
 
     var disposable: Disposable? = null
 
     val observer = object : Observer<T> {
         override fun onComplete() { close() }
         override fun onSubscribe(d: Disposable) { disposable = d }
-        override fun onNext(t: T) { sendBlocking(t) }
+        override fun onNext(t: T) { if(!offer(t) && !dropWhenFull) runBlocking { send(t) } }
         override fun onError(e: Throwable) { close(e) }
     }
     subscribe(observer)

reactive/kotlinx-coroutines-rx2/test/ObservableAsFlowTest.kt

@@ -143,6 +143,13 @@ class ObservableAsFlowTest : TestBase() {
         assertEquals(10_000, count)
     }
 
+    @Test
+    fun testLongRangeDroppingWhenFull() = runTest {
+        val source = Observable.range(1, 10_000)
+        val count = source.asFlow(dropWhenFull = true).buffer(10).count()
+        assertEquals(11, count)
+    }
+
     @Test
     fun testProduce() = runTest {
         val source = Observable.range(0, 10)
@@ -154,6 +161,17 @@ class ObservableAsFlowTest : TestBase() {
         check(listOf(0, 9), flow.conflate().produceIn(this))
     }
 
+    @Test
+    fun testProduceDroppingWhenFull() = runTest {
+        val source = Observable.range(0, 10)
+        val flow = source.asFlow(dropWhenFull = true)
+        check((0..9).toList(), flow.produceIn(this))
+        check((0..9).toList(), flow.buffer(Channel.UNLIMITED).produceIn(this))
+        check((0..2).toList(), flow.buffer(2).produceIn(this))
+        check(listOf(0), flow.buffer(0).produceIn(this))
+        check(listOf(0, 9), flow.conflate().produceIn(this))
+    }
+
     private suspend fun check(expected: List<Int>, channel: ReceiveChannel<Int>) {
         val result = ArrayList<Int>(10)
         channel.consumeEach { result.add(it) }