Class

OrderedChannel

A topic where every subscriber, in every worker, receives the messages in one and the same order, and each publisher's own messages keep theirs.

Introduction#

Plain Swerve::publish() from one publisher is already ordered for its subscribers, which is enough for a stream between two coroutines or workers. This is for many publishers that need one common order.

It is a broadcast, not a queue: nothing is written while nobody subscribes anywhere, and there are no writers to end it or close it. Retention is bounded (about 30 s): a subscriber lagging beyond it gets a SubscriberLagException. Each message costs one extra file append; a worker is woken by a datagram, best effort, and looks at the log at least every second. It is separate from the plain topic of the same name. See docs/publish-subscribe.md.

Class synopsis#

final class OrderedChannel implements \IteratorAggregate, \Traversable {
    /* Methods */
    public OrderedChannel::__construct(string $name)
    public OrderedChannel::write(mixed $message): void
    public OrderedChannel::subscribe(
        float $maxLag = 30.0,
        ?float $heartbeat = null
    ): Subscription
    public OrderedChannel::getIterator(): Subscription
}

Properties#

Name	Type	Description
`$name`	`readonly string`

Methods#

Method	Description
`__construct`	A channel by `$name`, separate from the plain topic of the same name.
`write`	Append `$message` to the channel, where every subscriber receives it in the channel's one order.
`subscribe`	Receive what is written from now on, as `Swerve::subscribe` does.
`getIterator`	Subscribes with the defaults, so `foreach ($channel as $message)` subscribes at the foreach.

Examples#

Example 1One order for every subscriber

use Swerve\OrderedChannel;

phasync::run(function () {
    $ledger = new OrderedChannel('ledger');

    // Two subscribers, which in a server would be in different workers
    foreach (['audit', 'report'] as $name) {
        $subscription = $ledger->subscribe();
        phasync::go(function () use ($subscription, $name) {
            $entries = [];
            foreach ($subscription as $entry) {
                $entries[] = implode(' ', $entry);
                if (3 === count($entries)) {
                    break;
                }
            }
            echo "$name: ", implode(', ', $entries), "\n";   // the same order for both
        });
    }

    // Several writers; every subscriber sees one common order
    phasync::go(fn () => $ledger->write(['debit', 12]));
    phasync::go(fn () => $ledger->write(['credit', 5]));
    phasync::go(fn () => $ledger->write(['debit', 3]));
});

Output

audit: debit 12, credit 5, debit 3
report: debit 12, credit 5, debit 3

Example 2

$ledger = new OrderedChannel('ledger');
$ledger->write(['debit', 12]);
foreach ($ledger as $entry) { ... }