Change default request size in Publisher.openSubscription to one

This makes Publisher.collect consistent with Publisher.asFlow and
ensures smooth integration with reactive backpressure operators
"out of the box".

Fixes #1267

Author: elizarov

reactive/coroutines-guide-reactive.md

@@ -292,19 +292,20 @@ OnSubscribe
 1
 2
 3
-4
 OnComplete
 Finally
+4
 5
 ```
 
 <!--- TEST -->
 
-Notice how "OnComplete" and "Finally" are printed before the last element "5". It happens because our `main` function in this
+Notice how "OnComplete" and "Finally" are printed before the lasts elements "4" and "5". 
+It happens because our `main` function in this
 example is a coroutine that we start with the [runBlocking] coroutine builder.
 Our main coroutine receives on the flowable using the `source.collect { ... }` expression.
 The main coroutine is _suspended_ while it waits for the source to emit an item.
-When the last item is emitted by `Flowable.range(1, 5)` it
+When the last items are emitted by `Flowable.range(1, 5)` it
 _resumes_ the main coroutine, which gets dispatched onto the main thread to print this
  last element at a later point in time, while the source completes and prints "Finally".
 

reactive/kotlinx-coroutines-reactive/src/Channel.kt

@@ -17,11 +17,11 @@ import org.reactivestreams.*
  * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
  *           See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
  *
- * @param request how many items to request from publisher in advance (optional, on-demand request by default).
+ * @param request how many items to request from publisher in advance (optional, one by default).
  */
 @ObsoleteCoroutinesApi
 @Suppress("CONFLICTING_OVERLOADS")
-public fun <T> Publisher<T>.openSubscription(request: Int = 0): ReceiveChannel<T> {
+public fun <T> Publisher<T>.openSubscription(request: Int = 1): ReceiveChannel<T> {
     val channel = SubscriptionChannel<T>(request)
     subscribe(channel)
     return channel

reactive/kotlinx-coroutines-reactor/test/BackpressureTest.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.reactor
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.reactive.*
+import kotlinx.coroutines.reactive.flow.*
+import org.junit.Test
+import reactor.core.publisher.*
+import kotlin.test.*
+
+class BackpressureTest : TestBase() {
+    @Test
+    fun testBackpressureDropDirect() = runTest {
+        expect(1)
+        Flux.fromArray(arrayOf(1))
+            .onBackpressureDrop()
+            .collect {
+                assertEquals(1, it)
+                expect(2)
+            }
+        finish(3)
+    }
+
+    @Test
+    fun testBackpressureDropFlow() = runTest {
+        expect(1)
+        Flux.fromArray(arrayOf(1))
+            .onBackpressureDrop()
+            .asFlow()
+            .collect {
+                assertEquals(1, it)
+                expect(2)
+            }
+        finish(3)
+    }
+}

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.rx2
+
+import io.reactivex.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.reactive.*
+import kotlinx.coroutines.reactive.flow.*
+import org.junit.Test
+import kotlin.test.*
+
+class BackpressureTest : TestBase() {
+    @Test
+    fun testBackpressureDropDirect() = runTest {
+        expect(1)
+        Flowable.fromArray(1)
+            .onBackpressureDrop()
+            .collect {
+                assertEquals(1, it)
+                expect(2)
+            }
+        finish(3)
+    }
+
+    @Test
+    fun testBackpressureDropFlow() = runTest {
+        expect(1)
+        Flowable.fromArray(1)
+            .onBackpressureDrop()
+            .asFlow()
+            .collect {
+                assertEquals(1, it)
+                expect(2)
+            }
+        finish(3)
+    }
+}

reactive/kotlinx-coroutines-rx2/test/guide/test/GuideReactiveTest.kt

@@ -52,9 +52,9 @@ class GuideReactiveTest : ReactiveTestBase() {
             "1",
             "2",
             "3",
-            "4",
             "OnComplete",
             "Finally",
+            "4",
             "5"
         )
     }