Menu

Reference

Reference / Async

Channel

finalclass
final class Channel implements \IteratorAggregate

A typed pipe between tasks, in the style of a Go channel.

A channel with capacity 0 (the default) is an unbuffered rendezvous: send() waits until a receiver takes the value. A buffered channel accepts up to capacity values without a waiting receiver and parks senders only when full and receivers only when empty. Values are delivered in FIFO order and waiting tasks are served in arrival order. Waiting suspends the task, never the process, and a waiting task can be cancelled.

Consume with foreach: the channel is itself iterable and the loop ends when the channel is closed and drained. next() is the explicit form for when you must tell "closed" from a legitimate null payload, and recv() is the terse form that cannot. Use select() to wait on several channels at once.

$ch = channel();
spawn(function () use ($ch) {
    $ch->send(1);
    $ch->send(2);
    $ch->close();
});
foreach ($ch as $value) {
    echo $value;
}

Example

use Async\ChannelClosedException;
use function Async\async;
use function Async\channel;
use function Async\spawn;

async(function () {
    // Capacity 0 is a rendezvous: send() waits for a receiver.
    $ch = channel();

    spawn(function () use ($ch) {
        foreach ([1, 2, 3] as $n) {
            $ch->send($n);
        }
        $ch->close();
    });

    // foreach ends once the channel is closed and drained.
    foreach ($ch as $value) {
        echo "got $value\n";
    }

    // next() is the explicit form: ok is false on a closed channel,
    // which lets null be a legitimate payload.
    $r = $ch->next();
    echo $r->ok ? "value\n" : "closed\n";

    try {
        $ch->send(4);
    } catch (ChannelClosedException $e) {
        echo "send failed: ", $e->getMessage(), "\n";
    }

    // A buffered channel takes values without a waiting receiver.
    $buf = channel(2);
    $buf->send("a");
    $buf->send("b");
    echo $buf->recv(), $buf->recv(), "\n";
});

Output

got 1
got 2
got 3
closed
send failed: send on a closed channel
ab

Example

use Async\SelectCase;
use function Async\async;
use function Async\channel;
use function Async\delay;
use function Async\select;
use function Async\selectNow;
use function Async\selectWithin;
use function Async\spawn;

async(function () {
    $fast = channel();
    $slow = channel();

    spawn(function () use ($fast) { delay(0.01); $fast->send("from fast"); });
    spawn(function () use ($slow) { delay(0.05); $slow->send("from slow"); });

    // A bare channel is shorthand for a receive case.
    $r = select([$slow, $fast]);
    echo "case {$r->index}: {$r->value}\n";

    // Nothing is ready right now.
    var_dump(selectNow([$fast]));

    // A deadline: null when it expires first.
    var_dump(selectWithin(0.01, [$fast]));

    // The other value is still on its way.
    $r = select([$slow]);
    echo $r->value, "\n";

    // A send arm completes when a receiver (or buffer space) is available.
    $out = channel(1);
    $r = select([SelectCase::send($out, 7)]);
    echo $r->isSend ? "sent " : "received ", $out->recv(), "\n";
});

Output

case 1: from fast
NULL
NULL
from slow
sent 7

See also: channel, select, Received

Methods

__construct()Create a channel; channel() is the usual spelling.
isClosed()Whether close() has been called.
getIterator()Iterate over the received values: foreach ($channel as $value).
send()Send $value, suspending until it is buffered or taken by a receiver.
next()Receive the next value as a Received object.
recv()Receive the next value, or null once the channel is closed and drained.
close()Close the channel.

__construct()#source

function __construct(int $capacity = 0)

Create a channel; channel() is the usual spelling.

Parameters

$capacity int = 0
buffer size; 0 (the default, and any negative value) means unbuffered

isClosed()#source

function isClosed(): bool

Whether close() has been called.

Values still buffered when the channel was closed can be received after this turns true.

Returns

bool — true once closed

getIterator()#source

function getIterator(): \Iterator

Iterate over the received values: foreach ($channel as $value).

Each step receives one value, suspending while the channel is empty. The loop ends once the channel is closed and drained. The keys are the 0-based count of values received by this iteration. The values are consumed, so two loops over one channel split the stream between them.

Returns

\Iterator — an iterator over the channel's values

send()#source

function send(mixed $value): void

Send $value, suspending until it is buffered or taken by a receiver.

Returns at once when a receiver is already waiting or the buffer has room. Otherwise the calling task parks (on an unbuffered channel, until a receiver takes this exact value). A task cancelled while parked withdraws its value, so it is never delivered.

Parameters

$value mixed
the value to deliver; may be null

Throws

ChannelClosedException
when the channel is closed, or closed while this call waits
CancelledException
when the task is cancelled while waiting

Example

use Async\ChannelClosedException;
use function Async\async;
use function Async\channel;
use function Async\spawn;

async(function () {
    // Capacity 0 is a rendezvous: send() waits for a receiver.
    $ch = channel();

    spawn(function () use ($ch) {
        foreach ([1, 2, 3] as $n) {
            $ch->send($n);
        }
        $ch->close();
    });

    // foreach ends once the channel is closed and drained.
    foreach ($ch as $value) {
        echo "got $value\n";
    }

    // next() is the explicit form: ok is false on a closed channel,
    // which lets null be a legitimate payload.
    $r = $ch->next();
    echo $r->ok ? "value\n" : "closed\n";

    try {
        $ch->send(4);
    } catch (ChannelClosedException $e) {
        echo "send failed: ", $e->getMessage(), "\n";
    }

    // A buffered channel takes values without a waiting receiver.
    $buf = channel(2);
    $buf->send("a");
    $buf->send("b");
    echo $buf->recv(), $buf->recv(), "\n";
});

Output

got 1
got 2
got 3
closed
send failed: send on a closed channel
ab

See also: next

next()#source

function next(): Received

Receive the next value as a Received object.

Buffered values come first, then values from parked senders; if the channel is empty the calling task parks until a sender arrives or the channel is closed. ok is false once the channel is closed and drained, the only unambiguous way to report that when null is a legal payload.

Returns

Received — value and ok; ok is false when the channel is closed and drained

Throws

CancelledException
when the task is cancelled while waiting

Example

use Async\ChannelClosedException;
use function Async\async;
use function Async\channel;
use function Async\spawn;

async(function () {
    // Capacity 0 is a rendezvous: send() waits for a receiver.
    $ch = channel();

    spawn(function () use ($ch) {
        foreach ([1, 2, 3] as $n) {
            $ch->send($n);
        }
        $ch->close();
    });

    // foreach ends once the channel is closed and drained.
    foreach ($ch as $value) {
        echo "got $value\n";
    }

    // next() is the explicit form: ok is false on a closed channel,
    // which lets null be a legitimate payload.
    $r = $ch->next();
    echo $r->ok ? "value\n" : "closed\n";

    try {
        $ch->send(4);
    } catch (ChannelClosedException $e) {
        echo "send failed: ", $e->getMessage(), "\n";
    }

    // A buffered channel takes values without a waiting receiver.
    $buf = channel(2);
    $buf->send("a");
    $buf->send("b");
    echo $buf->recv(), $buf->recv(), "\n";
});

Output

got 1
got 2
got 3
closed
send failed: send on a closed channel
ab

See also: recv, getIterator

recv()#source

function recv(): mixed

Receive the next value, or null once the channel is closed and drained.

Convenient, but it cannot tell a null payload from a closed channel: use next() or foreach when null is a real value.

Returns

mixed — the received value, or null when the channel is closed and drained

Throws

CancelledException
when the task is cancelled while waiting

Example

use Async\ChannelClosedException;
use function Async\async;
use function Async\channel;
use function Async\spawn;

async(function () {
    // Capacity 0 is a rendezvous: send() waits for a receiver.
    $ch = channel();

    spawn(function () use ($ch) {
        foreach ([1, 2, 3] as $n) {
            $ch->send($n);
        }
        $ch->close();
    });

    // foreach ends once the channel is closed and drained.
    foreach ($ch as $value) {
        echo "got $value\n";
    }

    // next() is the explicit form: ok is false on a closed channel,
    // which lets null be a legitimate payload.
    $r = $ch->next();
    echo $r->ok ? "value\n" : "closed\n";

    try {
        $ch->send(4);
    } catch (ChannelClosedException $e) {
        echo "send failed: ", $e->getMessage(), "\n";
    }

    // A buffered channel takes values without a waiting receiver.
    $buf = channel(2);
    $buf->send("a");
    $buf->send("b");
    echo $buf->recv(), $buf->recv(), "\n";
});

Output

got 1
got 2
got 3
closed
send failed: send on a closed channel
ab

See also: next

close()#source

function close(): void

Close the channel. Calling it again does nothing.

Values already buffered can still be received. Tasks parked in a receive wake with ok false (or null from recv()); tasks parked in a send, and every later send(), throw ChannelClosedException. The usual pattern is for the producer to close the channel, so that consumers' foreach loops end.

Example

use Async\ChannelClosedException;
use function Async\async;
use function Async\channel;
use function Async\spawn;

async(function () {
    // Capacity 0 is a rendezvous: send() waits for a receiver.
    $ch = channel();

    spawn(function () use ($ch) {
        foreach ([1, 2, 3] as $n) {
            $ch->send($n);
        }
        $ch->close();
    });

    // foreach ends once the channel is closed and drained.
    foreach ($ch as $value) {
        echo "got $value\n";
    }

    // next() is the explicit form: ok is false on a closed channel,
    // which lets null be a legitimate payload.
    $r = $ch->next();
    echo $r->ok ? "value\n" : "closed\n";

    try {
        $ch->send(4);
    } catch (ChannelClosedException $e) {
        echo "send failed: ", $e->getMessage(), "\n";
    }

    // A buffered channel takes values without a waiting receiver.
    $buf = channel(2);
    $buf->send("a");
    $buf->send("b");
    echo $buf->recv(), $buf->recv(), "\n";
});

Output

got 1
got 2
got 3
closed
send failed: send on a closed channel
ab

See also: ChannelClosedException

View source