WebSocket

WebSocket is a lasting two-way connection between a browser and a server. In Winter a WebSocket server is a process: you describe whom to let in and what to do with a message, and the framework holds the connections, checks that clients are alive, and closes everything gently on stop.

What WebSocket is and why

HTTP is a conversation the client always starts: the browser asks, the server answers, the exchange is over. Until it is asked, the server cannot say anything.

The problem. Many applications need the other direction too: the server has something to tell now. A chat message arrived, a price changed, an order became “paid”, another player made a move. The workaround is to ask the server every second (“anything new?”). That is a request per second from every open window, almost always answered “no”, and still up to a second late.

The solution is WebSocket (the RFC 6455 protocol). The browser sends an ordinary HTTP request asking to upgrade, the server agrees, and the same TCP connection stays open. From then on both sides send each other messages whenever they want, with no new requests, until the connection is closed.

What Winter takes on. Holding thousands of open connections in one process, answering the protocol’s control signals, noticing clients that vanished without a word, and closing everything cleanly on stop is repetitive work full of traps. Winter gives you the WebSocket stereotype: you write a few hooks (“whom to let in”, “what to do with a message”, “someone left”) and the framework does the rest. WebSocket is a process: it starts, stops and reports its status like any other.

Quick start

Extend WebSocket and implement the one required hook, onMessage():

<?php

namespace Main;

use Flytachi\Winter\Kernel\WebSocket\Connection;
use Flytachi\Winter\Kernel\WebSocket\Frame;
use Flytachi\Winter\Kernel\WebSocket\WebSocket;

final class ChatSocket extends WebSocket
{
  protected function onMessage(Connection $conn, Frame $frame): void
  {
      // every message goes to everyone but the sender
      $this->broadcast($frame->data, except: $conn);
  }
}

Declare it in the application manifest with a port:

#[EnableWeb]
#[EnableWebSocket(\Main\ChatSocket::class, port: 8001)]
final class Application extends WinterApplication { /* ... */ }

After php call run the socket process comes up next to the web server. You can connect straight from the browser console:

const ws = new WebSocket('ws://localhost:8001');
ws.onmessage = (e) => console.log('received:', e.data);
ws.onopen = () => ws.send('hello everyone');

Open two tabs: what you send from one shows up in the other. That is already a working chat.

The class skeleton can be generated with make:

php call make .Chat -W      # → main/ChatSocket.php

Running it

A port is always required: one process, one port. There are three ways to run a socket.

With the application — an attribute on the application class. This is the main way:

#[EnableWeb]
#[EnableWebSocket(ChatSocket::class, port: 8001)]
#[EnableWebSocket(NotifySocket::class, port: 8002, host: '127.0.0.1')]
final class Application extends WinterApplication { /* ... */ }

On its own from the console — like any process, plus --port:

php call process main.ChatSocket start --port=8001          # in the foreground
php call process main.ChatSocket start --port=8001 -d       # in the background
php call process main.ChatSocket status                     # how it is doing
php call process main.ChatSocket stop                       # stop it gently

From code — in the background, from an admin panel for instance:

ChatSocket::dispatch(port: 8001);

host is the network interface. The default 0.0.0.0 means all interfaces. 127.0.0.1 keeps the socket reachable from this machine only — what you want when a proxy on the same host stands in front of it.

Settings

How the socket behaves is set with class properties:

final class ChatSocket extends WebSocket
{
  protected ?string $path = '/chat';
  protected ?array $allowedOrigins = ['https://example.com'];
  protected float $heartbeat = 30.0;
  protected int $maxConnections = 10_000;
  protected int $maxFrameSize = 64 * 1024;
  protected float $grace = 10.0;

  // ...
}

Dependencies are injected with the usual #[Autowired] — the socket is built by the container, like any process.

Hooks

Hooks are the methods you override. Only onMessage() is required.

onStart

Preparation and background work of the process — before the first connection.

Called once, when the server exists but does not accept clients yet. The framework waits for the hook to finish, so whatever is loaded here is guaranteed to be ready for the first connection. This is also where work not tied to any one client starts: a Redis channel subscription, a periodic broadcast, reading an external stream.

Syntax

protected function onStart(): void

Parameters

None.

Errors

An exception from onStart() stops the process: the socket does not start and the error goes to the log. Under call run with the web server, the master brings the process up again.

Example

protected function onStart(): void
{
  $this->rooms = $this->roomRepository->findAllActive();   // warm up

  $this->spawn(function (): void {
      while ($this->isRunning()) {
          $this->broadcast(json_encode(['online' => $this->count()]));
          $this->sleep(5);
      }
  });
}

onHandshake

Decides whether the client gets in.

Called before the protocol switch: at this point the connection is still an ordinary HTTP request. This is where access is checked — a token, a session cookie, the Origin header — and where whatever later hooks need (the user that was found, for instance) is stored on the connection, so it is not looked up again on every message. Nothing can be sent from here: the WebSocket is not open yet.

Syntax

protected function onHandshake(Connection $conn, HttpRequest $request): bool

Parameters

$conn — the connection to be. set(), get() and has() work, and what is stored carries over into the open connection. send() throws LogicException.

$request — the request the client came with: headers, cookies, query parameters, path, IP. The same object controllers get.

Return values

true — let in: the connection switches to WebSocket, then onOpen() is called. false — refuse: the client gets a 403.

By default the hook returns true, letting everyone in.

Errors

To refuse with an exact status, throw a ResponseException — the client gets its code and message, just as from a controller (handlers declared with #[AdviceException] apply too). Any other exception is a 500 and a call to onError().

Example

use Flytachi\Winter\Base\HttpCode;
use Flytachi\Winter\Kernel\Http\Response\ResponseException;

protected function onHandshake(Connection $conn, HttpRequest $request): bool
{
  $token = $request->getQueryParams()['token'] ?? null;
  $user = $token === null ? null : $this->tokens->verify($token);

  if ($user === null) {
      return false;                                                  // 403
  }
  if ($user->isBlocked()) {
      throw new ResponseException('account blocked', HttpCode::FORBIDDEN);
  }

  $conn->set('user', $user);
  return true;
}

protected ?array $allowedOrigins = ['https://example.com', 'https://app.example.com'];

onOpen

The connection is open — sending works.

Called right after the protocol switch. The place for a greeting, sending the message history, telling the others a newcomer has joined.

Syntax

protected function onOpen(Connection $conn): void

Parameters

$conn — the open connection, with everything stored in onHandshake().

Example

protected function onOpen(Connection $conn): void
{
  $name = $conn->get('user')->name;

  $conn->send(json_encode(['type' => 'welcome', 'name' => $name]));
  $this->broadcast(json_encode(['type' => 'joined', 'name' => $name]), except: $conn);
}

onMessage

A message from the client. The one required hook.

Called for every text or binary message. Messages of one connection arrive strictly in turn: the next one is not handled until this call is done. Messages of different connections are handled at the same time — a slow hook for one client does not hold up the others.

Syntax

abstract protected function onMessage(Connection $conn, Frame $frame): void

Parameters

$conn — the connection the message came from.

$frame — the message itself: $frame->data is its content as a string, $frame->isBinary is true for a binary message.

The protocol’s control signals (ping, pong, close) never come here — the framework answers them.

Example

The message format is yours to choose. A common approach is JSON with a type field, so one connection carries several kinds of messages:

protected function onMessage(Connection $conn, Frame $frame): void
{
  $msg = json_decode($frame->data, true);

  match ($msg['type'] ?? null) {
      'join' => $conn->set('room', (int) $msg['room']),
      'say'  => $this->sendToRoom($conn->get('room'), $conn->get('user')->name, $msg['text']),
      default => $conn->send(json_encode(['type' => 'error', 'text' => 'unknown type'])),
  };
}

private function sendToRoom(?int $room, string $from, string $text): void
{
  $payload = json_encode(['type' => 'say', 'from' => $from, 'text' => $text]);

  foreach ($this->connections() as $member) {
      if ($member->get('room') === $room) {
          $member->send($payload);
      }
  }
}

onClose

The connection has closed.

Called whoever closed the connection and for whatever reason: the client left, the server called close(), the client stopped answering, the process is stopping. Only for connections that passed onHandshake() — a refused client never comes here.

Syntax

protected function onClose(Connection $conn, int $code, string $reason): void

Parameters

$code — the close code:

$reason — the reason text, if there was one.

Example

protected function onClose(Connection $conn, int $code, string $reason): void
{
  $user = $conn->get('user');
  if ($user !== null) {
      $this->broadcast(json_encode(['type' => 'left', 'name' => $user->name]));
  }
}

onError

An exception in one of the hooks.

Called when onOpen(), onMessage() or onClose() threw, and for an exception in onHandshake() that is not a ResponseException. By default it logs the error — with the connection number and the client address — and keeps the connection open: one failed message does not cut the whole session short, just as an error in a controller does not close a keep-alive connection.

Syntax

protected function onError(Connection $conn, \Throwable $e): void

Parameters

$conn — the connection the error happened on.

$e — the exception.

Example

Override it when the client should hear about the error, or the connection is better closed:

protected function onError(Connection $conn, \Throwable $e): void
{
  parent::onError($conn, $e);   // to the log

  $conn->send(json_encode(['type' => 'error', 'text' => 'something went wrong']));

  $errors = (int) $conn->get('errors') + 1;
  $conn->set('errors', $errors);
  if ($errors >= 5) {
      $conn->close(1008, 'too many errors');
  }
}

Inherited from Process

WebSocket is a process, so its hooks are there too: onTerminate(), onReload(), onUser1(), onUser2() react to signals, and onShutdown() is the last thing before exit. There is no need to close connections on stop — the framework does it. More in Processes.

Socket methods

These methods are called on $this inside the class.

broadcast

Sends a text message to every open connection of this process.

The message is queued on each connection separately, so a slow client does not delay delivery to the rest, and the method itself returns at once.

Syntax

final protected function broadcast(string $data, ?Connection $except = null): int

Parameters

$data — the message text.

$except — a connection to skip; usually the sender. null by default — everyone.

Return values

The number of connections the message was queued to.

Example

$this->broadcast(json_encode(['type' => 'say', 'text' => $frame->data]), except: $conn);

connections

Returns the open connections of the process.

For selective sending: to a room, to one user, by role.

Syntax

final protected function connections(): array

Return values

An array of Connection, keyed by connection number.

Example

foreach ($this->connections() as $conn) {
  if ($conn->get('user')?->id === $userId) {
      $conn->send($text);
  }
}

connection

Finds an open connection by number.

Syntax

final protected function connection(int $id): ?Connection

Parameters

$id — the connection number ($conn->id).

Return values

The connection, or null if it is closed already.

count

Returns the number of open connections.

Syntax

final protected function count(): int

Example

$this->logger->info('online: ' . $this->count());

From Process

Everything a process has is available: spawn() — a background task, sleep() — a pause that a stop interrupts, isRunning() — false once the process is stopping, $this->logger — the logger.

The connection — Connection

Connection is the object of one client connection. The same object travels through every hook of the connection, from onHandshake() to onClose(), so what was put in it with set() is there for the next message and is gone with the connection.

send

Sends a text message.

The message is queued on the connection and goes out in the order sent. The method returns at once without waiting for the client to read it, so it can be called from anywhere: from a hook, from a background task, for the same connection at the same time.

Syntax

public function send(string $data): bool

Parameters

$data — the message text.

Return values

true — the message is queued. false — the connection is closed or closing, or the client stopped reading and its queue (1024 messages) is full; in that case the message is dropped.

Errors

LogicException — when called in onHandshake(), before the protocol switch.

sendBinary

Sends a binary message. Otherwise like send().

Syntax

public function sendBinary(string $data): bool

close

Closes the connection.

Messages sent before the call still go out; then the client gets the close signal with the code and reason, and onClose() is called with the same code. Calling it again does nothing.

Syntax

public function close(int $code = 1000, string $reason = ''): void

Parameters

$code — the close code. 1000 normal, 1008 policy violation, 1011 server error, 4000–4999 your own.

$reason — a short text for the client (at most 123 bytes).

Example

$conn->send(json_encode(['type' => 'kicked']));
$conn->close(4001, 'kicked by moderator');

isOpen

Returns true while messages can be sent. false in onHandshake() and after the close.

Syntax

public function isOpen(): bool

request

Returns the request the client connected with: headers, cookies, query parameters. Available for the whole life of the connection.

Syntax

public function request(): HttpRequest

set, get, has

The connection’s own value store.

Syntax

public function set(string $key, mixed $value): void
public function get(string $key): mixed          // null when the key is missing
public function has(string $key): bool           // a stored null counts

Example

$conn->set('room', 42);

if ($conn->has('room')) {
  $room = $conn->get('room');
}

The message — Frame

Frame is an incoming client message, what onMessage() receives.

Connection lifecycle

connect
├─ path does not match $path            → 404
├─ no Upgrade: websocket header         → 426
├─ Origin not in $allowedOrigins        → 403
├─ $maxConnections reached              → 503
└─ onHandshake()
      ├─ false                           → 403
      ├─ ResponseException               → its status
      ├─ any other exception             → 500 + onError()
      └─ true → protocol switch → onOpen()
                  │
                  ├─→ onMessage()  (exception → onError(), the connection lives on)
                  │      ↑___|
                  │
                  └─ client closed / no answer to ping / close() /
                     message too big / process stopping
                                         → onClose($code, $reason)

Liveness check (heartbeat)

A client can disappear without closing the connection: a phone went into the subway, a laptop lid was shut, a router dropped the connection from its table. The server does not learn about it — for the server the connection is open, messages “successfully” go nowhere, and a slot of $maxConnections is held by a ghost. Besides, proxies cut silent connections: nginx does so after 60 seconds by default.

So every $heartbeat seconds the framework sends the client a ping control signal. The browser answers with pong on its own — nothing to write in JavaScript. If nothing has come from the client by the next check, the connection is closed with code 1001 and onClose() is called. Any message from the client counts as a sign of life too.

The default is 30 seconds, below nginx’s timeout. $heartbeat = 0 turns the check off — for internal services without a proxy, say, where the client checks the link itself.

Stopping

On stop, on SIGTERM, or when the whole application stops, the socket:

1. stops accepting new connections;
2. sends every open connection a close with code 1001 (“going away”) — the browser gets onclose with this code and can reconnect to another instance;
3. waits for hooks still running — at most $grace seconds;
4. calls onShutdown() and exits.

Behind a proxy

In production the socket usually sits behind nginx on the same domain as the site. Then the front end does not need to know the port: the browser connects to wss://example.com/ws/chat, and the proxy hands the connection to the process’s port and takes care of TLS on the way.

location /ws/chat {
  proxy_pass http://app:8001;
  proxy_http_version 1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
  proxy_read_timeout 120s;
}

location /ws/notify {
  proxy_pass http://app:8002;
  proxy_http_version 1.1;
  proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
}

The Upgrade and Connection headers are required: without them nginx does not pass the upgrade request on, and the socket answers 426.

Sending from a controller

A frequent task: “the order is paid — tell the user in the open tab”. The controller runs in a web worker while the connections live in the socket process — these are different processes, and send() cannot be called from the controller directly. A go-between is needed; a Redis channel (pub/sub) is the handiest: the controller publishes, the socket listens in onStart() and hands it out.

use Flytachi\Winter\Redis\RedisPool;

public function markPaid(int $id): array
{
  $order = $this->orders->markPaid($id);

  RedisPool::store(MainRedisConfig::class)->publish('notify', json_encode([
      'userId' => $order->userId,
      'text'   => "Order #{$order->id} is paid",
  ]));

  return ['status' => 'ok'];
}

use Flytachi\Winter\Redis\RedisPool;

protected float $grace = 5.0;   // the subscription never ends by itself — do not wait forever

protected function onStart(): void
{
  $this->spawn(function (): void {
      // a subscription takes the whole connection — use one of its own, outside the pool
      $config = RedisPool::dedicated(MainRedisConfig::class);
      $redis  = $config->connection();
      $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);

      $redis->subscribe(['notify'], function (\Redis $redis, string $channel, string $payload): void {
          $msg = json_decode($payload, true);

          foreach ($this->connections() as $conn) {
              if ($conn->get('user')?->id === $msg['userId']) {
                  $conn->send($msg['text']);
              }
          }
      });
  });
}

The same approach solves several instances of the application: every socket instance subscribes to the channel and hands the message to its own clients. The kernel does not impose a delivery method — a message broker or something of your own works as well.

Example: a chat with rooms

A self-contained example: login by token, rooms, messages to a room, leaving.

<?php

namespace Main;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Http\Contracts\HttpRequest;
use Flytachi\Winter\Kernel\WebSocket\Connection;
use Flytachi\Winter\Kernel\WebSocket\Frame;
use Flytachi\Winter\Kernel\WebSocket\WebSocket;

final class RoomSocket extends WebSocket
{
  protected int $maxFrameSize = 16 * 1024;

  #[Autowired]
  private TokenService $tokens;

  protected function onHandshake(Connection $conn, HttpRequest $request): bool
  {
      $user = $this->tokens->verify($request->getQueryParams()['token'] ?? '');
      if ($user === null) {
          return false;
      }
      $conn->set('name', $user->name);
      return true;
  }

  protected function onMessage(Connection $conn, Frame $frame): void
  {
      $msg = json_decode($frame->data, true);

      if (($msg['type'] ?? null) === 'join') {
          $conn->set('room', (string) $msg['room']);
          $this->toRoom($conn->get('room'), ['type' => 'joined', 'name' => $conn->get('name')]);
          return;
      }

      if (($msg['type'] ?? null) === 'say' && $conn->has('room')) {
          $this->toRoom($conn->get('room'), [
              'type' => 'say',
              'name' => $conn->get('name'),
              'text' => mb_substr((string) $msg['text'], 0, 2000),
          ]);
      }
  }

  protected function onClose(Connection $conn, int $code, string $reason): void
  {
      if ($conn->has('room')) {
          $this->toRoom($conn->get('room'), ['type' => 'left', 'name' => $conn->get('name')]);
      }
  }

  private function toRoom(string $room, array $message): void
  {
      $payload = json_encode($message);
      foreach ($this->connections() as $member) {
          if ($member->get('room') === $room) {
              $member->send($payload);
          }
      }
  }
}

const ws = new WebSocket('ws://localhost:8001?token=YOUR_TOKEN');
ws.onmessage = (e) => console.log(JSON.parse(e.data));
ws.onopen = () => {
  ws.send(JSON.stringify({ type: 'join', room: 'general' }));
  ws.send(JSON.stringify({ type: 'say', text: 'hello everyone' }));
};

Next

• Processes — what the socket is built on: signals, spawn(), control
• Application components — where #[EnableWebSocket] is declared
• The process command — start, status and stop from the console
• The make command — a socket skeleton with the -W flag
• Redis — channels connecting the socket to controllers and other instances