Konrad Kamiński | 3ae898c | 2017-03-30 17:37:00 +0200 | [diff] [blame] | 1 | package kotlinx.coroutines.experimental.reactor |
| 2 | |
Francesco Vasco | e19ee04 | 2017-11-29 22:12:35 +0100 | [diff] [blame] | 3 | import kotlinx.coroutines.experimental.DefaultDispatcher |
Konrad Kamiński | 3ae898c | 2017-03-30 17:37:00 +0200 | [diff] [blame] | 4 | import kotlinx.coroutines.experimental.channels.ProducerScope |
| 5 | import kotlinx.coroutines.experimental.reactive.publish |
| 6 | import reactor.core.publisher.Flux |
| 7 | import kotlin.coroutines.experimental.CoroutineContext |
| 8 | |
| 9 | /** |
| 10 | * Creates cold reactive [Flux] that runs a given [block] in a coroutine. |
| 11 | * Every time the returned flux is subscribed, it starts a new coroutine in the specified [context]. |
| 12 | * Coroutine emits items with `send`. Unsubscribing cancels running coroutine. |
| 13 | * |
| 14 | * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that |
| 15 | * `onNext` is not invoked concurrently. |
| 16 | * |
| 17 | * | **Coroutine action** | **Signal to subscriber** |
| 18 | * | -------------------------------------------- | ------------------------ |
| 19 | * | `send` | `onNext` |
| 20 | * | Normal completion or `close` without cause | `onComplete` |
| 21 | * | Failure with exception or `close` with cause | `onError` |
| 22 | */ |
Francesco Vasco | e19ee04 | 2017-11-29 22:12:35 +0100 | [diff] [blame] | 23 | @JvmOverloads // for binary compatibility with older code compiled before context had a default |
Konrad Kamiński | 3ae898c | 2017-03-30 17:37:00 +0200 | [diff] [blame] | 24 | fun <T> flux( |
Francesco Vasco | e19ee04 | 2017-11-29 22:12:35 +0100 | [diff] [blame] | 25 | context: CoroutineContext = DefaultDispatcher, |
Konrad Kamiński | 3ae898c | 2017-03-30 17:37:00 +0200 | [diff] [blame] | 26 | block: suspend ProducerScope<T>.() -> Unit |
Roman Elizarov | e8f694e | 2017-11-28 10:12:00 +0300 | [diff] [blame] | 27 | ): Flux<T> = Flux.from(publish(context, block = block)) |