Skip to content

Commit deda827

Browse files
qwwdfsadqwwdfsad
committed
Deprecate consumeEach on Publisher, ObservableSource and MaybeSource, introduce collect extension instead to be aligned with Flow
First wave of #1080
1 parent fcfabee commit deda827

31 files changed

+91
-68
lines changed

‎binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactive.txt

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ public final class kotlinx/coroutines/reactive/AwaitKt {
88
}
99

1010
public final class kotlinx/coroutines/reactive/ChannelKt {
11+
public static final fun collect (Lorg/reactivestreams/Publisher;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1112
public static final fun consumeEach (Lorg/reactivestreams/Publisher;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1213
public static final fun openSubscription (Lorg/reactivestreams/Publisher;I)Lkotlinx/coroutines/channels/ReceiveChannel;
1314
public static synthetic fun openSubscription$default (Lorg/reactivestreams/Publisher;IILjava/lang/Object;)Lkotlinx/coroutines/channels/ReceiveChannel;

‎binary-compatibility-validator/reference-public-api/kotlinx-coroutines-rx2.txt

+2
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@ public final class kotlinx/coroutines/rx2/RxAwaitKt {
1212
}
1313

1414
public final class kotlinx/coroutines/rx2/RxChannelKt {
15+
public static final fun collect (Lio/reactivex/MaybeSource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
16+
public static final fun collect (Lio/reactivex/ObservableSource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1517
public static final fun consumeEach (Lio/reactivex/MaybeSource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1618
public static final fun consumeEach (Lio/reactivex/ObservableSource;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
1719
public static final fun openSubscription (Lio/reactivex/MaybeSource;)Lkotlinx/coroutines/channels/ReceiveChannel;

‎reactive/coroutines-guide-reactive.md

+21-22
Original file line numberDiff line numberDiff line change
@@ -162,12 +162,12 @@ fun main() = runBlocking<Unit> {
162162
}
163163
// print elements from the source
164164
println("Elements:")
165-
source.consumeEach { // consume elements from it
165+
source.collect { // collect elements from it
166166
println(it)
167167
}
168168
// print elements from the source AGAIN
169169
println("Again:")
170-
source.consumeEach { // consume elements from it
170+
source.collect { // collect elements from it
171171
println(it)
172172
}
173173
}
@@ -198,7 +198,7 @@ elements is produced. It becomes the actual stream of elements on _subscription_
198198
a different stream of elements, depending on how the corresponding implementation of `Publisher` works.
199199

200200
The [publish] coroutine builder, that is used in the above example, launches a fresh coroutine on each subscription.
201-
Every [Publisher.consumeEach][org.reactivestreams.Publisher.consumeEach] invocation creates a fresh subscription.
201+
Every [Publisher.collect][org.reactivestreams.Publisher.collect] invocation creates a fresh subscription.
202202
We have two of them in this code and that is why we see "Begin" printed twice.
203203

204204
In Rx lingo this is called a _cold_ publisher. Many standard Rx operators produce cold streams, too. We can iterate
@@ -217,9 +217,9 @@ method with it.
217217

218218
### Subscription and cancellation
219219

220-
An example in the previous section uses `source.consumeEach { ... }` snippet to open a subscription
220+
An example in the previous section uses `source.collect { ... }` snippet to open a subscription
221221
and receive all the elements from it. If we need more control on how what to do with
222-
the elements that are being received from the channel, we can use [Publisher.openSubscription][org.reactivestreams.Publisher.openSubscription]
222+
the elements that are being received from the channel, we can use [Publisher.collect][org.reactivestreams.Publisher.collect]
223223
as shown in the following example:
224224

225225
<!--- INCLUDE
@@ -269,7 +269,7 @@ listener prints "Finally" to confirm that the subscription is actually being clo
269269
is never printed because we did not consume all of the elements.
270270

271271
We do not need to use an explicit `cancel` either if iteration is performed over all the items that are emitted
272-
by the publisher, because it is being cancelled automatically by `consumeEach`:
272+
by the publisher, because it is being cancelled automatically by `collect`:
273273

274274
<!--- INCLUDE
275275
import io.reactivex.*
@@ -285,7 +285,7 @@ fun main() = runBlocking<Unit> {
285285
.doOnComplete { println("OnComplete") } // ...
286286
.doFinally { println("Finally") } // ... into what's going on
287287
// iterate over the source fully
288-
source.consumeEach { println(it) }
288+
source.collect { println(it) }
289289
}
290290
```
291291

@@ -308,7 +308,7 @@ Finally
308308

309309
Notice, how "OnComplete" and "Finally" are printed before the last element "5". It happens because our `main` function in this
310310
example is a coroutine that we start with [runBlocking] coroutine builder.
311-
Our main coroutine receives on the channel using `source.consumeEach { ... }` expression.
311+
Our main coroutine receives on the flowable using `source.collect { ... }` expression.
312312
The main coroutine is _suspended_ while it waits for the source to emit an item.
313313
When the last item is emitted by `Flowable.range(1, 5)` it
314314
_resumes_ the main coroutine, which gets dispatched onto the main thread to print this
@@ -422,7 +422,7 @@ You can subscribe to subjects from a coroutine just as with any other reactive s
422422
<!--- INCLUDE
423423
import io.reactivex.subjects.BehaviorSubject
424424
import kotlinx.coroutines.*
425-
import kotlinx.coroutines.rx2.consumeEach
425+
import kotlinx.coroutines.rx2.collect
426426
-->
427427

428428
```kotlin
@@ -432,7 +432,7 @@ fun main() = runBlocking<Unit> {
432432
subject.onNext("two")
433433
// now launch a coroutine to print everything
434434
GlobalScope.launch(Dispatchers.Unconfined) { // launch coroutine in unconfined context
435-
subject.consumeEach { println(it) }
435+
subject.collect { println(it) }
436436
}
437437
subject.onNext("three")
438438
subject.onNext("four")
@@ -476,7 +476,7 @@ fun main() = runBlocking<Unit> {
476476
subject.onNext("two")
477477
// now launch a coroutine to print the most recent update
478478
launch { // use the context of the main thread for a coroutine
479-
subject.consumeEach { println(it) }
479+
subject.collect { println(it) }
480480
}
481481
subject.onNext("three")
482482
subject.onNext("four")
@@ -584,7 +584,7 @@ It is straightforward to use from a coroutine:
584584
```kotlin
585585
fun main() = runBlocking<Unit> {
586586
// Range inherits parent job from runBlocking, but overrides dispatcher with Dispatchers.Default
587-
range(Dispatchers.Default, 1, 5).consumeEach { println(it) }
587+
range(Dispatchers.Default, 1, 5).collect { println(it) }
588588
}
589589
```
590590

@@ -623,7 +623,7 @@ fun <T, R> Publisher<T>.fusedFilterMap(
623623
predicate: (T) -> Boolean, // the filter predicate
624624
mapper: (T) -> R // the mapper function
625625
) = GlobalScope.publish<R>(context) {
626-
consumeEach { // consume the source stream
626+
collect { // consume the source stream
627627
if (predicate(it)) // filter part
628628
send(mapper(it)) // map part
629629
}
@@ -644,7 +644,7 @@ fun CoroutineScope.range(start: Int, count: Int) = publish<Int> {
644644
fun main() = runBlocking<Unit> {
645645
range(1, 5)
646646
.fusedFilterMap(coroutineContext, { it % 2 == 0}, { "$it is even" })
647-
.consumeEach { println(it) } // print all the resulting strings
647+
.collect { println(it) } // print all the resulting strings
648648
}
649649
```
650650

@@ -717,7 +717,7 @@ The following code shows how `takeUntil` works:
717717
fun main() = runBlocking<Unit> {
718718
val slowNums = rangeWithInterval(200, 1, 10) // numbers with 200ms interval
719719
val stop = rangeWithInterval(500, 1, 10) // the first one after 500ms
720-
slowNums.takeUntil(coroutineContext, stop).consumeEach { println(it) } // let's test it
720+
slowNums.takeUntil(coroutineContext, stop).collect { println(it) } // let's test it
721721
}
722722
```
723723

@@ -749,9 +749,9 @@ import kotlin.coroutines.*
749749

750750
```kotlin
751751
fun <T> Publisher<Publisher<T>>.merge(context: CoroutineContext) = GlobalScope.publish<T>(context) {
752-
consumeEach { pub -> // for each publisher received on the source channel
752+
collect { pub -> // for each publisher collected
753753
launch { // launch a child coroutine
754-
pub.consumeEach { send(it) } // resend all element from this publisher
754+
pub.collect { send(it) } // resend all element from this publisher
755755
}
756756
}
757757
}
@@ -792,7 +792,7 @@ The test code is to use `merge` on `testPub` and to display the results:
792792

793793
```kotlin
794794
fun main() = runBlocking<Unit> {
795-
testPub().merge(coroutineContext).consumeEach { println(it) } // print the whole stream
795+
testPub().merge(coroutineContext).collect { println(it) } // print the whole stream
796796
}
797797
```
798798

@@ -975,7 +975,7 @@ fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int
975975

976976
fun main() = runBlocking<Unit> {
977977
rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
978-
.consumeEach { println("$it on thread ${Thread.currentThread().name}") }
978+
.collect { println("$it on thread ${Thread.currentThread().name}") }
979979
}
980980
```
981981

@@ -1021,7 +1021,7 @@ fun rangeWithIntervalRx(scheduler: Scheduler, time: Long, start: Int, count: Int
10211021
fun main() = runBlocking<Unit> {
10221022
val job = launch(Dispatchers.Unconfined) { // launch a new coroutine in Unconfined context (without its own thread pool)
10231023
rangeWithIntervalRx(Schedulers.computation(), 100, 1, 3)
1024-
.consumeEach { println("$it on thread ${Thread.currentThread().name}") }
1024+
.collect { println("$it on thread ${Thread.currentThread().name}") }
10251025
}
10261026
job.join() // wait for our coroutine to complete
10271027
}
@@ -1077,8 +1077,7 @@ coroutines for complex pipelines with fan-in and fan-out between multiple worker
10771077
<!--- MODULE kotlinx-coroutines-reactive -->
10781078
<!--- INDEX kotlinx.coroutines.reactive -->
10791079
[publish]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/kotlinx.coroutines.-coroutine-scope/publish.html
1080-
[org.reactivestreams.Publisher.consumeEach]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/consume-each.html
1081-
[org.reactivestreams.Publisher.openSubscription]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/open-subscription.html
1080+
[org.reactivestreams.Publisher.collect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/collect.html
10821081
<!--- MODULE kotlinx-coroutines-rx2 -->
10831082
<!--- INDEX kotlinx.coroutines.rx2 -->
10841083
[rxFlowable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.rx2/kotlinx.coroutines.-coroutine-scope/rx-flowable.html

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

+8-1
Original file line numberDiff line numberDiff line change
@@ -27,10 +27,17 @@ public fun <T> Publisher<T>.openSubscription(request: Int = 0): ReceiveChannel<T
2727
return channel
2828
}
2929

30+
// Will be promoted to error in 1.3.0, removed in 1.4.0
31+
@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
32+
public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) =
33+
openSubscription().consumeEach(action)
34+
3035
/**
3136
* Subscribes to this [Publisher] and performs the specified action for each received element.
37+
* Cancels subscription if any exception happens during collect.
3238
*/
33-
public suspend inline fun <T> Publisher<T>.consumeEach(action: (T) -> Unit) =
39+
@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
40+
public suspend inline fun <T> Publisher<T>.collect(action: (T) -> Unit) =
3441
openSubscription().consumeEach(action)
3542

3643
@Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")

‎reactive/kotlinx-coroutines-reactive/test/IntegrationTest.kt

+3-3
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ class IntegrationTest(
5050
assertNSE { pub.awaitLast() }
5151
assertNSE { pub.awaitSingle() }
5252
var cnt = 0
53-
pub.consumeEach { cnt++ }
53+
pub.collect { cnt++ }
5454
assertThat(cnt, IsEqual(0))
5555
}
5656

@@ -67,7 +67,7 @@ class IntegrationTest(
6767
assertThat(pub.awaitLast(), IsEqual("OK"))
6868
assertThat(pub.awaitSingle(), IsEqual("OK"))
6969
var cnt = 0
70-
pub.consumeEach {
70+
pub.collect {
7171
assertThat(it, IsEqual("OK"))
7272
cnt++
7373
}
@@ -125,7 +125,7 @@ class IntegrationTest(
125125

126126
private suspend fun checkNumbers(n: Int, pub: Publisher<Int>) {
127127
var last = 0
128-
pub.consumeEach {
128+
pub.collect {
129129
assertThat(it, IsEqual(++last))
130130
}
131131
assertThat(last, IsEqual(n))

‎reactive/kotlinx-coroutines-reactive/test/PublishTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -262,7 +262,7 @@ class PublishTest : TestBase() {
262262
}
263263
}
264264
try {
265-
pub.consumeEach {
265+
pub.collect {
266266
throw TestException()
267267
}
268268
} catch (e: TestException) {

‎reactive/kotlinx-coroutines-reactive/test/PublisherCompletionStressTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ class PublisherCompletionStressTest : TestBase() {
2424
runBlocking {
2525
withTimeout(5000) {
2626
var received = 0
27-
range(Dispatchers.Default, 1, count).consumeEach { x ->
27+
range(Dispatchers.Default, 1, count).collect { x ->
2828
received++
2929
if (x != received) error("$x != $received")
3030
}

‎reactive/kotlinx-coroutines-reactive/test/PublisherMultiTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ class PublisherMultiTest : TestBase() {
2424
jobs.forEach { it.join() }
2525
}
2626
val resultSet = mutableSetOf<Int>()
27-
observable.consumeEach {
27+
observable.collect {
2828
assertTrue(resultSet.add(it))
2929
}
3030
assertThat(resultSet.size, IsEqual(n))

‎reactive/kotlinx-coroutines-reactor/test/FluxCompletionStressTest.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ class FluxCompletionStressTest : TestBase() {
2525
runBlocking {
2626
withTimeout(5000) {
2727
var received = 0
28-
range(Dispatchers.Default, 1, count).consumeEach { x ->
28+
range(Dispatchers.Default, 1, count).collect { x ->
2929
received++
3030
if (x != received) error("$x != $received")
3131
}

‎reactive/kotlinx-coroutines-reactor/test/FluxMultiTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ class FluxMultiTest : TestBase() {
4646
fun testIteratorResendUnconfined() {
4747
val n = 10_000 * stressTestMultiplier
4848
val flux = GlobalScope.flux(Dispatchers.Unconfined) {
49-
Flux.range(0, n).consumeEach { send(it) }
49+
Flux.range(0, n).collect { send(it) }
5050
}
5151
checkMonoValue(flux.collectList()) { list ->
5252
assertEquals((0 until n).toList(), list)
@@ -57,7 +57,7 @@ class FluxMultiTest : TestBase() {
5757
fun testIteratorResendPool() {
5858
val n = 10_000 * stressTestMultiplier
5959
val flux = GlobalScope.flux {
60-
Flux.range(0, n).consumeEach { send(it) }
60+
Flux.range(0, n).collect { send(it) }
6161
}
6262
checkMonoValue(flux.collectList()) { list ->
6363
assertEquals((0 until n).toList(), list)

‎reactive/kotlinx-coroutines-reactor/test/FluxSingleTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,7 @@ class FluxSingleTest {
180180
fun testFluxIteration() {
181181
val flux = GlobalScope.flux {
182182
var result = ""
183-
Flux.just("O", "K").consumeEach { result += it }
183+
Flux.just("O", "K").collect { result += it }
184184
send(result)
185185
}
186186

@@ -193,7 +193,7 @@ class FluxSingleTest {
193193
fun testFluxIterationFailure() {
194194
val flux = GlobalScope.flux {
195195
try {
196-
Flux.error<String>(RuntimeException("OK")).consumeEach { fail("Should not be here") }
196+
Flux.error<String>(RuntimeException("OK")).collect { fail("Should not be here") }
197197
send("Fail")
198198
} catch (e: RuntimeException) {
199199
send(e.message!!)

‎reactive/kotlinx-coroutines-reactor/test/FluxTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ class FluxTest : TestBase() {
107107
expect(2)
108108
val job = launch(start = CoroutineStart.UNDISPATCHED) {
109109
expect(3)
110-
observable.consumeEach {
110+
observable.collect {
111111
expect(8)
112112
assertEquals("OK", it)
113113
}
@@ -131,7 +131,7 @@ class FluxTest : TestBase() {
131131
}
132132
}
133133
try {
134-
pub.consumeEach {
134+
pub.collect {
135135
throw TestException()
136136
}
137137
} catch (e: TestException) {

‎reactive/kotlinx-coroutines-rx2/src/RxChannel.kt

+16-2
Original file line numberDiff line numberDiff line change
@@ -41,16 +41,30 @@ public fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
4141
return channel
4242
}
4343

44+
// Will be promoted to error in 1.3.0, removed in 1.4.0
45+
@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
46+
public suspend inline fun <T> MaybeSource<T>.consumeEach(action: (T) -> Unit) =
47+
openSubscription().consumeEach(action)
48+
49+
// Will be promoted to error in 1.3.0, removed in 1.4.0
50+
@Deprecated(message = "Use collect instead", level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.collect(action)"))
51+
public suspend inline fun <T> ObservableSource<T>.consumeEach(action: (T) -> Unit) =
52+
openSubscription().consumeEach(action)
53+
4454
/**
4555
* Subscribes to this [MaybeSource] and performs the specified action for each received element.
56+
* Cancels subscription if any exception happens during collect.
4657
*/
47-
public suspend inline fun <T> MaybeSource<T>.consumeEach(action: (T) -> Unit) =
58+
@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
59+
public suspend inline fun <T> MaybeSource<T>.collect(action: (T) -> Unit) =
4860
openSubscription().consumeEach(action)
4961

5062
/**
5163
* Subscribes to this [ObservableSource] and performs the specified action for each received element.
64+
* Cancels subscription if any exception happens during collect.
5265
*/
53-
public suspend inline fun <T> ObservableSource<T>.consumeEach(action: (T) -> Unit) =
66+
@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
67+
public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit) =
5468
openSubscription().consumeEach(action)
5569

5670
@Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER")

‎reactive/kotlinx-coroutines-rx2/test/FlowAsObservableTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ class FlowAsObservableTest : TestBase() {
8282
expect(1)
8383
val job = launch(start = CoroutineStart.UNDISPATCHED) {
8484
expect(2)
85-
observable.consumeEach {
85+
observable.collect {
8686
expect(5)
8787
assertEquals("OK", it)
8888
}
@@ -106,7 +106,7 @@ class FlowAsObservableTest : TestBase() {
106106
}.asObservable()
107107

108108
try {
109-
observable.consumeEach {
109+
observable.collect {
110110
expect(3)
111111
throw TestException()
112112
}

‎reactive/kotlinx-coroutines-rx2/test/FlowableTest.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ class FlowableTest : TestBase() {
107107
expect(2)
108108
val job = launch(start = CoroutineStart.UNDISPATCHED) {
109109
expect(3)
110-
observable.consumeEach{
110+
observable.collect {
111111
expect(8)
112112
assertEquals("OK", it)
113113
}
@@ -131,7 +131,7 @@ class FlowableTest : TestBase() {
131131
}
132132
}
133133
try {
134-
pub.consumeEach {
134+
pub.collect {
135135
throw TestException()
136136
}
137137
} catch (e: TestException) {

0 commit comments

Comments
 (0)