<lambda>null1 package kotlinx.coroutines.channels
2
3 import kotlinx.coroutines.*
4 import kotlin.coroutines.*
5 import kotlinx.coroutines.flow.*
6
7 /**
8 * Scope for the [produce][CoroutineScope.produce], [callbackFlow] and [channelFlow] builders.
9 */
10 public interface ProducerScope<in E> : CoroutineScope, SendChannel<E> {
11 /**
12 * A reference to the channel this coroutine [sends][send] elements to.
13 * It is provided for convenience, so that the code in the coroutine can refer
14 * to the channel as `channel` as opposed to `this`.
15 * All the [SendChannel] functions on this interface delegate to
16 * the channel instance returned by this property.
17 */
18 public val channel: SendChannel<E>
19 }
20
21 /**
22 * Suspends the current coroutine until the channel is either [closed][SendChannel.close] or [cancelled][ReceiveChannel.cancel]
23 * and invokes the given [block] before resuming the coroutine.
24 *
25 * This suspending function is cancellable: if the [Job] of the current coroutine is cancelled while this
26 * suspending function is waiting, this function immediately resumes with [CancellationException].
27 * There is a **prompt cancellation guarantee**: even if this function is ready to return, but was cancelled
28 * while suspended, [CancellationException] will be thrown. See [suspendCancellableCoroutine] for low-level details.
29 *
30 * Note that when the producer channel is cancelled, this function resumes with a cancellation exception.
31 * Therefore, in case of cancellation, no code after the call to this function will be executed.
32 * That's why this function takes a lambda parameter.
33 *
34 * Example of usage:
35 * ```
36 * val callbackEventsStream = produce {
37 * val disposable = registerChannelInCallback(channel)
38 * awaitClose { disposable.dispose() }
39 * }
40 * ```
41 */
<lambda>null42 public suspend fun ProducerScope<*>.awaitClose(block: () -> Unit = {}) {
<lambda>null43 check(kotlin.coroutines.coroutineContext[Job] === this) { "awaitClose() can only be invoked from the producer context" }
44 try {
contnull45 suspendCancellableCoroutine<Unit> { cont ->
46 invokeOnClose {
47 cont.resume(Unit)
48 }
49 }
50 } finally {
51 block()
52 }
53 }
54
55 /**
56 * Launches a new coroutine to produce a stream of values by sending them to a channel
57 * and returns a reference to the coroutine as a [ReceiveChannel]. This resulting
58 * object can be used to [receive][ReceiveChannel.receive] elements produced by this coroutine.
59 *
60 * The scope of the coroutine contains the [ProducerScope] interface, which implements
61 * both [CoroutineScope] and [SendChannel], so that the coroutine can invoke
62 * [send][SendChannel.send] directly. The channel is [closed][SendChannel.close]
63 * when the coroutine completes.
64 * The running coroutine is cancelled when its receive channel is [cancelled][ReceiveChannel.cancel].
65 *
66 * The coroutine context is inherited from this [CoroutineScope]. Additional context elements can be specified with the [context] argument.
67 * If the context does not have any dispatcher or other [ContinuationInterceptor], then [Dispatchers.Default] is used.
68 * The parent job is inherited from the [CoroutineScope] as well, but it can also be overridden
69 * with a corresponding [context] element.
70 *
71 * Any uncaught exception in this coroutine will close the channel with this exception as the cause and
72 * the resulting channel will become _failed_, so that any attempt to receive from it thereafter will throw an exception.
73 *
74 * The kind of the resulting channel depends on the specified [capacity] parameter.
75 * See the [Channel] interface documentation for details.
76 *
77 * See [newCoroutineContext] for a description of debugging facilities available for newly created coroutines.
78 *
79 * **Note: This is an experimental api.** Behaviour of producers that work as children in a parent scope with respect
80 * to cancellation and error handling may change in the future.
81 *
82 * @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
83 * @param capacity capacity of the channel's buffer (no buffer by default).
84 * @param block the coroutine code.
85 */
86 @ExperimentalCoroutinesApi
producenull87 public fun <E> CoroutineScope.produce(
88 context: CoroutineContext = EmptyCoroutineContext,
89 capacity: Int = 0,
90 @BuilderInference block: suspend ProducerScope<E>.() -> Unit
91 ): ReceiveChannel<E> =
92 produce(context, capacity, BufferOverflow.SUSPEND, CoroutineStart.DEFAULT, onCompletion = null, block = block)
93
94 /**
95 * **This is an internal API and should not be used from general code.**
96 * The `onCompletion` parameter will be redesigned.
97 * If you have to use the `onCompletion` operator, please report to https://github.com/Kotlin/kotlinx.coroutines/issues/.
98 * As a temporary solution, [invokeOnCompletion][Job.invokeOnCompletion] can be used instead:
99 * ```
100 * fun <E> ReceiveChannel<E>.myOperator(): ReceiveChannel<E> = GlobalScope.produce(Dispatchers.Unconfined) {
101 * coroutineContext[Job]?.invokeOnCompletion { consumes() }
102 * }
103 * ```
104 * @suppress
105 */
106 @InternalCoroutinesApi
107 public fun <E> CoroutineScope.produce(
108 context: CoroutineContext = EmptyCoroutineContext,
109 capacity: Int = 0,
110 start: CoroutineStart = CoroutineStart.DEFAULT,
111 onCompletion: CompletionHandler? = null,
112 @BuilderInference block: suspend ProducerScope<E>.() -> Unit
113 ): ReceiveChannel<E> =
114 produce(context, capacity, BufferOverflow.SUSPEND, start, onCompletion, block)
115
116 // Internal version of produce that is maximally flexible, but is not exposed through public API (too many params)
117 internal fun <E> CoroutineScope.produce(
118 context: CoroutineContext = EmptyCoroutineContext,
119 capacity: Int = 0,
120 onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
121 start: CoroutineStart = CoroutineStart.DEFAULT,
122 onCompletion: CompletionHandler? = null,
123 @BuilderInference block: suspend ProducerScope<E>.() -> Unit
124 ): ReceiveChannel<E> {
125 val channel = Channel<E>(capacity, onBufferOverflow)
126 val newContext = newCoroutineContext(context)
127 val coroutine = ProducerCoroutine(newContext, channel)
128 if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
129 coroutine.start(start, coroutine, block)
130 return coroutine
131 }
132
133 private class ProducerCoroutine<E>(
134 parentContext: CoroutineContext, channel: Channel<E>
135 ) : ChannelCoroutine<E>(parentContext, channel, true, active = true), ProducerScope<E> {
136 override val isActive: Boolean
137 get() = super.isActive
138
onCompletednull139 override fun onCompleted(value: Unit) {
140 _channel.close()
141 }
142
onCancellednull143 override fun onCancelled(cause: Throwable, handled: Boolean) {
144 val processed = _channel.close(cause)
145 if (!processed && !handled) handleCoroutineException(context, cause)
146 }
147 }
148