Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 1 | /* |
Roman Elizarov | 1f74a2d | 2018-06-29 19:19:45 +0300 | [diff] [blame^] | 2 | * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
| 5 | package kotlinx.coroutines.experimental.rx2 |
| 6 | |
| 7 | import io.reactivex.Flowable |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 8 | import kotlinx.coroutines.experimental.CoroutineDispatcher |
| 9 | import kotlinx.coroutines.experimental.CoroutineScope |
| 10 | import kotlinx.coroutines.experimental.DefaultDispatcher |
| 11 | import kotlinx.coroutines.experimental.Job |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 12 | import kotlinx.coroutines.experimental.channels.ProducerScope |
| 13 | import kotlinx.coroutines.experimental.reactive.publish |
Roman Elizarov | c7d10a4 | 2018-03-13 18:28:42 +0300 | [diff] [blame] | 14 | import kotlin.coroutines.experimental.* |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 15 | |
| 16 | /** |
| 17 | * Creates cold [flowable][Flowable] that will run a given [block] in a coroutine. |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 18 | * Every time the returned flowable is subscribed, it starts a new coroutine. |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 19 | * Coroutine emits items with `send`. Unsubscribing cancels running coroutine. |
| 20 | * |
| 21 | * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that |
| 22 | * `onNext` is not invoked concurrently. |
| 23 | * |
| 24 | * | **Coroutine action** | **Signal to subscriber** |
| 25 | * | -------------------------------------------- | ------------------------ |
| 26 | * | `send` | `onNext` |
| 27 | * | Normal completion or `close` without cause | `onComplete` |
| 28 | * | Failure with exception or `close` with cause | `onError` |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 29 | * |
| 30 | * The [context] for the new coroutine can be explicitly specified. |
| 31 | * See [CoroutineDispatcher] for the standard context implementations that are provided by `kotlinx.coroutines`. |
Roman Elizarov | c7d10a4 | 2018-03-13 18:28:42 +0300 | [diff] [blame] | 32 | * The [coroutineContext] of the parent coroutine may be used, |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 33 | * in which case the [Job] of the resulting coroutine is a child of the job of the parent coroutine. |
| 34 | * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [DefaultDispatcher] is used. |
| 35 | * |
| 36 | * @param context context of the coroutine. The default value is [DefaultDispatcher]. |
| 37 | * @param block the coroutine code. |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 38 | */ |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 39 | @JvmOverloads // for binary compatibility with older code compiled before context had a default |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 40 | public fun <T> rxFlowable( |
Phil Glass | ff850b0 | 2017-10-16 23:31:32 +0100 | [diff] [blame] | 41 | context: CoroutineContext = DefaultDispatcher, |
Roman Elizarov | 50e3221 | 2017-03-10 17:40:50 +0300 | [diff] [blame] | 42 | block: suspend ProducerScope<T>.() -> Unit |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 43 | ): Flowable<T> = Flowable.fromPublisher(publish(context, block = block)) |