1 package kotlinx.coroutines.jdk9 2 3 import kotlinx.coroutines.* 4 import kotlinx.coroutines.channels.* 5 import kotlinx.coroutines.reactive.* 6 import java.util.concurrent.* 7 import kotlin.coroutines.* 8 import org.reactivestreams.FlowAdapters 9 10 /** 11 * Creates a cold reactive [Flow.Publisher] that runs a given [block] in a coroutine. 12 * 13 * Every time the returned flux is subscribed, it starts a new coroutine in the specified [context]. 14 * The coroutine emits (via [Flow.Subscriber.onNext]) values with [send][ProducerScope.send], 15 * completes (via [Flow.Subscriber.onComplete]) when the coroutine completes or channel is explicitly closed, and emits 16 * errors (via [Flow.Subscriber.onError]) if the coroutine throws an exception or closes channel with a cause. 17 * Unsubscribing cancels the running coroutine. 18 * 19 * Invocations of [send][ProducerScope.send] are suspended appropriately when subscribers apply back-pressure and to 20 * ensure that [onNext][Flow.Subscriber.onNext] is not invoked concurrently. 21 * 22 * Coroutine context can be specified with [context] argument. 23 * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is 24 * used. 25 * 26 * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect 27 * to cancellation and error handling may change in the future. 28 * 29 * @throws IllegalArgumentException if the provided [context] contains a [Job] instance. 30 */ flowPublishnull31public fun <T> flowPublish( 32 context: CoroutineContext = EmptyCoroutineContext, 33 @BuilderInference block: suspend ProducerScope<T>.() -> Unit 34 ): Flow.Publisher<T> = FlowAdapters.toFlowPublisher(publish(context, block)) 35