de.esoco.coroutine

Class Channel<T>

java.lang.Object

de.esoco.coroutine.Channel<T>

All Implemented Interfaces:

java.lang.AutoCloseable

public class Channel<T>
extends java.lang.Object
implements java.lang.AutoCloseable

A channel that allows communication between Coroutines. A channel has a fixed capacity and suspends any further sending of data after the capacity has been reached until capacity becomes available again when data is requested by receivers. Receiving will be suspended if no more data is available in a channel.

Channels can be queried with the method getChannel(ChannelId) which is available in CoroutineScope and CoroutineContext. If no channel exists at first access a new channel with a capacity of 1 will be created. A channel with a different capacity can be created with createChannel(ChannelId, int), but only if it doesn't exist already. Channels can be removed with removeChannel(ChannelId).

If a channel is needed for the communication between coroutines in different scopes it needs to be created in a common context of the scopes. If it is only needed in a particular scope it should be created there.

Channels can be closed by invoking close(). A closed channel rejects any further send or receive calls by throwing a ChannelClosedException. Upon a close all pending suspensions will also be failed with that exception.

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	Channel(ChannelId<T> rId, int nCapacity) Creates a new instance.

Method Summary

Modifier and Type	Method and Description
void	checkClosed() Throws a ChannelClosedException if this channel is already closed.
void	close() Closes this channel.
ChannelId<T>	getId() Returns the channel identifier.
boolean	isClosed() Returns the closed.
T	receiveBlocking() Receives a value from this channel, blocking if no data is available.
void	receiveSuspending(Suspension<T> rSuspension) Tries to receive a value from this channel and resumes the execution of a Coroutine at the given suspension as soon as a value becomes available.
int	remainingCapacity() Returns the number of values that can still be send to this channel.
void	sendBlocking(T rValue) Sends a value into this channel, blocking if no capacity is available.
void	sendSuspending(Suspension<T> rSuspension) Tries to send a value into this channel and resumes the execution of a Coroutine at the given step as soon as channel capacity becomes available.
int	size() Returns the current number of values in this channel.
java.lang.String	toString()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

Channel

protected Channel(ChannelId<T> rId,
                  int nCapacity)

Creates a new instance.

Parameters:

rId - The channel identifier

nCapacity - The maximum number of values the channel can hold before blocking

Method Detail

checkClosed

public final void checkClosed()

Throws a ChannelClosedException if this channel is already closed.

close

public void close()

Closes this channel. All send and receive operations on a closed channel will throw a ChannelClosedException. If there are remaining suspensions in this channel they will also be failed with such an exception.

Specified by:

close in interface java.lang.AutoCloseable

getId

public ChannelId<T> getId()

Returns the channel identifier.

Returns:

The channel ID

isClosed

public final boolean isClosed()

Returns the closed.

Returns:

The closed

receiveBlocking

public T receiveBlocking()

Receives a value from this channel, blocking if no data is available.

Returns:

The next value from this channel or NULL if the waiting for a value has been interrupted

receiveSuspending

public void receiveSuspending(Suspension<T> rSuspension)

Tries to receive a value from this channel and resumes the execution of a Coroutine at the given suspension as soon as a value becomes available. This can be immediately or, if the channel is empty, only after some other code sends a values into this channel. Suspended senders will be served with a first-suspended-first-served policy.

Parameters:

rSuspension - The coroutine suspension to resume after data has been receive

remainingCapacity

public int remainingCapacity()

Returns the number of values that can still be send to this channel. Due to the concurrent nature of channels this can only be a momentary value which needs to be interpreted with caution and necessary synchronization should be performed if applicable. Concurrently running coroutines could affect this value at any time.

Returns:

The remaining channel capacity

sendBlocking

public void sendBlocking(T rValue)

Sends a value into this channel, blocking if no capacity is available.

Parameters:

rValue - The value to send

sendSuspending

public void sendSuspending(Suspension<T> rSuspension)

Tries to send a value into this channel and resumes the execution of a Coroutine at the given step as soon as channel capacity becomes available. This can be immediately or, if the channel is full, only after some other code receives a values from this channel. Suspended senders will be served with a first-suspended-first-served policy.

Parameters:

rSuspension - rValue The value to send

size

public int size()

Returns the current number of values in this channel. Due to the concurrent nature of channels this can only be a momentary value which needs to be interpreted with caution and necessary synchronization should be performed if applicable. Concurrently running coroutines could affect this value at any time.

Returns:

The current number of channel entries

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object