websockets.md

WebSockets

Atomic ships a routed WebSocket server in Engine\Atomic\Plugins\WebSockets\RoutedWebSocketServer, a reusable server base class in Engine\Atomic\Plugins\WebSockets\Server, and a thin connection wrapper in Engine\Atomic\Plugins\WebSockets\Connection.

Current scope

The repository includes:

• a routed WebSocket server for route-only applications
• an abstract server base class for custom lifecycle/pubsub behavior
• a connection wrapper
• route registration and message dispatch helpers
• a single-shot test client utility class

The repository does not include application-specific WebSocket handlers by default.

Enable Engine\Atomic\Plugins\WebSockets\WebSockets as a plugin before loading WebSocket routes. The plugin owns the Workerman dependencies; run composer install in engine/Atomic/Plugins/WebSockets when the root application does not already provide them.

Minimal bootstrap for a routed WebSocket worker:

use Engine\Atomic\Core\App;
use Engine\Atomic\Plugins\WebSockets\RoutedWebSocketServer;
use Engine\Atomic\Plugins\WebSockets\WebSockets;

$app = App::instance($atomic)
    ->register_config()
    ->register_middleware()
    ->register_core_plugins(WebSockets::class)
    ->register_plugins()
    ->register_routes();

(new RoutedWebSocketServer($app->get('WS.listen'), 2))->run();

WebSockets::register() adds the websocket route type. During register_routes(), Atomic loads app and plugin routes/websocket.php files before the routed server starts.

Route dispatch

WebSocket routes are registered through the plugin router in routes/websocket.php:

use Engine\Atomic\Plugins\WebSockets\WebSocketRouter;

WebSocketRouter::register('/jobs/@job_id', App\WebSockets\JobsHandler::class . '::receive', [
    'ws-auth',
    'ws-rate-limit:jobs',
]);

Route middleware aliases are registered in config/middleware.php through the same alias registry as HTTP middleware. WebSocket middleware classes must implement Engine\Atomic\Plugins\WebSockets\WebSocketMiddleware; they do not need to implement the HTTP MiddlewareInterface.

When a message is dispatched, WebSocketDispatcher matches the connection path, resolves the route's message middleware aliases, runs them in order, and then calls the handler. Returning false from WebSocket middleware stops dispatch for that message.

If the middleware array includes phase keys, message is used for message dispatch and connect is ignored by the message dispatcher:

WebSocketRouter::register('/jobs/@job_id', App\WebSockets\JobsHandler::class . '::receive', [
    'connect' => ['ws-origin-check'],
    'message' => ['ws-auth', 'ws-rate-limit:jobs'],
]);

Routed server

For route-only WebSocket applications, start the bundled routed server:

use Engine\Atomic\Plugins\WebSockets\RoutedWebSocketServer;
use Engine\Atomic\Plugins\WebSockets\Server;

(new RoutedWebSocketServer(
    'tcp://0.0.0.0:8080',
    2,
    Server::MODE_FOREGROUND
))->run();

The WebSockets plugin registers the websocket route type through the normal plugin lifecycle. Load app and plugin routes/websocket.php files during application/plugin bootstrap before starting RoutedWebSocketServer; the server dispatches each message through WebSocketDispatcher, rejects unknown paths cleanly, and leaves connect/disconnect hooks empty. Extend Server directly when the application needs custom connection lifecycle, pub/sub, or task mapping behavior.

Server modes

WebSocket servers accept an explicit mode as the third constructor argument. Use the class constants instead of raw strings:

use Engine\Atomic\Plugins\WebSockets\RoutedWebSocketServer;
use Engine\Atomic\Plugins\WebSockets\Server;

new RoutedWebSocketServer($listen, 2, Server::MODE_FOREGROUND);
new RoutedWebSocketServer($listen, 2, Server::MODE_BACKGROUND);

Available modes:

• Server::MODE_FOREGROUND (foreground) runs Workerman attached to the current process. This is the default and is best for local development, terminal logs, containers, and process managers that expect the command to stay attached. Workerman reports this internally as DEBUG mode, but Atomic does not use it to mean verbose logging.
• Server::MODE_BACKGROUND (background) runs Workerman detached in the background. Atomic passes Workerman start -d, writes the pid file under LOGS/ws, and keeps stdout detached.

Only one mode can be active. Passing any other value throws an InvalidArgumentException before Workerman starts.

Testing

Default unit tests should not require a WebSocket server, Redis server, or any other external process to already be running. Keep dispatcher, router, route-loader, and scaffold coverage process-free.

Use a smoke test to verify plugin dependencies can autoload after running Composer in the plugin directory:

cd engine/Atomic/Plugins/WebSockets
composer install

Longer end-to-end tests that start a real WebSocket worker should live in an integration suite, own the worker lifecycle, choose a free local port, clean up after themselves, and be gated behind an environment flag such as RUN_WS_INTEGRATION=1.

Base server API

Required methods in subclasses:

• setup(): void
• on_websocket_connect(Connection $conn, Request $request): void
• on_message(Connection $conn, string $data, int $op): void
• on_disconnect(Connection $conn): void

Optional hook:

• on_worker_start(): void

Runtime behavior

When run() is called:

1. worker_count must be at least 1, otherwise an InvalidArgumentException is thrown.
2. If the listen string starts with tcp://, it is rewritten to websocket://.
3. A Workerman Worker is created for that listener.
4. LOGS/ws is used for pid and log files. LOGS must be configured.
5. setup() is executed before Workerman callbacks are assigned.
6. On worker start, Redis async client initialization runs, then on_worker_start(), then optional pubsub startup.
7. The server runs in foreground or daemon mode through Workerman\Worker::runAll().

Generated filenames:

• workerman.<lowercased-server-class>.pid
• workerman.<lowercased-server-class>.log

Example class tag:

App\WebSockets\JobsServer -> workerman.app.websockets.jobsserver.*

Connection wrapper

Connection wraps Workerman TcpConnection and exposes:

• id(): string
• socket_int(): int
• path(): string
• send(string $data): bool
• close(): void

Use the wrapper in your hooks instead of accessing the raw Workerman connection directly.

Task mapping

The base server keeps two in-memory structures:

• task_id -> socket_int
• socket_int -> task_id[]

Helpers available to subclasses:

• subscribe_to_channel(string $channel): void
• map_task(string $task_id, int $socket_int): void
• get_socket_tasks(int $socket_int): array
• unmap_task(string $task_id): void

Redis pubsub bridge

If subscribe_to_channel(...) was called in setup(), run() will start a Redis pubsub listener on worker start.

Behavior of start_pubsub(...):

1. Connects to Redis using REDIS.host, REDIS.port, optional REDIS.password, and optional REDIS.db.
2. Subscribes to the configured channel.
3. Decodes each payload as JSON.
4. Reads task_id.
5. Finds the mapped socket.
6. Sends the original payload back to that socket.
7. Unmaps the task.

Expected payload shape:

{
  "task_id": "uuid-or-id",
  "status": "completed",
  "data": {}
}

If the payload is not valid JSON or does not contain task_id, it is ignored.

Minimal server example

<?php
declare(strict_types=1);

namespace App\WebSockets;

use Engine\Atomic\Plugins\WebSockets\Connection;
use Engine\Atomic\Plugins\WebSockets\Server;
use Workerman\Protocols\Http\Request;

final class JobsServer extends Server
{
    protected function setup(): void
    {
        $this->subscribe_to_channel('queue:results');
    }

    protected function on_websocket_connect(Connection $conn, Request $request): void
    {
        $conn->send(json_encode(['type' => 'welcome', 'id' => $conn->id()]));
    }

    protected function on_message(Connection $conn, string $data, int $op): void
    {
        $msg = json_decode($data, true);
        if (!is_array($msg) || empty($msg['task_id'])) {
            $conn->send(json_encode(['error' => 'task_id required']));
            return;
        }

        $this->map_task((string)$msg['task_id'], $conn->socket_int());
    }

    protected function on_disconnect(Connection $conn): void
    {
        foreach ($this->get_socket_tasks($conn->socket_int()) as $taskId) {
            $this->unmap_task($taskId);
        }
    }
}

Operational notes

• REDIS is read from application configuration; the framework config loaders default it to 127.0.0.1:6379.
• LOGS is mandatory for both the server and the bundled test client.
• Use Server::MODE_BACKGROUND when the worker should detach; use Server::MODE_FOREGROUND when it should stay attached.