Channel Laravel Package

Getting Started

Channels are a low-level concurrency primitive, not a Laravel feature. Before installing, confirm you have somewhere to run blocking code: a fiber context (Psl\Async\run()), or a worker under RoadRunner/Swoole/Octane. If your code path is a plain PHP-FPM controller, stop here — a blocking receive() will hang the request.

Install

composer require php-standard-library/channel

This pulls in php-standard-library/async as a dependency. The PSL ships as a lockstep monorepo, so keep every php-standard-library/* package on the same minor version (composer show 'php-standard-library/*' after install). Mixing a 5.x split package with 6.x async internals is the most common first-day breakage.

Where to look first

The split repo is nearly empty — README points at php-standard-library.dev, and the release notes are stubs that defer to the monorepo tag. The authoritative reference is the source tree in the main repo under src/Psl/Channel/:

• ChannelInterface — send(), receive(), close(), isClosed()
• BufferedChannel — constructed with a capacity
• UnbufferedChannel — strict handoff, capacity 0
• Pair — factory returning a [SenderInterface, ReceiverInterface] tuple, which is how you should hand channels to producers/consumers so each side only sees half the API

Read ChannelInterface and BufferedChannel for ten minutes before writing anything. The surface is small; the semantics (what blocks, what throws after close()) are the whole story.

First use case

Fan-out N independent I/O calls inside an Artisan command and collect the results, instead of foreach-ing them serially:

use Psl\Async;
use Psl\Channel;

$rows = Async\run(static function (iterable $ids): array {
    [$tx, $rx] = Channel\Pair::buffered(32);

    Async\run(static function () use ($tx, $ids): void {
        try {
            foreach ($ids as $id) {
                $tx->send(fetchRow($id));   // suspends once the buffer is full
            }
        } finally {
            $tx->close();                   // consumers stop here
        }
    });

    $out = [];
    foreach ($rx as $row) {                 // iterates until close()
        $out[] = $row;
    }

    return $out;
}, $ids);

That's the whole mental model: one side sends, one side receives, close() is the terminator.

Implementation Patterns

Worker pool with a fixed concurrency ceiling. Spawn K consumer fibers that all loop on the same receiver, and feed the channel from one producer. Buffer size ≈ K × 2 is a reasonable starting point. Each consumer calls receive() in a while (! $rx->isClosed()) loop (or just foreach, which is cleaner).

Staged pipeline. Chain channels: stage A reads a source and sends to $a, stage B reads $a and sends to $b, stage C drains $b. Each stage is its own fiber, each with its own buffer capacity — which means you can tune backpressure at each boundary independently. This is the pattern that actually beats Laravel's chunked lazy() + queue jobs for CPU-light, I/O-heavy transforms.

Choosing buffered vs. unbuffered as a design lever. Unbuffered = rendezvous; a send() does not return until a receiver takes the value. Use it for coordination (a "done" signal, a permit semaphore, sequencing two stages). Buffered = decoupling; use it to smooth bursts. If you find yourself raising the buffer capacity to fix a hang, you have a consumer-throughput bug, not a capacity bug.

Encapsulate behind an application service. Don't hand ChannelInterface to controllers. Bind a small service in a provider and expose intent methods:

// AppServiceProvider
$this->app->scoped(InventoryFeed::class, function () {
    [$tx, $rx] = Channel\Pair::buffered(64);
    return new InventoryFeed($tx, $rx);
});

Use scoped(), not singleton(), if you run under Octane — a closed channel surviving across requests is a debugging nightmare.

Keep the concurrent section narrow. Fibers share one PDO connection per worker. Anything you send into a channel should be a scalar, an array, or a DTO — never an Eloquent model with lazy relations attached. Resolve data before the send, persist after the receive. If you must touch the DB from multiple fibers under Octane, set the connection pool size or you'll serialize on the connection anyway.

Non-blocking polling where you need it. The README advertises non-blocking send/receive alongside the blocking ones (check the concrete class for the exact method names in your version). Use those for "try to drain the queue for 50ms, then go do housekeeping" loops rather than trying to build a timeout out of a blocking call.

Testing. Channels are trivially testable because they have no I/O. In PHPUnit, wrap the exercise in Async\run() and assert on the collected output. For code that uses a channel internally, inject a narrow interface (SenderInterface/ReceiverInterface) so tests can pass a pre-filled buffered channel instead of driving real concurrency.

Gotchas and Tips

Channels are in-process only. No cross-process, no cross-container, no persistence. Two Laravel containers cannot talk over a channel; two fibers in one worker can. If you need durability, use the queue — or implement ChannelInterface over Redis and keep the same call sites. That interface is the intended extension point.

Deadlock #1: single fiber. Sending on a full buffer with no concurrent receiver blocks forever. Any test that does send() then receive() in the same fiber must use an unbuffered-free path or run the producer in Async\run().

Deadlock #2: missing close(). foreach over a receiver never terminates unless the sender closes. Always put close() in a finally — an exception in the producer otherwise looks identical to a slow producer.

Exceptions don't propagate across the channel. A producer that throws disappears; the consumer waits forever. Capture into the payload (['ok' => false, 'error' => ...]) or keep a reference to the Awaitable and rethrow after the join. Logging at the send()/receive() boundary is the single highest-value debugging investment here.

Close semantics are asymmetric. Send-after-close throws; receive-after-close drains buffered items and then ends (or throws, depending on the variant). Write your consumers to tolerate both, and check isClosed() before treating a receive() failure as an error.

Buffer sizing has a memory cost. A buffered channel holds strong references to every queued value until consumed. Capacity 10,000 × a 1MB payload is an OOM. Under Octane, monitor with memory_get_peak_usage() in a middleware; leaked channels show up as slowly climbing worker RSS.

No timeouts, no metrics, no backpressure signal. There is no built-in "how full is the buffer" or "how many senders are parked." You get to instrument it yourself — count sends/receives around the calls and log when a send takes longer than N ms.

Version lockstep. PSL split packages are released together. composer update php-standard-library/channel alone can leave you on 6.3 while async is on 6.1. Update the wildcard: composer update 'php-standard-library/*'.

Don't replace your queue with this. The package's own pitch says it best: reserve channels for internal, ephemeral coordination; keep Illuminate\Queue for anything that must survive a deploy, be retried, or be observed by a dashboard. The value here is sub-millisecond handoff between fibers in one process — nothing more, nothing less.