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