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