blob: 8df6caeeed8153790b42cac9972a12f05d23bcf1 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
/*
* Copyright 2016-2018 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.channels.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactive.*
import kotlin.coroutines.*
/**
* Converts this job to the hot reactive completable that signals
* with [onCompleted][CompletableSubscriber.onCompleted] when the corresponding job completes.
*
* Every subscriber gets the signal at the same time.
* Unsubscribing from the resulting completable **does not** affect the original job in any way.
*
* **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
* in the future to account for the concept of structured concurrency.
*
* @param context -- the coroutine context from which the resulting completable is going to be signalled
*/
@ExperimentalCoroutinesApi
public fun Job.asCompletable(context: CoroutineContext): Completable = rxCompletable(context) {
this@asCompletable.join()
}
/**
* Converts this deferred value to the hot reactive maybe that signals
* [onComplete][MaybeEmitter.onComplete], [onSuccess][MaybeEmitter.onSuccess] or [onError][MaybeEmitter.onError].
*
* Every subscriber gets the same completion value.
* Unsubscribing from the resulting maybe **does not** affect the original deferred value in any way.
*
* **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
* in the future to account for the concept of structured concurrency.
*
* @param context -- the coroutine context from which the resulting maybe is going to be signalled
*/
@ExperimentalCoroutinesApi
public fun <T> Deferred<T?>.asMaybe(context: CoroutineContext): Maybe<T> = rxMaybe(context) {
this@asMaybe.await()
}
/**
* Converts this deferred value to the hot reactive single that signals either
* [onSuccess][SingleSubscriber.onSuccess] or [onError][SingleSubscriber.onError].
*
* Every subscriber gets the same completion value.
* Unsubscribing from the resulting single **does not** affect the original deferred value in any way.
*
* **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
* in the future to account for the concept of structured concurrency.
*
* @param context -- the coroutine context from which the resulting single is going to be signalled
*/
@ExperimentalCoroutinesApi
public fun <T : Any> Deferred<T>.asSingle(context: CoroutineContext): Single<T> = rxSingle(context) {
this@asSingle.await()
}
/**
* Converts a stream of elements received from the channel to the hot reactive observable.
*
* Every subscriber receives values from this channel in **fan-out** fashion. If the are multiple subscribers,
* they'll receive values in round-robin way.
*/
@Deprecated(
message = "Deprecated in the favour of Flow",
level = DeprecationLevel.WARNING, replaceWith = ReplaceWith("this.consumeAsFlow().asObservable()")
)
public fun <T : Any> ReceiveChannel<T>.asObservable(context: CoroutineContext): Observable<T> = rxObservable(context) {
for (t in this@asObservable)
send(t)
}
/**
* Converts the given flow to a cold observable.
* The original flow is cancelled when the observable subscriber is disposed.
*/
@JvmName("from")
@ExperimentalCoroutinesApi
public fun <T: Any> Flow<T>.asObservable() : Observable<T> = Observable.create { emitter ->
/*
* ATOMIC is used here to provide stable behaviour of subscribe+dispose pair even if
* asObservable is already invoked from unconfined
*/
val job = GlobalScope.launch(Dispatchers.Unconfined, start = CoroutineStart.ATOMIC) {
try {
collect { value -> emitter.onNext(value) }
emitter.onComplete()
} catch (e: Throwable) {
// 'create' provides safe emitter, so we can unconditionally call on* here if exception occurs in `onComplete`
if (e !is CancellationException) emitter.onError(e)
else emitter.onComplete()
}
}
emitter.setCancellable(RxCancellable(job))
}
/**
* Converts the given flow to a cold flowable.
* The original flow is cancelled when the flowable subscriber is disposed.
*/
@JvmName("from")
@ExperimentalCoroutinesApi
public fun <T: Any> Flow<T>.asFlowable(): Flowable<T> = Flowable.fromPublisher(asPublisher())
|
