Examples

use Swerve\Swerve;

phasync::run(function () {
    $mine  = Swerve::claim('import');
    $yours = Swerve::claim('import');        // two handles on one name conflict, also in one process

    var_dump($mine->available());            // nobody holds it; nothing is claimed
    var_dump($mine->acquire() === $mine);    // the handle, now holding it
    var_dump($yours->acquire());             // null: held by another
    var_dump($mine->held(), $yours->held());

    $mine->release();
    var_dump($yours->acquire() === $yours);
});

if ($claim = Swerve::claim('nightly-report')->acquire()) {
    ...
}                                     // released when $claim goes out of scope

$claim = Swerve::claim('import');
if ($claim->acquire(timeout: 5.0)) {      // tries every 20 ms for up to 5 s
    import_feed();
    $claim->release();
}

use Swerve\Swerve;

phasync::run(function () {
    phasync::go(function () {
        $claim = Swerve::claim('import')->acquire();
        echo "first: importing\n";
        phasync::sleep(0.2);
        echo "first: done\n";
    });                                      // released as $claim goes

    phasync::go(function () {
        echo "second: waiting\n";
        if ($claim = Swerve::claim('import')->acquire(timeout: 5.0)) {   // tries every 20 ms
            echo "second: importing\n";
        }
    });
});

if ($leader = Swerve::claim('leader')->acquire()) {
    // this worker holds the name until $leader is released or destroyed
}

use Swerve\Swerve;

phasync::run(function () {
    $claim = Swerve::claim('leader');
    $claim->acquire();
    echo 'held: ', var_export($claim->held(), true), "\n";

    $claim->release();                       // give it up while the handle lives on
    echo 'held: ', var_export($claim->held(), true), "\n";
    echo 'available: ', var_export($claim->available(), true), "\n";

    $claim->release();                       // releasing what is not held does nothing
});

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]));
});

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

use Swerve\OrderedChannel;

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

    $events->write('lost');                  // a broadcast: nobody subscribes, nothing is kept

    $subscription = $events->subscribe();    // receives what is written from here on
    $events->write('seen');

    foreach ($subscription as $event) {
        echo "$event\n";
        break;
    }
});

$ledger = new OrderedChannel('ledger');
$ledger->write(['debit', 12]);

$subscription = (new OrderedChannel('ledger'))->subscribe(maxLag: 10.0);
foreach ($subscription as $entry) { ... }

use Swerve\OrderedChannel;

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

    phasync::go(function () use ($alerts) {
        phasync::sleep(0.1);                 // after the foreach below has subscribed
        $alerts->write('disk 91% full');
        $alerts->write('disk 95% full');
    });

    // foreach subscribes with the defaults, at the foreach: what was written before is not seen
    foreach ($alerts as $n => $alert) {
        echo "$alert\n";
        if (1 === $n) {
            break;
        }
    }
});

use Swerve\SubscriberLagException;
use Swerve\Swerve;

phasync::run(function () {
    $subscription = Swerve::subscribe('prices', maxLag: 0.1);
    Swerve::publish('prices', 101.5);

    phasync::sleep(0.3);                      // busy elsewhere for longer than maxLag

    try {
        foreach ($subscription as $price) {
            echo "price $price\n";
        }
    } catch (SubscriberLagException) {
        echo "fell behind: disconnect, and let the client catch up from storage\n";
    }
});

try {
    foreach (Swerve::subscribe('chat', maxLag: 5.0) as $message) {
        $ws->send(json_encode($message));
    }
} catch (SubscriberLagException) {
    $ws->end(1013, 'too slow');
}

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;
    }
});

$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));
}

foreach (Swerve::subscribe('chat') as $message) {
    handle($message);
}

use Swerve\Swerve;

phasync::run(function () {
    // Two subscribers, as two requests in a worker would have
    foreach (['ann', 'bob'] as $name) {
        $subscription = Swerve::subscribe('game');     // subscribed now, not at the foreach
        phasync::go(function () use ($subscription, $name) {
            foreach ($subscription as $message) {
                echo "$name got ", json_encode($message), "\n";
                if ('end' === $message) {
                    break;
                }
            }
        });
    }

    Swerve::publish('game', ['type' => 'score', 'player' => 7, 'points' => 10]);
    Swerve::publish('game', [7, 'kill']);
    Swerve::publish('game', 'end');
});

Swerve::publish('game', ['kill', $playerId]);
Swerve::publish('game', ['type' => 'score', 'player' => $playerId, 'points' => 10]);

use Swerve\ClientRequest;
use Swerve\RequestHandler;
use Swerve\Swerve;
use Swerve\WebSocket;

// Binary WebSocket frames, forwarded to every worker's subscribers: publish() would fail to
// encode them as JSON, since a binary frame is not valid UTF-8.
$app = new RequestHandler(function (ClientRequest $request) {
    WebSocket::from($request, function (WebSocket $ws) {
        $ws->onMessage->listen(function (string $data, bool $binary) {
            if ($binary) {
                Swerve::publishRaw('frames', $data);      // to this and every other worker
            }
        });
        foreach (Swerve::subscribe('frames') as $frame) { // what this and every other connection sent
            $ws->sendBinary($frame);
        }
    });
});

if (debug_backtrace()) {
    return $app;   // swerve loads this file: it is the application's swerve.php
}

// Run directly: serve the application, send one binary frame, and print what comes back
require __DIR__ . '/../serve.php';
[$swerve, $addr] = swerve_start(__FILE__);
try {
    $connection = stream_socket_client("tcp://$addr", $errno, $error, 5);
    fwrite($connection, "GET /frames HTTP/1.1\r\nHost: example\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
        . "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n");
    $head = '';
    while (!str_contains($head, "\r\n\r\n")) {
        $head .= fread($connection, 1);
    }
    $mask    = "\1\2\3\4";
    $payload = pack('nN', 1, 42);                         // any bytes, not valid UTF-8
    fwrite($connection, "\x82" . chr(0x80 | strlen($payload)) . $mask . ($payload ^ str_repeat($mask, 32)));
    $head   = fread($connection, 2);
    $length = ord($head[1]) & 0x7F;
    echo 'sent ', bin2hex($payload), ', got back ', bin2hex((string) fread($connection, $length)), "\n";
    fclose($connection);
} finally {
    swerve_stop($swerve, false);
}

Swerve::publishRaw('frames', $binaryWebSocketFrame);

use Swerve\Swerve;
use Swerve\Util\SealedObject;

phasync::run(function () {
    $subscription = Swerve::subscribe('news');

    Swerve::publish('news', 'a string stays a string');
    Swerve::publish('news', ['a', 'list']);
    Swerve::publish('news', ['title' => 'An object', 'tags' => ['x', 'y']]);

    foreach ($subscription as $message) {
        echo get_debug_type($message), ': ', json_encode($message), "\n";
        if ($message instanceof SealedObject) {
            echo "title: $message->title\n";
            break;
        }
    }
});

use Swerve\Swerve;

phasync::run(function () {
    phasync::go(function () {
        phasync::sleep(0.5);
        Swerve::publish('quiet', 'finally');
    });

    // null after each 0.2 s without a message: time for a keep-alive
    foreach (Swerve::subscribe('quiet', heartbeat: 0.2) as $message) {
        if (null === $message) {
            echo "keep-alive\n";
            continue;
        }
        echo "message: $message\n";
        break;
    }
});

foreach (Swerve::subscribe('chat') as $message) {
    echo $message->text, "\n";   // an object arrives as a read-only SealedObject
}

use Swerve\Swerve;

phasync::run(function () {
    // Three coroutines (in a server: workers) want to run the nightly report; one may
    foreach (['a', 'b', 'c'] as $who) {
        phasync::go(function () use ($who) {
            if ($claim = Swerve::claim('nightly-report')->acquire()) {
                echo "$who runs the report\n";
                phasync::sleep(0.1);
            } else {
                echo "$who: someone else is on it\n";
            }
        });                                   // $claim is released when it goes out of scope
    }
});

echo Swerve::claim('nightly-report')->available() ? "free again\n" : "still held\n";

if ($claim = Swerve::claim('nightly-report')->acquire()) {
    run_report();
}                                     // released when $claim goes out of scope

use Swerve\Swerve;

phasync::run(function () {
    $subscription = Swerve::subscribe('game');
    Swerve::publish('game', ['type' => 'move', 'player' => ['id' => 7, 'name' => 'ann'], 'to' => [3, 4]]);

    foreach ($subscription as $message) {
        echo $message->type, ' by ', $message->player->name, "\n";   // nested objects are sealed too
        var_dump(isset($message->end));

        foreach ($message as $name => $value) {
            echo "$name: ", get_debug_type($value), "\n";
        }

        try {
            $message->type = 'cheat';            // shared by every subscriber: refused
        } catch (LogicException $e) {
            echo get_class($e), "\n";
        }

        echo json_encode($message), "\n";       // as it was published
        break;
    }
});

foreach (Swerve::subscribe('game') as $message) {
    if ($message instanceof SealedObject && isset($message->end)) {   // a property
        break;
    }
    foreach ($message as $name => $value) { ... }                     // iterable
}

use Swerve\Util\SealedObject;

$value = SealedObject::seal(json_decode('[{"id": 1, "tags": {"a": true}}, "text", 3]'));

echo get_debug_type($value), "\n";                 // a list stays an array
echo get_debug_type($value[0]), "\n";              // its objects are sealed
echo get_debug_type($value[0]->tags), "\n";
echo $value[1], ' ', $value[2], "\n";