Permalink
Fetching contributors…
Cannot retrieve contributors at this time
180 lines (166 sloc) 7.86 KB
/*
* Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.experimental.channels
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.Channel.Factory.UNLIMITED
import kotlin.coroutines.experimental.*
/**
* Scope for [produce][CoroutineScope.produce] coroutine builder.
*
* **Note: This is an experimental api.** Behaviour of producers that work as children in a parent scope with respect
* to cancellation and error handling may change in the future.
*/
@ExperimentalCoroutinesApi
public interface ProducerScope<in E> : CoroutineScope, SendChannel<E> {
/**
* A reference to the channel that this coroutine [sends][send] elements to.
* It is provided for convenience, so that the code in the coroutine can refer
* to the channel as `channel` as apposed to `this`.
* All the [SendChannel] functions on this interface delegate to
* the channel instance returned by this function.
*/
val channel: SendChannel<E>
}
/**
* @suppress **Deprecated**: Use `ReceiveChannel`.
*/
@Deprecated(message = "Use `ReceiveChannel`", replaceWith = ReplaceWith("ReceiveChannel"))
@Suppress("MULTIPLE_DEFAULTS_INHERITED_FROM_SUPERTYPES_WHEN_NO_EXPLICIT_OVERRIDE")
interface ProducerJob<out E> : ReceiveChannel<E>, Job {
@Deprecated(message = "Use ReceiveChannel itself")
val channel: ReceiveChannel<E>
}
/**
* Launches new coroutine to produce a stream of values by sending them to a channel
* and returns a reference to the coroutine as a [ReceiveChannel]. This resulting
* object can be used to [receive][ReceiveChannel.receive] elements produced by this coroutine.
*
* The scope of the coroutine contains [ProducerScope] interface, which implements
* both [CoroutineScope] and [SendChannel], so that coroutine can invoke
* [send][SendChannel.send] directly. The channel is [closed][SendChannel.close]
* when the coroutine completes.
* The running coroutine is cancelled when its receive channel is [cancelled][ReceiveChannel.cancel].
*
* Coroutine context is inherited from a [CoroutineScope], additional context elements can be specified with [context] argument.
* If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
* The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
* with corresponding [coroutineContext] element.
*
* Uncaught exceptions in this coroutine close the channel with this exception as a cause and
* the resulting channel becomes _failed_, so that any attempt to receive from such a channel throws exception.
*
* The kind of the resulting channel depends on the specified [capacity] parameter.
* See [Channel] interface documentation for details.
*
* See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
*
* **Note: This is an experimental api.** Behaviour of producers that work as children in a parent scope with respect
* to cancellation and error handling may change in the future.
*
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param capacity capacity of the channel's buffer (no buffer by default).
* @param block the coroutine code.
*/
@ExperimentalCoroutinesApi
public fun <E> CoroutineScope.produce(
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = 0,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> {
val channel = Channel<E>(capacity)
val newContext = newCoroutineContext(context)
val coroutine = ProducerCoroutine(newContext, channel)
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
return coroutine
}
/**
* @suppress **This an internal API and should not be used from general code.**
* onCompletion parameter will be redesigned.
*/
@InternalCoroutinesApi
public fun <E> CoroutineScope.produce(
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = 0,
onCompletion: CompletionHandler? = null,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> {
val channel = Channel<E>(capacity)
val newContext = newCoroutineContext(context)
val coroutine = ProducerCoroutine(newContext, channel)
if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
return coroutine
}
/**
* Launches new coroutine to produce a stream of values by sending them to a channel
* and returns a reference to the coroutine as a [ReceiveChannel].
* @suppress **Deprecated** Use [CoroutineScope.produce] instead.
*/
@Deprecated(
message = "Standalone coroutine builders are deprecated, use extensions on CoroutineScope instead",
replaceWith = ReplaceWith("GlobalScope.produce(context, capacity, onCompletion, block)",
imports = ["kotlinx.coroutines.experimental.GlobalScope", "kotlinx.coroutines.experimental.channels.produce"])
)
public fun <E> produce(
context: CoroutineContext = Dispatchers.Default,
capacity: Int = 0,
onCompletion: CompletionHandler? = null,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> =
GlobalScope.produce(context, capacity, onCompletion, block)
/**
* Launches new coroutine to produce a stream of values by sending them to a channel
* and returns a reference to the coroutine as a [ReceiveChannel].
* @suppress **Deprecated** Use [CoroutineScope.produce] instead.
*/
@Deprecated(
message = "Standalone coroutine builders are deprecated, use extensions on CoroutineScope instead",
replaceWith = ReplaceWith("GlobalScope.produce(context + parent, capacity, onCompletion, block)",
imports = ["kotlinx.coroutines.experimental.GlobalScope", "kotlinx.coroutines.experimental.channels.produce"])
)
public fun <E> produce(
context: CoroutineContext = Dispatchers.Default,
capacity: Int = 0,
parent: Job? = null,
onCompletion: CompletionHandler? = null,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> =
GlobalScope.produce(context + (parent ?: EmptyCoroutineContext), capacity, onCompletion, block)
/** @suppress **Deprecated**: Binary compatibility */
@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
public fun <E> produce(
context: CoroutineContext = Dispatchers.Default,
capacity: Int = 0,
parent: Job? = null,
block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E> = GlobalScope.produce(context + (parent ?: EmptyCoroutineContext), capacity, block = block)
/** @suppress **Deprecated**: Binary compatibility */
@Deprecated(message = "Binary compatibility", level = DeprecationLevel.HIDDEN)
public fun <E> produce(
context: CoroutineContext = Dispatchers.Default,
capacity: Int = 0,
block: suspend ProducerScope<E>.() -> Unit
): ProducerJob<E> =
GlobalScope.produce(context, capacity, block = block) as ProducerJob<E>
/**
* @suppress **Deprecated**: Renamed to `produce`.
*/
@Deprecated(message = "Renamed to `produce`", replaceWith = ReplaceWith("produce(context, capacity, block)"))
public fun <E> buildChannel(
context: CoroutineContext,
capacity: Int = 0,
block: suspend ProducerScope<E>.() -> Unit
): ProducerJob<E> =
GlobalScope.produce(context, capacity, block = block) as ProducerJob<E>
private class ProducerCoroutine<E>(
parentContext: CoroutineContext, channel: Channel<E>
) : ChannelCoroutine<E>(parentContext, channel, active = true), ProducerScope<E>, ProducerJob<E> {
override val isActive: Boolean
get() = super<ChannelCoroutine>.isActive
override fun onCompletionInternal(state: Any?, mode: Int, suppressed: Boolean) {
val cause = (state as? CompletedExceptionally)?.cause
val processed = _channel.close(cause)
if (cause != null && !processed && suppressed) handleExceptionViaHandler(context, cause)
}
}