Menu

Reference

Reference / Http\WebSocket

Hub

finalclass
final class Hub

A set of connections that a message can be sent to at once.

Closed connections, and ones whose send fails, are dropped from the set. The broadcast is sequential: each send returns only once that client has taken the whole frame, so a slow client delays the ones after it. It suits small rooms and small messages; for a wide or large fan-out, give each connection its own writer task fed from a bounded queue.

Example

// Callback style: a Handler on each side, and a Hub that fans a message out.

use function Async\async;
use function Async\spawn;
use Http\WebSocket as WS;

final class Greeter implements WS\Handler
{
    public function __construct(private WS\Hub $hub) {}

    public function onOpen(WS\Connection $c): void
    {
        $this->hub->add($c);
        $c->send('welcome');
    }

    public function onMessage(WS\Connection $c, WS\Message $m): void
    {
        $n = $this->hub->broadcast('said: ' . $m->data);
        echo "server: broadcast to $n\n";
    }

    public function onClose(WS\Connection $c, int $code, string $reason): void
    {
        $this->hub->remove($c);
        echo "server: closed with $code '$reason', hub has ", $this->hub->count(), "\n";
    }

    public function onError(WS\Connection $c, \Throwable $e): void
    {
        echo "server: error ", $e->getMessage(), "\n";
    }
}

$l = stream_socket_server('tcp://127.0.0.1:0');
stream_set_blocking($l, false);
$name = stream_socket_get_name($l, false);
$port = (int)substr($name, strrpos($name, ':') + 1);
$server = \Http\Server::onListener($l)->acceptWait(0.02);
$hub = new WS\Hub();

async(function () use ($server, $port, $hub) {
    spawn(function () use ($server, $hub) {
        $server->serve(function (\Http\Request $req) use ($hub): \Http\Response {
            return WS\upgrade($req, WS\handler(new Greeter($hub)));
        });
    });

    $ws = WS\connect("ws://127.0.0.1:$port/");
    $m = $ws->receive();
    echo 'client: ', $m->data, "\n";
    $ws->send('hi');
    $m = $ws->receive();
    echo 'client: ', $m->data, "\n";
    $ws->close(4000, 'bye');
    \Async\delay(0.05);
    echo 'client: peer close code ', $ws->closeCode(), "\n";
    $server->stop();
});

Output

client: welcome
server: broadcast to 1
client: said: hi
server: closed with 4000 'bye', hub has 0
client: peer close code 4000

See also: Connection

Methods

add()Add a connection to the set.
remove()Remove a connection from the set.
count()The number of connections in the set, closed ones not yet dropped included.
broadcast()Send a message to every open connection in the set.

add()#source

function add(Connection $c): void

Add a connection to the set. Adding one already in the set changes nothing.

Parameters

$c Connection
the connection

remove()#source

function remove(Connection $c): void

Remove a connection from the set. Removing one that is not in it changes nothing.

Parameters

$c Connection
the connection

count()#source

function count(): int

The number of connections in the set, closed ones not yet dropped included.

Returns

int — the number of connections

broadcast()#source

function broadcast(string $data, bool $binary = false): int

Send a message to every open connection in the set.

Parameters

$data string
the message
$binary bool = false
true to send it as a binary message, false for text

Returns

int — how many connections the message was sent to

View source