Class

Subscription

The messages published to a topic, from Swerve::subscribe: iterate over it to receive them.

Introduction#

It receives what is published from the moment it was created, not from the first iteration. It ends when its last reference goes: a break out of the loop, the variable going out of scope, or the request's coroutine ending. There is no unsubscribe() to forget. When the worker stops (shutdown, reload, the end of a recycle), its iteration throws a WorkerStoppingException at its wait, as every wait does then.

A subscriber that falls behind gets a SubscriberLagException from the iteration: when the message it is about to receive arrived in this process more than $maxLag seconds ago. Messages wait for it meanwhile, shared with the topic's other subscribers.

Class synopsis#

final class Subscription implements \IteratorAggregate, \Traversable {
    /* Methods */
    public Subscription::__construct(
        string $topic,
        float $maxLag = 30.0,
        ?float $heartbeat = null
    )
    public Subscription::getIterator(): Generator
    public Subscription::__destruct()
}

Properties#

Name	Type	Description
`$topic`	`readonly string`
`$maxLag`	`readonly float`
`$heartbeat`	`readonly ?float`

Methods#

Method	Description
`__construct`	Subscribe to `$topic`: `Swerve::subscribe` is how an application gets one.
`getIterator`	The messages, as they come; null after each `$heartbeat` seconds without one.
`__destruct`	Ends the subscription: the topic stops keeping messages for it.

Examples#

Example 1A subscription ends with its last reference

use Swerve\Swerve;

phasync::run(function () {
    $subscription = Swerve::subscribe('ticks');   // receives from here on
    phasync::go(function () {
        foreach ([1, 2, 3] as $n) {
            Swerve::publish('ticks', $n);
        }
    });

    foreach ($subscription as $tick) {
        echo "tick $tick\n";
        if (2 === $tick) {
            break;
        }
    }
    unset($subscription);                         // its last reference: the subscription ends

    $later = Swerve::subscribe('ticks');
    Swerve::publish('ticks', 4);
    foreach ($later as $tick) {
        echo "a new subscription gets only tick $tick\n";
        break;
    }
});

Output

tick 1
tick 2
a new subscription gets only tick 4

Example 2

$subscription = Swerve::subscribe('chat', heartbeat: 15);   // subscribed from here
foreach ($subscription as $message) {
    if (null === $message) {
        continue;                      // 15 s without a message
    }
    $ws->send(json_encode($message));
}