Channel Reference

Thread-safe queue for sending values from producers to consumers

What it is #

Routines #

Signatures are reproduced from the documentation, including their declared return types. A routine listed here is part of the type's published interface; whether Raku++ implements it is a separate question, answered by the examples below.

method send #

method send(Channel:D: \item)

Enqueues an item into the Channel. Throws an exception of type X::Channel::SendOnClosed if the Channel has been closed already. This call will not block waiting for a consumer to take the object. There is no set limit on the number of items that may be queued, so care should be taken to prevent runaway queueing.

method receive #

method receive(Channel:D:)

Receives and removes an item from the Channel. It blocks if no item is present, waiting for a send from another thread. Throws an exception of type X::Channel::ReceiveOnClosed if the Channel has been closed, and the last item has been removed already, or if close is called while receive is waiting for an item to arrive. If the Channel has been marked as erratic with method fail, and the last

method poll #

method poll(Channel:D:)

Receives and removes an item from the Channel. If no item is present, returns Nil instead of waiting. See method receive for a blocking version that properly responds to Channel closing and failure.

method close #

method close(Channel:D:)

Close the Channel, normally. This makes subsequent send calls die with X::Channel::SendOnClosed. Subsequent calls of .receive may still drain any remaining items that were previously sent, but if the queue is empty, will throw an X::Channel::ReceiveOnClosed exception. Since you can produce a Seq from a Channel by contextualizing to array with @()

method list #

method list(Channel:D:)

Returns a list based on the Seq which will iterate items in the queue and remove each item from it as it iterates. This can only terminate once the close method has been called.

method closed #

method closed(Channel:D: --> Promise:D)

Returns Promise:D.

Returns a promise that will be kept once the Channel is closed by a call to method close.

method fail #

method fail(Channel:D: $error)

Closes the Channel (that is, makes subsequent send calls die), and enqueues the error to be thrown as the final element in the Channel. Method receive will throw that error as an exception. Does nothing if the Channel has already been closed or .fail has already been called on it.

method Capture #

method Capture(Channel:D: --> Capture:D)

Returns Capture:D.

Equivalent to calling .List.Capture on the invocant.

method Supply #

method Supply(Channel:D:)

This returns an on-demand Supply that emits a value for every value received on the Channel. done will be called on the Supply when the Channel is closed. Multiple calls to this method produce multiple instances of Supply, which compete over the values from the Channel.

sub await #

multi await(Channel:D)
multi await(*@)

Waits until all of one or more Channels has a value available, and returns those values (it calls .receive on the Channel). Also works with Promises. Since 6.d, it no longer blocks a thread while waiting.

Examples, run three ways #

Every example below comes from the official documentation, together with the output that documentation asserts. Each was then executed by Rakudo and by Raku++ when this page was built. Where the three agree, one result is shown; where they do not, all three are — because which of them is wrong is exactly the information worth having.

6 no-output · 5 ok

class Channel {}

my $c = Channel.new;
await (^10).map: {
    start {
        my $r = rand;
        sleep $r;
        $c.send($r);
    }
}
$c.close;
say $c.list;

my $c = Channel.new;
$c.send(1);
$c.send([2, 3, 4, 5]);
$c.close;
say $c.list; # OUTPUT: «(1 [2 3 4 5])␤»

(1 [2 3 4 5])

my $c = Channel.new;
$c.send(1);
say $c.receive; # OUTPUT: «1␤»

1

my $c = Channel.new;
Promise.in(2).then: { $c.close; }
^10 .map({ $c.send($_); });
loop {
    if $c.poll -> $item { $item.say };
    if $c.closed  { last };
    sleep 0.1;
}

my $c = Channel.new;
$c.close;
$c.send(1);
CATCH { default { put .^name, ': ', .Str } };
# OUTPUT: «X::Channel::SendOnClosed: Cannot send a message on a closed channel␤»

X::Channel::SendOnClosed: Cannot send a message on a closed channel

my $c = Channel.new; $c.send(1); $c.send(2);
$c.close;
say $c.list; # OUTPUT: «(1 2)␤»

(1 2)

my $c = Channel.new;
$c.closed.then({ say "It's closed!" });
$c.close;
sleep 1;

my $c = Channel.new;
$c.fail("Bad error happens!");
$c.receive;
CATCH { default { put .^name, ': ', .Str } };
# OUTPUT: «X::AdHoc: Bad error happens!␤»

X::AdHoc: Bad error happens!

my $c = Channel.new;
my Supply $s1 = $c.Supply;
my Supply $s2 = $c.Supply;
$s1.tap(-> $v { say "First $v" });
$s2.tap(-> $v { say "Second $v" });
^10 .map({ $c.send($_) });
sleep 1;

my $c = Channel.new;
Promise.in(1).then({$c.send(1)});
say await $c;