@file:Suppress("FunctionName") package kotlinx.coroutines.channels import kotlinx.coroutines.* import kotlinx.coroutines.channels.Channel.Factory.BUFFERED import kotlinx.coroutines.channels.Channel.Factory.CHANNEL_DEFAULT_CAPACITY import kotlinx.coroutines.channels.Channel.Factory.CONFLATED import kotlinx.coroutines.channels.Channel.Factory.RENDEZVOUS import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED import kotlinx.coroutines.internal.* import kotlinx.coroutines.selects.* import kotlin.contracts.* import kotlin.internal.* import kotlin.jvm.* /** * Sender's interface to a [Channel]. * * Combined, [SendChannel] and [ReceiveChannel] define the complete [Channel] interface. * * It is not expected that this interface will be implemented directly. * Instead, the existing [Channel] implementations can be used or delegated to. */ public interface SendChannel { /** * Returns `true` if this channel was closed by an invocation of [close] or its receiving side was [cancelled][ReceiveChannel.cancel]. * This means that calling [send] will result in an exception. * * Note that if this property returns `false`, it does not guarantee that a subsequent call to [send] will succeed, * as the channel can be concurrently closed right after the check. * For such scenarios, [trySend] is the more robust solution: it attempts to send the element and returns * a result that says whether the channel was closed, and if not, whether sending a value was successful. * * ``` * // DANGER! THIS CHECK IS NOT RELIABLE! * if (!channel.isClosedForSend) { * channel.send(element) // can still fail! * } else { * println("Can not send: the channel is closed") * } * // DO THIS INSTEAD: * channel.trySend(element).onClosed { * println("Can not send: the channel is closed") * } * ``` * * The primary intended usage of this property is skipping some portions of code that should not be executed if the * channel is already known to be closed. * For example: * * ``` * if (channel.isClosedForSend) { * // fast path * return * } else { * // slow path: actually computing the value * val nextElement = run { * // some heavy computation * } * channel.send(nextElement) // can fail anyway, * // but at least we tried to avoid the computation * } * ``` * * However, in many cases, even that can be achieved more idiomatically by cancelling the coroutine producing the * elements to send. * See [produce] for a way to launch a coroutine that produces elements and cancels itself when the channel is * closed. * * [isClosedForSend] can also be used for assertions and diagnostics to verify the expected state of the channel. * * @see SendChannel.trySend * @see SendChannel.close * @see ReceiveChannel.cancel */ @DelicateCoroutinesApi public val isClosedForSend: Boolean /** * Sends the specified [element] to this channel. * * This function suspends if it does not manage to pass the element to the channel's buffer * (or directly the receiving side if there's no buffer), * and it can be cancelled with or without having successfully passed the element. * See the "Suspending and cancellation" section below for details. * If the channel is [closed][close], an exception is thrown (see below). * * ``` * val channel = Channel() * launch { * check(channel.receive() == 5) * } * channel.send(5) // suspends until 5 is received * ``` * * ## Suspending and cancellation * * If the [BufferOverflow] strategy of this channel is [BufferOverflow.SUSPEND], * this function may suspend. * The exact scenarios differ depending on the channel's capacity: * - If the channel is [rendezvous][RENDEZVOUS], * the sender will be suspended until the receiver calls [ReceiveChannel.receive]. * - If the channel is [unlimited][UNLIMITED] or [conflated][CONFLATED], * the sender will never be suspended even with the [BufferOverflow.SUSPEND] strategy. * - If the channel is buffered (either [BUFFERED] or uses a non-default buffer capacity), * the sender will be suspended until the buffer has free space. * * This suspending function is cancellable: if the [Job] of the current coroutine is cancelled while this * suspending function is waiting, this function immediately resumes with [CancellationException]. * There is a **prompt cancellation guarantee**: even if [send] managed to send the element, but was cancelled * while suspended, [CancellationException] will be thrown. See [suspendCancellableCoroutine] for low-level details. * * Because of the prompt cancellation guarantee, an exception does not always mean a failure to deliver the element. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * * Note that this function does not check for cancellation when it is not suspended. * Use [ensureActive] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed: * * ``` * // because of UNLIMITED, sending to this channel never suspends * val channel = Channel(Channel.UNLIMITED) * val job = launch { * while (isActive) { * channel.send(42) * } * // the loop exits when the job is cancelled * } * ``` * * This isn't needed if other cancellable functions are called inside the loop, like [delay]. * * ## Sending to a closed channel * * If a channel was [closed][close] before [send] was called and no cause was specified, * an [ClosedSendChannelException] will be thrown from [send]. * If a channel was [closed][close] with a cause before [send] was called, * then [send] will rethrow the same (in the `===` sense) exception that was passed to [close]. * * In both cases, it is guaranteed that the element was not delivered to the consumer, * and the `onUndeliveredElement` callback will be called. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * * [Closing][close] a channel _after_ this function suspends does not cause this suspended [send] invocation * to abort: although subsequent invocations of [send] fail, the existing ones will continue to completion, * unless the sending coroutine is cancelled. * * ## Related * * This function can be used in [select] invocations with the [onSend] clause. * Use [trySend] to try sending to this channel without waiting and throwing. */ public suspend fun send(element: E) /** * Clause for the [select] expression of the [send] suspending function that selects when the element that is * specified as the parameter is sent to the channel. * When the clause is selected, the reference to this channel is passed into the corresponding block. * * The [select] invocation fails with an exception if the channel [is closed for `send`][isClosedForSend] before * the [select] suspends (see the "Sending to a closed channel" section of [send]). * * Example: * ``` * val sendChannels = List(4) { index -> * Channel(onUndeliveredElement = { * println("Undelivered element $it for $index") * }).also { channel -> * // launch a consumer for this channel * launch { * withTimeout(1.seconds) { * println("Consumer $index receives: ${channel.receive()}") * } * } * } * } * val element = 42 * select { * for (channel in sendChannels) { * channel.onSend(element) { * println("Sent to channel $it") * } * } * } * ``` * Here, we start a [select] expression that waits for exactly one of the four [onSend] invocations * to successfully send the element to the receiver, * and the other three will instead invoke the `onUndeliveredElement` callback. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * * Like [send], [onSend] obeys the rules of prompt cancellation: * [select] may finish with a [CancellationException] even if the element was successfully sent. */ public val onSend: SelectClause2> /** * Attempts to add the specified [element] to this channel without waiting. * * [trySend] never suspends and never throws exceptions. * Instead, it returns a [ChannelResult] that encapsulates the result of the operation. * This makes it different from [send], which can suspend and throw exceptions. * * If this channel is currently full and cannot receive new elements at the time or is [closed][close], * this function returns a result that indicates [a failure][ChannelResult.isFailure]. * In this case, it is guaranteed that the element was not delivered to the consumer and the * `onUndeliveredElement` callback, if one is provided during the [Channel]'s construction, does *not* get called. * * [trySend] can be used as a non-`suspend` alternative to [send] in cases where it's known beforehand * that the channel's buffer can not overflow. * ``` * class Coordinates(val x: Int, val y: Int) * // A channel for a single subscriber that stores the latest mouse position update. * // If more than one subscriber is expected, consider using a `StateFlow` instead. * val mousePositionUpdates = Channel(Channel.CONFLATED) * // Notifies the subscriber about the new mouse position. * // If the subscriber is slow, the intermediate updates are dropped. * fun moveMouse(coordinates: Coordinates) { * val result = mousePositionUpdates.trySend(coordinates) * if (result.isClosed) { * error("Mouse position is no longer being processed") * } * } * ``` */ public fun trySend(element: E): ChannelResult /** * Closes this channel so that subsequent attempts to [send] to it fail. * * Returns `true` if the channel was not closed previously and the call to this function closed it. * If the channel was already closed, this function does nothing and returns `false`. * * The existing elements in the channel remain there, and likewise, * the calls to [send] an [onSend] that have suspended before [close] was called will not be affected. * Only the subsequent calls to [send], [trySend], or [onSend] will fail. * [isClosedForSend] will start returning `true` immediately after this function is called. * * Once all the existing elements are received, the channel will be considered closed for `receive` as well. * This means that [receive][ReceiveChannel.receive] will also start throwing exceptions. * At that point, [isClosedForReceive][ReceiveChannel.isClosedForReceive] will start returning `true`. * * If the [cause] is non-null, it will be thrown from all the subsequent attempts to [send] to this channel, * as well as from all the attempts to [receive][ReceiveChannel.receive] from the channel after no elements remain. * * If the [cause] is null, the channel is considered to have completed normally. * All subsequent calls to [send] will throw a [ClosedSendChannelException], * whereas calling [receive][ReceiveChannel.receive] will throw a [ClosedReceiveChannelException] * after there are no more elements. * * ``` * val channel = Channel() * channel.send(1) * channel.close() * try { * channel.send(2) * error("The channel is closed, so this line is never reached") * } catch (e: ClosedSendChannelException) { * // expected * } * ``` */ public fun close(cause: Throwable? = null): Boolean /** * Registers a [handler] that is synchronously invoked once the channel is [closed][close] * or the receiving side of this channel is [cancelled][ReceiveChannel.cancel]. * Only one handler can be attached to a channel during its lifetime. * The `handler` is invoked when [isClosedForSend] starts to return `true`. * If the channel is closed already, the handler is invoked immediately. * * The meaning of `cause` that is passed to the handler: * - `null` if the channel was [closed][close] normally with `cause = null`. * - Instance of [CancellationException] if the channel was [cancelled][ReceiveChannel.cancel] normally * without the corresponding argument. * - The cause of `close` or `cancel` otherwise. * * ### Execution context and exception safety * * The [handler] is executed as part of the closing or cancelling operation, * and only after the channel reaches its final state. * This means that if the handler throws an exception or hangs, * the channel will still be successfully closed or cancelled. * Unhandled exceptions from [handler] are propagated to the closing or cancelling operation's caller. * * Example of usage: * ``` * val events = Channel(Channel.UNLIMITED) * callbackBasedApi.registerCallback { event -> * events.trySend(event) * .onClosed { /* channel is already closed, but the callback hasn't stopped yet */ } * } * * val uiUpdater = uiScope.launch(Dispatchers.Main) { * events.consume { /* handle events */ } * } * // Stop the callback after the channel is closed or cancelled * events.invokeOnClose { callbackBasedApi.stop() } * ``` * * **Stability note.** This function constitutes a stable API surface, with the only exception being * that an [IllegalStateException] is thrown when multiple handlers are registered. * This restriction could be lifted in the future. * * @throws UnsupportedOperationException if the underlying channel does not support [invokeOnClose]. * Implementation note: currently, [invokeOnClose] is unsupported only by Rx-like integrations. * * @throws IllegalStateException if another handler was already registered */ public fun invokeOnClose(handler: (cause: Throwable?) -> Unit) /** * **Deprecated** offer method. * * This method was deprecated in the favour of [trySend]. * It has proven itself as the most error-prone method in Channel API: * * - `Boolean` return type creates the false sense of security, implying that `false` * is returned instead of throwing an exception. * - It was used mostly from non-suspending APIs where CancellationException triggered * internal failures in the application (the most common source of bugs). * - Due to signature and explicit `if (ch.offer(...))` checks it was easy to * oversee such error during code review. * - Its name was not aligned with the rest of the API and tried to mimic Java's queue instead. * * **NB** Automatic migration provides best-effort for the user experience, but requires removal * or adjusting of the code that relied on the exception handling. * The complete replacement has a more verbose form: * ``` * channel.trySend(element) * .onClosed { throw it ?: ClosedSendChannelException("Channel was closed normally") } * .isSuccess * ``` * * See https://github.com/Kotlin/kotlinx.coroutines/issues/974 for more context. * * @suppress **Deprecated**. */ @Deprecated( level = DeprecationLevel.ERROR, message = "Deprecated in the favour of 'trySend' method", replaceWith = ReplaceWith("trySend(element).isSuccess") ) // Warning since 1.5.0, error since 1.6.0, not hidden until 1.8+ because API is quite widespread public fun offer(element: E): Boolean { val result = trySend(element) if (result.isSuccess) return true throw recoverStackTrace(result.exceptionOrNull() ?: return false) } } /** * Receiver's interface to a [Channel]. * * Combined, [SendChannel] and [ReceiveChannel] define the complete [Channel] interface. */ public interface ReceiveChannel { /** * Returns `true` if the sending side of this channel was [closed][SendChannel.close] * and all previously sent items were already received (which also happens for [cancelled][cancel] channels). * * Note that if this property returns `false`, * it does not guarantee that a subsequent call to [receive] will succeed, * as the channel can be concurrently cancelled or closed right after the check. * For such scenarios, [receiveCatching] is the more robust solution: * if the channel is closed, instead of throwing an exception, [receiveCatching] returns a result that allows * querying it. * * ``` * // DANGER! THIS CHECK IS NOT RELIABLE! * if (!channel.isClosedForReceive) { * channel.receive() // can still fail! * } else { * println("Can not receive: the channel is closed") * null * } * // DO THIS INSTEAD: * channel.receiveCatching().onClosed { * println("Can not receive: the channel is closed") * }.getOrNull() * ``` * * The primary intended usage of this property is for assertions and diagnostics to verify the expected state of * the channel. * Using it in production code is discouraged. * * @see ReceiveChannel.receiveCatching * @see ReceiveChannel.cancel * @see SendChannel.close */ @DelicateCoroutinesApi public val isClosedForReceive: Boolean /** * Returns `true` if the channel contains no elements and isn't [closed for `receive`][isClosedForReceive]. * * If [isEmpty] returns `true`, it means that calling [receive] at exactly the same moment would suspend. * However, calling [receive] immediately after checking [isEmpty] may or may not suspend, as new elements * could have been added or removed or the channel could have been closed for `receive` between the two invocations. * Consider using [tryReceive] in cases when suspensions are undesirable: * * ``` * // DANGER! THIS CHECK IS NOT RELIABLE! * while (!channel.isEmpty) { * // can still suspend if other `receive` happens in parallel! * val element = channel.receive() * println(element) * } * // DO THIS INSTEAD: * while (true) { * val element = channel.tryReceive().getOrNull() ?: break * println(element) * } * ``` */ @ExperimentalCoroutinesApi public val isEmpty: Boolean /** * Retrieves an element, removing it from the channel. * * This function suspends if the channel is empty, waiting until an element is available. * If the channel is [closed for `receive`][isClosedForReceive], an exception is thrown (see below). * ``` * val channel = Channel() * launch { * val element = channel.receive() // suspends until 5 is available * check(element == 5) * } * channel.send(5) * ``` * * ## Suspending and cancellation * * This suspending function is cancellable: if the [Job] of the current coroutine is cancelled while this * suspending function is waiting, this function immediately resumes with [CancellationException]. * There is a **prompt cancellation guarantee**: even if [receive] managed to retrieve the element from the channel, * but was cancelled while suspended, [CancellationException] will be thrown, and, if * the channel has an `onUndeliveredElement` callback installed, the retrieved element will be passed to it. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * See [suspendCancellableCoroutine] for the low-level details of prompt cancellation. * * Note that this function does not check for cancellation when it manages to immediately receive an element without * suspending. * Use [ensureActive] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed: * * ``` * val channel = Channel() * launch { // a very fast producer * while (true) { * channel.send(42) * } * } * val consumer = launch { // a slow consumer * while (isActive) { * val element = channel.receive() * // some slow computation involving `element` * } * } * delay(100.milliseconds) * consumer.cancelAndJoin() * ``` * * ## Receiving from a closed channel * * - Attempting to [receive] from a [closed][SendChannel.close] channel while there are still some elements * will successfully retrieve an element from the channel. * - When a channel is [closed][SendChannel.close] and there are no elements remaining, * the channel becomes [closed for `receive`][isClosedForReceive]. * After that, * [receive] will rethrow the same (in the `===` sense) exception that was passed to [SendChannel.close], * or [ClosedReceiveChannelException] if none was given. * * ## Related * * This function can be used in [select] invocations with the [onReceive] clause. * Use [tryReceive] to try receiving from this channel without waiting and throwing. * Use [receiveCatching] to receive from this channel without throwing. */ public suspend fun receive(): E /** * Clause for the [select] expression of the [receive] suspending function that selects with the element * received from the channel. * * The [select] invocation fails with an exception if the channel [is closed for `receive`][isClosedForReceive] * at any point, even if other [select] clauses could still work. * * Example: * ``` * class ScreenSize(val width: Int, val height: Int) * class MouseClick(val x: Int, val y: Int) * val screenResizes = Channel(Channel.CONFLATED) * val mouseClicks = Channel(Channel.CONFLATED) * * launch(Dispatchers.Main) { * while (true) { * select { * screenResizes.onReceive { newSize -> * // update the UI to the new screen size * } * mouseClicks.onReceive { click -> * // react to a mouse click * } * } * } * } * ``` * * Like [receive], [onReceive] obeys the rules of prompt cancellation: * [select] may finish with a [CancellationException] even if an element was successfully retrieved, * in which case the `onUndeliveredElement` callback will be called. */ public val onReceive: SelectClause1 /** * Retrieves an element, removing it from the channel. * * A difference from [receive] is that this function encapsulates a failure in its return value instead of throwing * an exception. * However, it will still throw [CancellationException] if the coroutine calling [receiveCatching] is cancelled. * * It is guaranteed that the only way this function can return a [failed][ChannelResult.isFailure] result is when * the channel is [closed for `receive`][isClosedForReceive], so [ChannelResult.isClosed] is also true. * * This function suspends if the channel is empty, waiting until an element is available or the channel becomes * closed. * ``` * val channel = Channel() * launch { * while (true) { * val result = channel.receiveCatching() // suspends * when (val element = result.getOrNull()) { * null -> break // the channel is closed * else -> check(element == 5) * } * } * } * channel.send(5) * ``` * * ## Suspending and cancellation * * This suspending function is cancellable: if the [Job] of the current coroutine is cancelled while this * suspending function is waiting, this function immediately resumes with [CancellationException]. * There is a **prompt cancellation guarantee**: even if [receiveCatching] managed to retrieve the element from the * channel, but was cancelled while suspended, [CancellationException] will be thrown, and, if * the channel has an `onUndeliveredElement` callback installed, the retrieved element will be passed to it. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * See [suspendCancellableCoroutine] for the low-level details of prompt cancellation. * * Note that this function does not check for cancellation when it manages to immediately receive an element without * suspending. * Use [ensureActive] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed: * * ``` * val channel = Channel() * launch { // a very fast producer * while (true) { * channel.send(42) * } * } * val consumer = launch { // a slow consumer * while (isActive) { * val element = channel.receiveCatching().getOrNull() ?: break * // some slow computation involving `element` * } * } * delay(100.milliseconds) * consumer.cancelAndJoin() * ``` * * ## Receiving from a closed channel * * - Attempting to [receiveCatching] from a [closed][SendChannel.close] channel while there are still some elements * will successfully retrieve an element from the channel. * - When a channel is [closed][SendChannel.close] and there are no elements remaining, * the channel becomes [closed for `receive`][isClosedForReceive]. * After that, [receiveCatching] will return a result with [ChannelResult.isClosed] set. * [ChannelResult.exceptionOrNull] will be the exact (in the `===` sense) exception * that was passed to [SendChannel.close], * or `null` if none was given. * * ## Related * * This function can be used in [select] invocations with the [onReceiveCatching] clause. * Use [tryReceive] to try receiving from this channel without waiting and throwing. * Use [receive] to receive from this channel and throw exceptions on error. */ public suspend fun receiveCatching(): ChannelResult /** * Clause for the [select] expression of the [receiveCatching] suspending function that selects * with a [ChannelResult] when an element is retrieved or the channel gets closed. * * Like [receiveCatching], [onReceiveCatching] obeys the rules of prompt cancellation: * [select] may finish with a [CancellationException] even if an element was successfully retrieved, * in which case the `onUndeliveredElement` callback will be called. */ // TODO: think of an example of when this could be useful public val onReceiveCatching: SelectClause1> /** * Attempts to retrieve an element without waiting, removing it from the channel. * * - When the channel is non-empty, a [successful][ChannelResult.isSuccess] result is returned, * and [ChannelResult.getOrNull] returns the retrieved element. * - When the channel is empty, a [failed][ChannelResult.isFailure] result is returned. * - When the channel is already [closed for `receive`][isClosedForReceive], * returns the ["channel is closed"][ChannelResult.isClosed] result. * If the channel was [closed][SendChannel.close] with a cause (for example, [cancelled][cancel]), * [ChannelResult.exceptionOrNull] contains the cause. * * This function is useful when implementing on-demand allocation of resources to be stored in the channel: * * ``` * val resourcePool = Channel(maxResources) * * suspend fun withResource(block: (Resource) -> Unit) { * val result = resourcePool.tryReceive() * val resource = result.getOrNull() * ?: tryCreateNewResource() // try to create a new resource * ?: resourcePool.receive() // could not create: actually wait for the resource * try { * block(resource) * } finally { * resourcePool.trySend(resource) * } * } * ``` */ public fun tryReceive(): ChannelResult /** * Returns a new iterator to receive elements from this channel using a `for` loop. * Iteration completes normally when the channel [is closed for `receive`][isClosedForReceive] without a cause and * throws the exception passed to [close][SendChannel.close] if there was one. * * Instances of [ChannelIterator] are not thread-safe and shall not be used from concurrent coroutines. * * Example: * * ``` * val channel = produce { * repeat(1000) { * send(it) * } * } * for (v in channel) { * println(v) * } * ``` * * Note that if an early return happens from the `for` loop, the channel does not get cancelled. * To forbid sending new elements after the iteration is completed, use [consumeEach] or * call [cancel] manually. */ public operator fun iterator(): ChannelIterator /** * [Closes][SendChannel.close] the channel for new elements and removes all existing ones. * * A [cause] can be used to specify an error message or to provide other details on * the cancellation reason for debugging purposes. * If the cause is not specified, then an instance of [CancellationException] with a * default message is created to [close][SendChannel.close] the channel. * * If the channel was already [closed][SendChannel.close], * [cancel] only has the effect of removing all elements from the channel. * * Immediately after the invocation of this function, * [isClosedForReceive] and, on the [SendChannel] side, [isClosedForSend][SendChannel.isClosedForSend] * start returning `true`. * Any attempt to send to or receive from this channel will lead to a [CancellationException]. * This also applies to the existing senders and receivers that are suspended at the time of the call: * they will be resumed with a [CancellationException] immediately after [cancel] is called. * * If the channel has an `onUndeliveredElement` callback installed, this function will invoke it for each of the * elements still in the channel, since these elements will be inaccessible otherwise. * If the callback is not installed, these elements will simply be removed from the channel for garbage collection. * * ``` * val channel = Channel() * channel.send(1) * channel.send(2) * channel.cancel() * channel.trySend(3) // returns ChannelResult.isClosed * for (element in channel) { println(element) } // prints nothing * ``` * * [consume] and [consumeEach] are convenient shorthands for cancelling the channel after the single consumer * has finished processing. */ public fun cancel(cause: CancellationException? = null) /** * @suppress This method implements old version of JVM ABI. Use [cancel]. */ @Deprecated(level = DeprecationLevel.HIDDEN, message = "Since 1.2.0, binary compatibility with versions <= 1.1.x") public fun cancel(): Unit = cancel(null) /** * @suppress This method has bad semantics when cause is not a [CancellationException]. Use [cancel]. */ @Deprecated(level = DeprecationLevel.HIDDEN, message = "Since 1.2.0, binary compatibility with versions <= 1.1.x") public fun cancel(cause: Throwable? = null): Boolean /** * **Deprecated** poll method. * * This method was deprecated in the favour of [tryReceive]. * It has proven itself as error-prone method in Channel API: * * - Nullable return type creates the false sense of security, implying that `null` * is returned instead of throwing an exception. * - It was used mostly from non-suspending APIs where CancellationException triggered * internal failures in the application (the most common source of bugs). * - Its name was not aligned with the rest of the API and tried to mimic Java's queue instead. * * See https://github.com/Kotlin/kotlinx.coroutines/issues/974 for more context. * * ### Replacement note * * The replacement `tryReceive().getOrNull()` is a default that ignores all close exceptions and * proceeds with `null`, while `poll` throws an exception if the channel was closed with an exception. * Replacement with the very same 'poll' semantics is `tryReceive().onClosed { if (it != null) throw it }.getOrNull()` * * @suppress **Deprecated**. */ @Deprecated( level = DeprecationLevel.ERROR, message = "Deprecated in the favour of 'tryReceive'. " + "Please note that the provided replacement does not rethrow channel's close cause as 'poll' did, " + "for the precise replacement please refer to the 'poll' documentation", replaceWith = ReplaceWith("tryReceive().getOrNull()") ) // Warning since 1.5.0, error since 1.6.0, not hidden until 1.8+ because API is quite widespread public fun poll(): E? { val result = tryReceive() if (result.isSuccess) return result.getOrThrow() throw recoverStackTrace(result.exceptionOrNull() ?: return null) } /** * This function was deprecated since 1.3.0 and is no longer recommended to use * or to implement in subclasses. * * It had the following pitfalls: * - Didn't allow to distinguish 'null' as "closed channel" from "null as a value" * - Was throwing if the channel has failed even though its signature may suggest it returns 'null' * - It didn't really belong to core channel API and can be exposed as an extension instead. * * ### Replacement note * * The replacement `receiveCatching().getOrNull()` is a safe default that ignores all close exceptions and * proceeds with `null`, while `receiveOrNull` throws an exception if the channel was closed with an exception. * Replacement with the very same `receiveOrNull` semantics is `receiveCatching().onClosed { if (it != null) throw it }.getOrNull()`. * * @suppress **Deprecated** */ @Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER") @LowPriorityInOverloadResolution @Deprecated( message = "Deprecated in favor of 'receiveCatching'. " + "Please note that the provided replacement does not rethrow channel's close cause as 'receiveOrNull' did, " + "for the detailed replacement please refer to the 'receiveOrNull' documentation", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("receiveCatching().getOrNull()") ) // Warning since 1.3.0, error in 1.5.0, cannot be hidden due to deprecated extensions public suspend fun receiveOrNull(): E? = receiveCatching().getOrNull() /** * This function was deprecated since 1.3.0 and is no longer recommended to use * or to implement in subclasses. * See [receiveOrNull] documentation. * * @suppress **Deprecated**: in favor of onReceiveCatching extension. */ @Suppress("DEPRECATION_ERROR") @Deprecated( message = "Deprecated in favor of onReceiveCatching extension", level = DeprecationLevel.ERROR, replaceWith = ReplaceWith("onReceiveCatching") ) // Warning since 1.3.0, error in 1.5.0, will be hidden or removed in 1.7.0 public val onReceiveOrNull: SelectClause1 get() = (this as BufferedChannel).onReceiveOrNull } /** * A discriminated union representing a channel operation result. * It encapsulates the knowledge of whether the operation succeeded, failed with an option to retry, * or failed because the channel was closed. * * If the operation was [successful][isSuccess], [T] is the result of the operation: * for example, for [ReceiveChannel.receiveCatching] and [ReceiveChannel.tryReceive], * it is the element received from the channel, and for [Channel.trySend], it is [Unit], * as the channel does not receive anything in return for sending a channel. * This value can be retrieved with [getOrNull] or [getOrThrow]. * * If the operation [failed][isFailure], it does not necessarily mean that the channel itself is closed. * For example, [ReceiveChannel.receiveCatching] and [ReceiveChannel.tryReceive] can fail because the channel is empty, * and [Channel.trySend] can fail because the channel is full. * * If the operation [failed][isFailure] because the channel was closed for that operation, [isClosed] returns `true`. * The opposite is also true: if [isClosed] returns `true`, then the channel is closed for that operation * ([ReceiveChannel.isClosedForReceive] or [SendChannel.isClosedForSend]). * In this case, retrying the operation is meaningless: once closed, the channel will remain closed. * The [exceptionOrNull] function returns the reason the channel was closed, if any was given. * * Manually obtaining a [ChannelResult] instance is not supported. * See the documentation for [ChannelResult]-returning functions for usage examples. */ @JvmInline public value class ChannelResult @PublishedApi internal constructor(@PublishedApi internal val holder: Any?) { /** * Whether the operation succeeded. * * If this returns `true`, the operation was successful. * In this case, [getOrNull] and [getOrThrow] can be used to retrieve the value. * * If this returns `false`, the operation failed. * [isClosed] can be used to determine whether the operation failed because the channel was closed * (and therefore retrying the operation is meaningless). * * ``` * val result = channel.tryReceive() * if (result.isSuccess) { * println("Successfully received the value ${result.getOrThrow()}") * } else { * println("Failed to receive the value.") * if (result.isClosed) { * println("The channel is closed.") * if (result.exceptionOrNull() != null) { * println("The reason: ${result.exceptionOrNull()}") * } * } * } * ``` * * [isFailure] is a shorthand for `!isSuccess`. * [getOrNull] can simplify [isSuccess] followed by [getOrThrow] into just one check if [T] is known * to be non-nullable. */ public val isSuccess: Boolean get() = holder !is Failed /** * Whether the operation failed. * * A shorthand for `!isSuccess`. See [isSuccess] for more details. */ public val isFailure: Boolean get() = holder is Failed /** * Whether the operation failed because the channel was closed. * * If this returns `true`, the channel was closed for the operation that returned this result. * In this case, retrying the operation is meaningless: once closed, the channel will remain closed. * [isSuccess] will return `false`. * [exceptionOrNull] can be used to determine the reason the channel was [closed][SendChannel.close] * if one was given. * * If this returns `false`, subsequent attempts to perform the same operation may succeed. * * ``` * val result = channel.trySend(42) * if (result.isClosed) { * println("The channel is closed.") * if (result.exceptionOrNull() != null) { * println("The reason: ${result.exceptionOrNull()}") * } * } */ public val isClosed: Boolean get() = holder is Closed /** * Returns the encapsulated [T] if the operation succeeded, or `null` if it failed. * * For non-nullable [T], the following code can be used to handle the result: * ``` * val result = channel.tryReceive() * val value = result.getOrNull() * if (value == null) { * if (result.isClosed) { * println("The channel is closed.") * if (result.exceptionOrNull() != null) { * println("The reason: ${result.exceptionOrNull()}") * } * } * return * } * println("Successfully received the value $value") * ``` * * If [T] is nullable, [getOrThrow] together with [isSuccess] is a more reliable way to handle the result. */ @Suppress("UNCHECKED_CAST") public fun getOrNull(): T? = if (holder !is Failed) holder as T else null /** * Returns the encapsulated [T] if the operation succeeded, or throws the encapsulated exception if it failed. * * Example: * ``` * val result = channel.tryReceive() * if (result.isSuccess) { * println("Successfully received the value ${result.getOrThrow()}") * } * ``` * * @throws IllegalStateException if the operation failed, but the channel was not closed with a cause. */ public fun getOrThrow(): T { @Suppress("UNCHECKED_CAST") if (holder !is Failed) return holder as T if (holder is Closed) { check(holder.cause != null) { "Trying to call 'getOrThrow' on a channel closed without a cause" } throw holder.cause } error("Trying to call 'getOrThrow' on a failed result of a non-closed channel") } /** * Returns the exception with which the channel was closed, or `null` if the channel was not closed or was closed * without a cause. * * [exceptionOrNull] can only return a non-`null` value if [isClosed] is `true`, * but even if [isClosed] is `true`, * [exceptionOrNull] can still return `null` if the channel was closed without a cause. * * ``` * val result = channel.tryReceive() * if (result.isClosed) { * // Now we know not to retry the operation later. * // Check if the channel was closed with a cause and rethrow the exception: * result.exceptionOrNull()?.let { throw it } * // Otherwise, the channel was closed without a cause. * } * ``` */ public fun exceptionOrNull(): Throwable? = (holder as? Closed)?.cause internal open class Failed { override fun toString(): String = "Failed" } internal class Closed(@JvmField val cause: Throwable?): Failed() { override fun equals(other: Any?): Boolean = other is Closed && cause == other.cause override fun hashCode(): Int = cause.hashCode() override fun toString(): String = "Closed($cause)" } /** * @suppress **This is internal API and it is subject to change.** */ @InternalCoroutinesApi public companion object { private val failed = Failed() @InternalCoroutinesApi public fun success(value: E): ChannelResult = ChannelResult(value) @InternalCoroutinesApi public fun failure(): ChannelResult = ChannelResult(failed) @InternalCoroutinesApi public fun closed(cause: Throwable?): ChannelResult = ChannelResult(Closed(cause)) } public override fun toString(): String = when (holder) { is Closed -> holder.toString() else -> "Value($holder)" } } /** * Returns the encapsulated value if the operation [succeeded][ChannelResult.isSuccess], or the * result of [onFailure] function for [ChannelResult.exceptionOrNull] otherwise. * * A shorthand for `if (isSuccess) getOrNull() else onFailure(exceptionOrNull())`. * * @see ChannelResult.getOrNull * @see ChannelResult.exceptionOrNull */ @OptIn(ExperimentalContracts::class) public inline fun ChannelResult.getOrElse(onFailure: (exception: Throwable?) -> T): T { contract { callsInPlace(onFailure, InvocationKind.AT_MOST_ONCE) } @Suppress("UNCHECKED_CAST") return if (holder is ChannelResult.Failed) onFailure(exceptionOrNull()) else holder as T } /** * Performs the given [action] on the encapsulated value if the operation [succeeded][ChannelResult.isSuccess]. * Returns the original `ChannelResult` unchanged. * * A shorthand for `this.also { if (isSuccess) action(getOrThrow()) }`. */ @OptIn(ExperimentalContracts::class) public inline fun ChannelResult.onSuccess(action: (value: T) -> Unit): ChannelResult { contract { callsInPlace(action, InvocationKind.AT_MOST_ONCE) } @Suppress("UNCHECKED_CAST") if (holder !is ChannelResult.Failed) action(holder as T) return this } /** * Performs the given [action] if the operation [failed][ChannelResult.isFailure]. * The result of [ChannelResult.exceptionOrNull] is passed to the [action] parameter. * * Returns the original `ChannelResult` unchanged. * * A shorthand for `this.also { if (isFailure) action(exceptionOrNull()) }`. */ @OptIn(ExperimentalContracts::class) public inline fun ChannelResult.onFailure(action: (exception: Throwable?) -> Unit): ChannelResult { contract { callsInPlace(action, InvocationKind.AT_MOST_ONCE) } if (holder is ChannelResult.Failed) action(exceptionOrNull()) return this } /** * Performs the given [action] if the operation failed because the channel was [closed][ChannelResult.isClosed] for * that operation. * The result of [ChannelResult.exceptionOrNull] is passed to the [action] parameter. * * It is guaranteed that if action is invoked, then the channel is either [closed for send][Channel.isClosedForSend] * or is [closed for receive][Channel.isClosedForReceive] depending on the failed operation. * * Returns the original `ChannelResult` unchanged. * * A shorthand for `this.also { if (isClosed) action(exceptionOrNull()) }`. */ @OptIn(ExperimentalContracts::class) public inline fun ChannelResult.onClosed(action: (exception: Throwable?) -> Unit): ChannelResult { contract { callsInPlace(action, InvocationKind.AT_MOST_ONCE) } if (holder is ChannelResult.Closed) action(exceptionOrNull()) return this } /** * Iterator for a [ReceiveChannel]. * Instances of this interface are *not thread-safe* and shall not be used from concurrent coroutines. */ public interface ChannelIterator { /** * Prepare an element for retrieval by the invocation of [next]. * * - If the element that was retrieved by an earlier [hasNext] call was not yet consumed by [next], returns `true`. * - If the channel has an element available, returns `true` and removes it from the channel. * This element will be returned by the subsequent invocation of [next]. * - If the channel is [closed for receiving][ReceiveChannel.isClosedForReceive] without a cause, returns `false`. * - If the channel is closed with a cause, throws the original [close][SendChannel.close] cause exception. * - If the channel is not closed but does not contain an element, * suspends until either an element is sent to the channel or the channel gets closed. * * This suspending function is cancellable: if the [Job] of the current coroutine is cancelled while this * suspending function is waiting, this function immediately resumes with [CancellationException]. * There is a **prompt cancellation guarantee**: even if [hasNext] retrieves the element from the channel during * its operation, but was cancelled while suspended, [CancellationException] will be thrown. * See [suspendCancellableCoroutine] for low-level details. * * Because of the prompt cancellation guarantee, some values retrieved from the channel can become lost. * See the "Undelivered elements" section in the [Channel] documentation * for details on handling undelivered elements. * * Note that this function does not check for cancellation when it is not suspended, that is, * if the next element is immediately available. * Use [ensureActive] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed. */ public suspend operator fun hasNext(): Boolean @Deprecated(message = "Since 1.3.0, binary compatibility with versions <= 1.2.x", level = DeprecationLevel.HIDDEN) @Suppress("INAPPLICABLE_JVM_NAME") @JvmName("next") public suspend fun next0(): E { /* * Before 1.3.0 the "next()" could have been used without invoking "hasNext" first and there were code samples * demonstrating this behavior, so we preserve this logic for full binary backwards compatibility with previously * compiled code. */ if (!hasNext()) throw ClosedReceiveChannelException(DEFAULT_CLOSE_MESSAGE) return next() } /** * Retrieves the element removed from the channel by the preceding call to [hasNext], or * throws an [IllegalStateException] if [hasNext] was not invoked. * * This method can only be used together with [hasNext]: * ``` * while (iterator.hasNext()) { * val element = iterator.next() * // ... handle the element ... * } * ``` * * A more idiomatic way to iterate over a channel is to use a `for` loop: * ``` * for (element in channel) { * // ... handle the element ... * } * ``` * * This method never throws if [hasNext] returned `true`. * If [hasNext] threw the cause with which the channel was closed, this method will rethrow the same exception. * If [hasNext] returned `false` because the channel was closed without a cause, this method throws * a [ClosedReceiveChannelException]. */ public operator fun next(): E } /** * Channel is a non-blocking primitive for communication between a sender (via [SendChannel]) and a receiver (via [ReceiveChannel]). * Conceptually, a channel is similar to `java.util.concurrent.BlockingQueue`, * but it has suspending operations instead of blocking ones and can be [closed][SendChannel.close]. * * ### Channel capacity * * Most ways to create a [Channel] (in particular, the `Channel()` factory function) allow specifying a capacity, * which determines how elements are buffered in the channel. * There are several predefined constants for the capacity that have special behavior: * * - [Channel.RENDEZVOUS] (or 0) creates a _rendezvous_ channel, which does not have a buffer at all. * Instead, the sender and the receiver must rendezvous (meet): * [SendChannel.send] suspends until another coroutine invokes [ReceiveChannel.receive], and vice versa. * - [Channel.CONFLATED] creates a buffer for a single element and automatically changes the * [buffer overflow strategy][BufferOverflow] to [BufferOverflow.DROP_OLDEST]. * - [Channel.UNLIMITED] creates a channel with an unlimited buffer, which never suspends the sender. * - [Channel.BUFFERED] creates a channel with a buffer whose size depends on * the [buffer overflow strategy][BufferOverflow]. * * See each constant's documentation for more details. * * If the capacity is positive but less than [Channel.UNLIMITED], the channel has a buffer with the specified capacity. * It is safe to construct a channel with a large buffer, as memory is only allocated gradually as elements are added. * * Constructing a channel with a negative capacity not equal to a predefined constant is not allowed * and throws an [IllegalArgumentException]. * * ### Buffer overflow * * Some ways to create a [Channel] also expose a [BufferOverflow] parameter (by convention, `onBufferOverflow`), * which does not affect the receiver but determines the behavior of the sender when the buffer is full. * The options include [suspending][BufferOverflow.SUSPEND] until there is space in the buffer, * [dropping the oldest element][BufferOverflow.DROP_OLDEST] to make room for the new one, or * [dropping the element to be sent][BufferOverflow.DROP_LATEST]. See the [BufferOverflow] documentation. * * By convention, the default value for [BufferOverflow] whenever it can not be configured is [BufferOverflow.SUSPEND]. * * See the [Channel.RENDEZVOUS], [Channel.CONFLATED], and [Channel.UNLIMITED] documentation for a description of how * they interact with the [BufferOverflow] parameter. * * ### Prompt cancellation guarantee * * All suspending functions with channels provide **prompt cancellation guarantee**. * If the job was cancelled while send or receive function was suspended, it will not resume successfully, even if it * already changed the channel's state, but throws a [CancellationException]. * With a single-threaded [dispatcher][CoroutineDispatcher] like [Dispatchers.Main], this gives a * guarantee that the coroutine promptly reacts to the cancellation of its [Job] and does not resume its execution. * * > **Prompt cancellation guarantee** for channel operations was added in `kotlinx.coroutines` version `1.4.0` * > and has replaced the channel-specific atomic cancellation that was not consistent with other suspending functions. * > The low-level mechanics of prompt cancellation are explained in the [suspendCancellableCoroutine] documentation. * * ### Undelivered elements * * As a result of the prompt cancellation guarantee, when a closeable resource * (like an open file or a handle to another native resource) is transferred via a channel, * it can be successfully extracted from the channel, * but still be lost if the receiving operation is cancelled in parallel. * * The `Channel()` factory function has the optional parameter `onUndeliveredElement`. * When that parameter is set, the corresponding function is called once for each element * that was sent to the channel with the call to the [send][SendChannel.send] function but failed to be delivered, * which can happen in the following cases: * * - When an element is dropped due to the limited buffer capacity. * This can happen when the overflow strategy is [BufferOverflow.DROP_LATEST] or [BufferOverflow.DROP_OLDEST]. * - When the sending operations like [send][SendChannel.send] or [onSend][SendChannel.onSend] * throw an exception because it was cancelled * before it had a chance to actually send the element * or because the channel was [closed][SendChannel.close] or [cancelled][ReceiveChannel.cancel]. * - When the receiving operations like [receive][ReceiveChannel.receive], * [onReceive][ReceiveChannel.onReceive], or [hasNext][ChannelIterator.hasNext] * throw an exception after retrieving the element from the channel * because of being cancelled before the code following them had a chance to resume. * - When the channel was [cancelled][ReceiveChannel.cancel], in which case `onUndeliveredElement` is called on every * remaining element in the channel's buffer. * * Note that `onUndeliveredElement` is called synchronously in an arbitrary context. * It should be fast, non-blocking, and should not throw exceptions. * Any exception thrown by `onUndeliveredElement` is wrapped into an internal runtime exception * which is either rethrown from the caller method or handed off to the exception handler in the current context * (see [CoroutineExceptionHandler]) when one is available. * * A typical usage for `onUndeliveredElement` is to close a resource that is being transferred via the channel. The * following code pattern guarantees that opened resources are closed even if the producer, the consumer, * and/or the channel are cancelled. Resources are never lost. * * ``` * // Create a channel with an onUndeliveredElement block that closes a resource * val channel = Channel(onUndeliveredElement = { resource -> resource.close() }) * * // Producer code * val resourceToSend = openResource() * channel.send(resourceToSend) * * // Consumer code * val resourceReceived = channel.receive() * try { * // work with received resource * } finally { * resourceReceived.close() * } * ``` * * > Note that if any work happens between `openResource()` and `channel.send(...)`, * > it is your responsibility to ensure that resource gets closed in case this additional code fails. */ public interface Channel : SendChannel, ReceiveChannel { /** * Constants for the channel factory function `Channel()`. */ public companion object Factory { /** * An unlimited buffer capacity. * * `Channel(UNLIMITED)` creates a channel with an unlimited buffer, which never suspends the sender. * The total amount of elements that can be sent to the channel is limited only by the available memory. * * If [BufferOverflow] is specified for the channel, it is completely ignored, * as the channel never suspends the sender. * * ``` * val channel = Channel(Channel.UNLIMITED) * repeat(1000) { * channel.trySend(it) * } * repeat(1000) { * check(channel.tryReceive().getOrNull() == it) * } * ``` */ public const val UNLIMITED: Int = Int.MAX_VALUE /** * The zero buffer capacity. * * For the default [BufferOverflow] value of [BufferOverflow.SUSPEND], * `Channel(RENDEZVOUS)` creates a channel without a buffer. * An element is transferred from the sender to the receiver only when [send] and [receive] invocations meet * in time (that is, they _rendezvous_), * so [send] suspends until another coroutine invokes [receive], * and [receive] suspends until another coroutine invokes [send]. * * ``` * val channel = Channel(Channel.RENDEZVOUS) * check(channel.trySend(5).isFailure) // sending fails: no receiver is waiting * launch(start = CoroutineStart.UNDISPATCHED) { * val element = channel.receive() // suspends * check(element == 3) * } * check(channel.trySend(3).isSuccess) // sending succeeds: receiver is waiting * ``` * * If a different [BufferOverflow] is specified, * `Channel(RENDEZVOUS)` creates a channel with a buffer of size 1: * * ``` * val channel = Channel(0, onBufferOverflow = BufferOverflow.DROP_OLDEST) * // None of the calls suspend, since the buffer overflow strategy is not SUSPEND * channel.send(1) * channel.send(2) * channel.send(3) * check(channel.receive() == 3) * ``` */ public const val RENDEZVOUS: Int = 0 /** * A single-element buffer with conflating behavior. * * Specifying [CONFLATED] as the capacity in the `Channel(...)` factory function is equivalent to * creating a channel with a buffer of size 1 and a [BufferOverflow] strategy of [BufferOverflow.DROP_OLDEST]: * `Channel(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)`. * Such a channel buffers at most one element and conflates all subsequent `send` and `trySend` invocations * so that the receiver always gets the last element sent, **losing** the previously sent elements: * see the "Undelivered elements" section in the [Channel] documentation. * [Sending][send] to this channel never suspends, and [trySend] always succeeds. * * ``` * val channel = Channel(Channel.CONFLATED) * channel.send(1) * channel.send(2) * channel.send(3) * check(channel.receive() == 3) * ``` * * Specifying a [BufferOverflow] other than [BufferOverflow.SUSPEND] is not allowed with [CONFLATED], and * an [IllegalArgumentException] is thrown if such a combination is used. * For creating a conflated channel that instead keeps the existing element in the channel and throws out * the new one, use `Channel(1, onBufferOverflow = BufferOverflow.DROP_LATEST)`. */ public const val CONFLATED: Int = -1 /** * A channel capacity marker that is substituted by the default buffer capacity. * * When passed as a parameter to the `Channel(...)` factory function, the default buffer capacity is used. * For [BufferOverflow.SUSPEND] (the default buffer overflow strategy), the default capacity is 64, * but on the JVM it can be overridden by setting the [DEFAULT_BUFFER_PROPERTY_NAME] system property. * The overridden value is used for all channels created with a default buffer capacity, * including those created in third-party libraries. * * ``` * val channel = Channel(Channel.BUFFERED) * repeat(100) { * channel.trySend(it) * } * channel.close() * // The check can fail if the default buffer capacity is changed * check(channel.toList() == (0..<64).toList()) * ``` * * If a different [BufferOverflow] is specified, `Channel(BUFFERED)` creates a channel with a buffer of size 1: * * ``` * val channel = Channel(Channel.BUFFERED, onBufferOverflow = BufferOverflow.DROP_OLDEST) * channel.send(1) * channel.send(2) * channel.send(3) * channel.close() * check(channel.toList() == listOf(3)) * ``` */ public const val BUFFERED: Int = -2 // only for internal use, cannot be used with Channel(...) internal const val OPTIONAL_CHANNEL = -3 /** * Name of the JVM system property for the default channel capacity (64 by default). * * See [BUFFERED] for details on how this property is used. * * Setting this property affects the default channel capacity for channel constructors, * channel-backed coroutines and flow operators that imply channel usage, * including ones defined in 3rd-party libraries. * * Usage of this property is highly discouraged and is intended to be used as a last-ditch effort * as an immediate measure for hot fixes and duct-taping. */ @DelicateCoroutinesApi public const val DEFAULT_BUFFER_PROPERTY_NAME: String = "kotlinx.coroutines.channels.defaultBuffer" internal val CHANNEL_DEFAULT_CAPACITY = systemProp(DEFAULT_BUFFER_PROPERTY_NAME, 64, 1, UNLIMITED - 1 ) } } /** * Creates a channel. See the [Channel] interface documentation for details. * * This function is the most flexible way to create a channel. * It allows specifying the channel's capacity, buffer overflow strategy, and an optional function to call * to handle undelivered elements. * * ``` * val allocatedResources = HashSet() * // An autocloseable resource that must be closed when it is no longer needed * class Resource(val id: Int): AutoCloseable { * init { * allocatedResources.add(id) * } * override fun close() { * allocatedResources.remove(id) * } * } * // A channel with a 15-element buffer that drops the oldest element on buffer overflow * // and closes the elements that were not delivered to the consumer * val channel = Channel( * capacity = 15, * onBufferOverflow = BufferOverflow.DROP_OLDEST, * onUndeliveredElement = { element -> element.close() } * ) * // A sender's view of the channel * val sendChannel: SendChannel = channel * repeat(100) { * sendChannel.send(Resource(it)) * } * sendChannel.close() * // A receiver's view of the channel * val receiveChannel: ReceiveChannel = channel * val receivedResources = receiveChannel.toList() * // Check that the last 15 sent resources were received * check(receivedResources.map { it.id } == (85 until 100).toList()) * // Close the resources that were successfully received * receivedResources.forEach { it.close() } * // The dropped resources were closed by the channel itself * check(allocatedResources.isEmpty()) * ``` * * For a full explanation of every parameter and their interaction, see the [Channel] interface documentation. * * @param capacity either a positive channel capacity or one of the constants defined in [Channel.Factory]. * See the "Channel capacity" section in the [Channel] documentation. * @param onBufferOverflow configures an action on buffer overflow. * See the "Buffer overflow" section in the [Channel] documentation. * @param onUndeliveredElement a function that is called when element was sent but was not delivered to the consumer. * See the "Undelivered elements" section in the [Channel] documentation. * @throws IllegalArgumentException when [capacity] < -2 */ public fun Channel( capacity: Int = RENDEZVOUS, onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND, onUndeliveredElement: ((E) -> Unit)? = null ): Channel = when (capacity) { RENDEZVOUS -> { if (onBufferOverflow == BufferOverflow.SUSPEND) BufferedChannel(RENDEZVOUS, onUndeliveredElement) // an efficient implementation of rendezvous channel else ConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement) // support buffer overflow with buffered channel } CONFLATED -> { require(onBufferOverflow == BufferOverflow.SUSPEND) { "CONFLATED capacity cannot be used with non-default onBufferOverflow" } ConflatedBufferedChannel(1, BufferOverflow.DROP_OLDEST, onUndeliveredElement) } UNLIMITED -> BufferedChannel(UNLIMITED, onUndeliveredElement) // ignores onBufferOverflow: it has buffer, but it never overflows BUFFERED -> { // uses default capacity with SUSPEND if (onBufferOverflow == BufferOverflow.SUSPEND) BufferedChannel(CHANNEL_DEFAULT_CAPACITY, onUndeliveredElement) else ConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement) } else -> { if (onBufferOverflow === BufferOverflow.SUSPEND) BufferedChannel(capacity, onUndeliveredElement) else ConflatedBufferedChannel(capacity, onBufferOverflow, onUndeliveredElement) } } @Deprecated(level = DeprecationLevel.HIDDEN, message = "Since 1.4.0, binary compatibility with earlier versions") public fun Channel(capacity: Int = RENDEZVOUS): Channel = Channel(capacity) /** * Indicates an attempt to [send][SendChannel.send] to a [closed-for-sending][SendChannel.isClosedForSend] channel * that was [closed][SendChannel.close] without a cause. * * If a cause was provided, that cause is thrown from [send][SendChannel.send] instead of this exception. * In particular, if the channel was closed because it was [cancelled][ReceiveChannel.cancel], * this exception will never be thrown: either the `cause` of the cancellation is thrown, * or a new [CancellationException] gets constructed to be thrown from [SendChannel.send]. * * This exception is a subclass of [IllegalStateException], because the sender should not attempt to send to a closed * channel after it itself has [closed][SendChannel.close] it, and indicates an error on the part of the programmer. * Usually, this exception can be avoided altogether by restructuring the code. */ public class ClosedSendChannelException(message: String?) : IllegalStateException(message) /** * Indicates an attempt to [receive][ReceiveChannel.receive] from a * [closed-for-receiving][ReceiveChannel.isClosedForReceive] channel * that was [closed][SendChannel.close] without a cause. * * If a clause was provided, that clause is thrown from [receive][ReceiveChannel.receive] instead of this exception. * In particular, if the channel was closed because it was [cancelled][ReceiveChannel.cancel], * this exception will never be thrown: either the `cause` of the cancellation is thrown, * or a new [CancellationException] gets constructed to be thrown from [ReceiveChannel.receive]. * * This exception is a subclass of [NoSuchElementException] to be consistent with plain collections. */ public class ClosedReceiveChannelException(message: String?) : NoSuchElementException(message)