1372 lines
46 KiB
PHP
1372 lines
46 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace BlaxSoftware\LaravelWebSockets\Websocket;
|
|
|
|
use BlaxSoftware\LaravelWebSockets\Apps\App;
|
|
use BlaxSoftware\LaravelWebSockets\Channels\Channel;
|
|
use BlaxSoftware\LaravelWebSockets\Channels\PresenceChannel;
|
|
use BlaxSoftware\LaravelWebSockets\Channels\PrivateChannel;
|
|
use BlaxSoftware\LaravelWebSockets\Contracts\ChannelManager;
|
|
use BlaxSoftware\LaravelWebSockets\Events\ConnectionClosed;
|
|
use BlaxSoftware\LaravelWebSockets\Events\NewConnection;
|
|
use BlaxSoftware\LaravelWebSockets\Ipc\SocketPairIpc;
|
|
use BlaxSoftware\LaravelWebSockets\Websocket\MockConnectionSocketPair;
|
|
use BlaxSoftware\LaravelWebSockets\Server\Exceptions\ConnectionsOverCapacity;
|
|
use BlaxSoftware\LaravelWebSockets\Server\Exceptions\OriginNotAllowed;
|
|
use BlaxSoftware\LaravelWebSockets\Server\Exceptions\UnknownAppKey;
|
|
use BlaxSoftware\LaravelWebSockets\Server\Exceptions\WebSocketException as ExceptionsWebSocketException;
|
|
use BlaxSoftware\LaravelWebSockets\Server\QueryParameters;
|
|
use Exception;
|
|
use Illuminate\Support\Facades\Auth;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Log;
|
|
use Ratchet\ConnectionInterface;
|
|
use Ratchet\RFC6455\Messaging\MessageInterface;
|
|
use Ratchet\WebSocket\MessageComponentInterface;
|
|
|
|
class Handler implements MessageComponentInterface
|
|
{
|
|
/**
|
|
* Track channel connections using associative arrays for O(1) lookup
|
|
* Structure: [channel_name => [socket_id => true]]
|
|
*/
|
|
protected array $channel_connections = [];
|
|
|
|
/**
|
|
* Cache write buffer for batching operations
|
|
* Reduces file I/O when multiple rapid requests occur
|
|
*/
|
|
protected array $cacheWriteBuffer = [];
|
|
protected array $cacheDeleteBuffer = [];
|
|
protected bool $cacheBufferScheduled = false;
|
|
|
|
/**
|
|
* Pre-encoded static JSON responses for performance
|
|
* Encoding once at startup is faster than encoding every time
|
|
*/
|
|
private static string $PONG_RESPONSE = '{"event":"pusher.pong"}';
|
|
|
|
/**
|
|
* GC collection counter - only collect every N pings
|
|
*/
|
|
private int $gcCounter = 0;
|
|
private const GC_INTERVAL = 100;
|
|
|
|
/**
|
|
* Whether hot reload is enabled (cached for performance)
|
|
*/
|
|
private static ?bool $hotReload = null;
|
|
|
|
/**
|
|
* Whether debug mode is enabled (cached to avoid container resolution per message)
|
|
*/
|
|
private static ?bool $debug = null;
|
|
|
|
/**
|
|
* Track active child processes to limit concurrent DB connections.
|
|
* Each forked child may open its own MySQL connection, so we must
|
|
* cap concurrency to avoid exhausting MySQL's max_connections.
|
|
*/
|
|
private int $activeChildCount = 0;
|
|
|
|
/**
|
|
* Maximum concurrent child processes (and thus DB connections).
|
|
* Configurable via websockets.max_concurrent_children config.
|
|
* Default 50 leaves headroom for PHP-FPM, queue workers, etc.
|
|
*/
|
|
private int $maxConcurrentChildren = 50;
|
|
|
|
/**
|
|
* Queue of deferred messages waiting for a child slot.
|
|
* Each entry is [ConnectionInterface, Channel, array $message].
|
|
*/
|
|
private array $deferredMessages = [];
|
|
|
|
/**
|
|
* Initialize a new handler.
|
|
*/
|
|
public function __construct(
|
|
protected ChannelManager $channelManager
|
|
) {
|
|
$this->maxConcurrentChildren = (int) config('websockets.max_concurrent_children', 50);
|
|
}
|
|
|
|
/**
|
|
* Handle incoming WebSocket message with optimized fast path for ping/pong
|
|
*/
|
|
public function onMessage(
|
|
ConnectionInterface $connection,
|
|
MessageInterface $message
|
|
): void {
|
|
if (!isset($connection->app)) {
|
|
return;
|
|
}
|
|
|
|
// FAST PATH: Check for ping before any heavy processing
|
|
// Use raw string comparison on payload to avoid JSON decode overhead
|
|
$payload = $message->getPayload();
|
|
|
|
// Quick ping detection using strpos (faster than json_decode + array access)
|
|
if ($this->tryHandlePingFast($payload, $connection)) {
|
|
return;
|
|
}
|
|
|
|
// SLOW PATH: Full message processing
|
|
try {
|
|
$this->processFullMessage($connection, $message, $payload);
|
|
} catch (\Throwable $e) {
|
|
$this->handleMessageError($e);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Fast path for ping/pong - avoids JSON decode, object creation, promises
|
|
* Target: < 1ms processing time
|
|
*/
|
|
private function tryHandlePingFast(string $payload, ConnectionInterface $connection): bool
|
|
{
|
|
// Quick string check - if doesn't contain "ping", skip fast path
|
|
// strpos is O(n) but very fast for short strings
|
|
if (strpos($payload, 'ping') === false) {
|
|
return false;
|
|
}
|
|
|
|
// Now do minimal JSON decode to confirm it's a ping
|
|
$data = json_decode($payload, true);
|
|
if ($data === null) {
|
|
return false;
|
|
}
|
|
|
|
$event = $data['event'] ?? '';
|
|
|
|
// Direct string comparison (faster than strtolower + comparison)
|
|
if ($event !== 'pusher:ping' && $event !== 'pusher.ping') {
|
|
return false;
|
|
}
|
|
|
|
// Update connection pong timestamp in BOTH local memory AND Redis sorted set.
|
|
// This is critical: removeObsoleteConnections() checks the Redis set score,
|
|
// so a direct $connection->lastPongedAt assignment alone is insufficient —
|
|
// the Redis-based cleanup would still unsubscribe channels after 120s.
|
|
// connectionPonged() is async (returns a Promise resolved by the event loop),
|
|
// so this does not block the ping response.
|
|
$this->channelManager->connectionPonged($connection)
|
|
->then(null, function (\Throwable $e) use ($connection) {
|
|
// If the Redis pong update fails, the connection will appear stale
|
|
// and removeObsoleteConnections() will unsubscribe its channels.
|
|
// Log this so we can diagnose connection drops.
|
|
Log::channel('websocket')->error('connectionPonged failed for ' . ($connection->socketId ?? '?') . ': ' . $e->getMessage());
|
|
});
|
|
|
|
// Send pre-encoded pong response immediately
|
|
$connection->send(self::$PONG_RESPONSE);
|
|
|
|
// Periodic GC instead of every ping
|
|
if (++$this->gcCounter >= self::GC_INTERVAL) {
|
|
$this->gcCounter = 0;
|
|
gc_collect_cycles();
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Debug ping latency - call this to measure server-side processing time
|
|
* Add to onMessage: $start = hrtime(true); ... $this->logPingLatency($start);
|
|
*/
|
|
protected function logPingLatency(int $startNs): void
|
|
{
|
|
$elapsed = (hrtime(true) - $startNs) / 1_000_000; // Convert to ms
|
|
if ($elapsed > 1.0) {
|
|
Log::channel('websocket')->warning('Slow ping: ' . round($elapsed, 2) . 'ms');
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Full message processing for non-ping messages
|
|
*/
|
|
private function processFullMessage(
|
|
ConnectionInterface $connection,
|
|
MessageInterface $message,
|
|
string $payload
|
|
): void {
|
|
// Any received message proves the client is alive — update local pong timestamp.
|
|
// This is a safety net for LocalChannelManager::removeObsoleteConnections().
|
|
// The primary Redis score update happens in tryHandlePingFast() via connectionPonged().
|
|
$connection->lastPongedAt = time();
|
|
|
|
// Set remote address once (moved from per-message to reduce overhead)
|
|
if (isset($connection->remoteAddress)) {
|
|
request()->server->set('REMOTE_ADDR', $connection->remoteAddress);
|
|
}
|
|
|
|
// Decode message (we already have payload string)
|
|
$messageArray = json_decode($payload, true, 512, JSON_THROW_ON_ERROR);
|
|
|
|
// Handle pusher protocol messages (subscribe, unsubscribe, etc.)
|
|
$this->handlePusherProtocolMessage($message, $connection, $messageArray);
|
|
|
|
$channel = $this->handleChannelSubscriptions($messageArray, $connection);
|
|
|
|
if ($this->shouldRejectMessage($channel, $connection, $messageArray)) {
|
|
return;
|
|
}
|
|
|
|
$this->authenticateConnection($connection, $channel, $messageArray);
|
|
|
|
// Only log in debug mode to reduce I/O
|
|
if (self::$debug ??= (bool) config('app.debug')) {
|
|
Log::channel('websocket')->debug('[' . $connection->socketId . ']@' . $channel->getName() . ' | ' . $payload);
|
|
}
|
|
|
|
if ($this->handlePusherEvent($messageArray, $connection)) {
|
|
return;
|
|
}
|
|
|
|
$this->forkWithSocketPair($connection, $channel, $messageArray);
|
|
}
|
|
|
|
/**
|
|
* Handle pusher protocol messages (formerly in PusherMessageFactory)
|
|
* Inlined for performance - avoids object creation
|
|
*/
|
|
private function handlePusherProtocolMessage(
|
|
MessageInterface $message,
|
|
ConnectionInterface $connection,
|
|
array $messageArray
|
|
): void {
|
|
$event = $messageArray['event'] ?? '';
|
|
|
|
// Fast check - most messages don't start with 'pusher' or 'client-'
|
|
$firstChar = $event[0] ?? '';
|
|
if ($firstChar !== 'p' && $firstChar !== 'c') {
|
|
return;
|
|
}
|
|
|
|
// Check for client- messages
|
|
if (strpos($event, 'client-') === 0) {
|
|
if (!$connection->app->clientMessagesEnabled) {
|
|
return;
|
|
}
|
|
|
|
$channelName = $messageArray['channel'] ?? null;
|
|
if (!$channelName) {
|
|
return;
|
|
}
|
|
|
|
$channel = $this->channelManager->find($connection->app->id, $channelName);
|
|
if ($channel) {
|
|
$channel->broadcastToEveryoneExcept(
|
|
(object) $messageArray,
|
|
$connection->socketId,
|
|
$connection->app->id
|
|
);
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Check for pusher: or pusher. messages (subscribe/unsubscribe handled elsewhere)
|
|
// This is handled by handleChannelSubscriptions for subscribe/unsubscribe
|
|
}
|
|
|
|
public function onOpen(ConnectionInterface $connection): void
|
|
{
|
|
if (!$this->connectionCanBeMade($connection)) {
|
|
$connection->close();
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$this->setupConnectionAddress($connection);
|
|
$this->verifyAppKey($connection);
|
|
$this->verifyOrigin($connection);
|
|
$this->limitConcurrentConnections($connection);
|
|
$this->generateSocketId($connection);
|
|
$this->establishConnection($connection);
|
|
$this->initializeAppConnection($connection);
|
|
} catch (UnknownAppKey $e) {
|
|
Log::channel('websocket')->error('Root level error: ' . $e->getMessage(), [
|
|
'file' => $e->getFile(),
|
|
'line' => $e->getLine(),
|
|
'trace' => $e->getTraceAsString(),
|
|
]);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle the websocket close.
|
|
*/
|
|
public function onClose(ConnectionInterface $connection): void
|
|
{
|
|
$this->authenticateConnection($connection, null);
|
|
|
|
if (isset($connection->remoteAddress)) {
|
|
request()->server->set('REMOTE_ADDR', $connection->remoteAddress);
|
|
}
|
|
|
|
$this->cleanupChannelConnections($connection);
|
|
$this->finalizeConnectionClose($connection);
|
|
|
|
// Clean up per-connection session from Redis
|
|
cache()->forget('ws_session_' . $connection->socketId);
|
|
}
|
|
|
|
|
|
protected function setupConnectionAddress(ConnectionInterface $connection): void
|
|
{
|
|
$connection->remoteAddress = trim(
|
|
explode(
|
|
',',
|
|
$connection->httpRequest->getHeaderLine('X-Forwarded-For')
|
|
)[0] ?? $connection->remoteAddress
|
|
);
|
|
request()->server->set('REMOTE_ADDR', $connection->remoteAddress);
|
|
Log::channel('websocket')->info('WS onOpen IP: ' . $connection->remoteAddress);
|
|
}
|
|
|
|
protected function initializeAppConnection(ConnectionInterface $connection): void
|
|
{
|
|
if (!isset($connection->app)) {
|
|
return;
|
|
}
|
|
|
|
// Register connection pong in both local memory and Redis sorted set.
|
|
// The Redis score is checked by removeObsoleteConnections() every 10s.
|
|
// Without this, the connection wouldn't have a Redis score until the
|
|
// first channel subscription, leaving a window for stale removal.
|
|
$this->channelManager->connectionPonged($connection);
|
|
|
|
$this->channelManager->subscribeToApp($connection->app->id);
|
|
|
|
NewConnection::dispatch(
|
|
$connection->app->id,
|
|
$connection->socketId
|
|
);
|
|
}
|
|
|
|
protected function shouldRejectMessage(?Channel $channel, ConnectionInterface $connection, array $message): bool
|
|
{
|
|
$event = $message['event'] ?? '';
|
|
$isUnsubscribe = $event === 'pusher:unsubscribe' || $event === 'pusher.unsubscribe';
|
|
|
|
if (!$channel?->hasConnection($connection) && !$isUnsubscribe) {
|
|
$connection->send(json_encode([
|
|
'event' => $event . ':error',
|
|
'data' => [
|
|
'message' => 'Subscription not established',
|
|
'meta' => $message,
|
|
],
|
|
]));
|
|
return true;
|
|
}
|
|
|
|
if (!$channel) {
|
|
$connection->send(json_encode([
|
|
'event' => $message['event'] . ':error',
|
|
'data' => [
|
|
'message' => 'Channel not found',
|
|
'meta' => $message,
|
|
],
|
|
]));
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
protected function handlePusherEvent(array $message, ConnectionInterface $connection): bool
|
|
{
|
|
if (!str_contains($message['event'], 'pusher')) {
|
|
return false;
|
|
}
|
|
|
|
$connection->send(json_encode([
|
|
'event' => $message['event'] . ':response',
|
|
'data' => [
|
|
'message' => 'Success',
|
|
],
|
|
]));
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Check if hot reload mode is enabled
|
|
*/
|
|
protected static function isHotReload(): bool
|
|
{
|
|
if (self::$hotReload === null) {
|
|
self::$hotReload = (bool) config('websockets.hot_reload', false);
|
|
}
|
|
return self::$hotReload;
|
|
}
|
|
|
|
/**
|
|
* Hot reload: Clear all caches in child process for fresh code loading
|
|
* This allows Models, Resources, Services, and everything else to be reloaded
|
|
* without restarting the WebSocket server.
|
|
*
|
|
* Only called when websockets.hot_reload is enabled.
|
|
*/
|
|
protected function hotReloadChild(): void
|
|
{
|
|
if (!self::isHotReload()) {
|
|
return;
|
|
}
|
|
|
|
// 1. Clear OPcache - forces PHP to recompile files from disk
|
|
if (function_exists('opcache_reset')) {
|
|
opcache_reset();
|
|
}
|
|
|
|
// 2. Clear Laravel's compiled services and config cache in container
|
|
$container = \Illuminate\Container\Container::getInstance();
|
|
|
|
// 3. Flush resolved instances - forces fresh instantiation
|
|
// This clears all singleton instances so they get rebuilt
|
|
$container->forgetScopedInstances();
|
|
|
|
// 4. Clear config repository cache (forces fresh config reads)
|
|
// Re-read all config files from disk
|
|
try {
|
|
/** @var \Illuminate\Config\Repository $config */
|
|
$config = $container->make('config');
|
|
|
|
// Get the path to config files
|
|
$configPath = base_path('config');
|
|
|
|
if (is_dir($configPath)) {
|
|
$files = glob($configPath . '/*.php');
|
|
foreach ($files as $file) {
|
|
$key = basename($file, '.php');
|
|
// Invalidate opcache for this config file
|
|
if (function_exists('opcache_invalidate')) {
|
|
opcache_invalidate($file, true);
|
|
}
|
|
// Force re-require the config file
|
|
$freshConfig = require $file;
|
|
$config->set($key, $freshConfig);
|
|
}
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Config refresh failed, continue anyway
|
|
Log::channel('websocket')->debug('Hot reload config refresh failed: ' . $e->getMessage());
|
|
}
|
|
|
|
// 5. Clear view cache (if views are being used in responses)
|
|
try {
|
|
if ($container->bound('view')) {
|
|
$container->forgetInstance('view');
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// View refresh failed, continue anyway
|
|
}
|
|
|
|
// 6. Clear route cache (if routes are dynamically resolved)
|
|
try {
|
|
if ($container->bound('router')) {
|
|
$container->forgetInstance('router');
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Router refresh failed, continue anyway
|
|
}
|
|
|
|
// 7. Clear translation cache
|
|
try {
|
|
if ($container->bound('translator')) {
|
|
$container->forgetInstance('translator');
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Translator refresh failed, continue anyway
|
|
}
|
|
|
|
// 8. Clear validation factory (for custom rules)
|
|
try {
|
|
if ($container->bound('validator')) {
|
|
$container->forgetInstance('validator');
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Validator refresh failed, continue anyway
|
|
}
|
|
|
|
// 9. Clear event dispatcher cache (for fresh event/listener bindings)
|
|
try {
|
|
if ($container->bound('events')) {
|
|
$container->forgetInstance('events');
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Events refresh failed, continue anyway
|
|
}
|
|
|
|
// 10. Clear WebSocket ControllerResolver cache for fresh controller loading
|
|
ControllerResolver::clearCache();
|
|
|
|
Log::channel('websocket')->debug('Hot reload: caches cleared in child process');
|
|
}
|
|
|
|
/**
|
|
* Fork with event-driven socket pair IPC (no polling!)
|
|
* Parent is notified INSTANTLY when child sends data
|
|
*
|
|
* Includes a concurrency limiter: if max_concurrent_children is reached,
|
|
* the message is queued and processed when a child slot frees up.
|
|
* This prevents exhausting MySQL's max_connections under load.
|
|
*/
|
|
protected function forkWithSocketPair(
|
|
ConnectionInterface $connection,
|
|
Channel $channel,
|
|
array $message
|
|
): void {
|
|
// Check concurrency limit before forking
|
|
if ($this->activeChildCount >= $this->maxConcurrentChildren) {
|
|
// Queue the message for later processing
|
|
$this->deferredMessages[] = [$connection, $channel, $message];
|
|
|
|
if (count($this->deferredMessages) === 1) {
|
|
// Log only on first deferral to avoid log spam
|
|
Log::channel('websocket')->warning('Fork concurrency limit reached (' . $this->maxConcurrentChildren . '), queueing message', [
|
|
'active_children' => $this->activeChildCount,
|
|
'queued' => count($this->deferredMessages),
|
|
]);
|
|
}
|
|
return;
|
|
}
|
|
|
|
$this->activeChildCount++;
|
|
|
|
// Create socket pair BEFORE fork
|
|
$ipc = SocketPairIpc::create($this->channelManager->loop);
|
|
|
|
$pid = pcntl_fork();
|
|
|
|
if ($pid === -1) {
|
|
$this->activeChildCount--;
|
|
Log::error('Fork error');
|
|
$this->processDeferredMessages();
|
|
return;
|
|
}
|
|
|
|
if ($pid === 0) {
|
|
// === CHILD PROCESS ===
|
|
$ipc->setupChild();
|
|
|
|
// Hot reload: clear all caches for fresh code loading (only in dev mode)
|
|
$this->hotReloadChild();
|
|
|
|
try {
|
|
// Lazy DB reconnect: disconnect now, reconnect only when first query runs
|
|
// This saves ~5-15ms for methods that don't use the database
|
|
DB::disconnect();
|
|
|
|
// Purge inherited Redis/cache connections from parent process.
|
|
// After fork(), child inherits parent's Redis socket fd — using it
|
|
// would corrupt parent's protocol state. Purging forces fresh
|
|
// connections on next cache() call (predis connects lazily).
|
|
app()->forgetInstance('cache');
|
|
app()->forgetInstance('cache.store');
|
|
app()->forgetInstance('redis');
|
|
|
|
// Configure DB reconnect-on-lost-connection for this child.
|
|
// If MySQL returns "Too many connections" or "server has gone away",
|
|
// Laravel will retry the query once after reconnecting.
|
|
try {
|
|
$dbConfig = config('database.connections.' . config('database.default'), []);
|
|
if (empty($dbConfig['retry_on_connection_loss'] ?? null)) {
|
|
config(['database.connections.' . config('database.default') . '.retry_on_connection_loss' => true]);
|
|
}
|
|
} catch (\Throwable $e) {
|
|
// Non-critical, continue without retry config
|
|
}
|
|
|
|
$this->setRequest($message, $connection);
|
|
|
|
// Set up per-connection session (backed by Redis)
|
|
$session = new ConnectionSession($connection->socketId);
|
|
app()->instance('ws.session', $session);
|
|
|
|
// Create mock that sends via socket pair
|
|
$mock = new MockConnectionSocketPair($connection, $ipc);
|
|
|
|
$this->executeControllerWithDbResilience(
|
|
$mock,
|
|
$channel,
|
|
$message,
|
|
$this->channelManager
|
|
);
|
|
|
|
\Illuminate\Container\Container::getInstance()
|
|
->make(\Illuminate\Support\Defer\DeferredCallbackCollection::class)
|
|
->invokeWhen(fn($callback) => true);
|
|
|
|
// Persist session changes to Redis before exit
|
|
$session->save();
|
|
} catch (Exception $e) {
|
|
// Send error via socket pair
|
|
$ipc->sendToParent(json_encode([
|
|
'event' => $message['event'] . ':error',
|
|
'data' => ['message' => $e->getMessage()],
|
|
]));
|
|
|
|
// Log DB connection failures specifically for monitoring
|
|
if ($this->isDbConnectionError($e)) {
|
|
Log::channel('websocket')->error('DB connection failure in child process', [
|
|
'error' => $e->getMessage(),
|
|
'event' => $message['event'] ?? 'unknown',
|
|
]);
|
|
}
|
|
|
|
if (app()->bound('sentry')) {
|
|
app('sentry')->captureException($e);
|
|
}
|
|
}
|
|
|
|
// Flush Sentry before the child exits so captured events are actually sent.
|
|
// Without this, events from report()/captureException() may be lost because
|
|
// the child calls exit(0) before the async transport can dispatch them.
|
|
if (app()->bound('sentry')) {
|
|
app('sentry')->flush();
|
|
}
|
|
|
|
$ipc->closeChild();
|
|
exit(0);
|
|
}
|
|
|
|
// === PARENT PROCESS ===
|
|
// Setup event-driven reading - NO POLLING!
|
|
$startTime = microtime(true);
|
|
|
|
$ipc->setupParent(
|
|
// onData callback - called INSTANTLY when child sends
|
|
function ($data) use ($connection, $message, $startTime) {
|
|
$this->handleChildData($connection, $message, $data);
|
|
|
|
// Log latency for debugging
|
|
$elapsed = (microtime(true) - $startTime) * 1000;
|
|
if ($elapsed > 10) {
|
|
Log::channel('websocket')->debug('IPC latency: ' . round($elapsed, 2) . 'ms');
|
|
}
|
|
},
|
|
// onClose callback - child process ended
|
|
function () {
|
|
// Cleanup zombie process
|
|
pcntl_waitpid(-1, $status, WNOHANG);
|
|
|
|
// Free up a child slot and process any queued messages
|
|
$this->activeChildCount = max(0, $this->activeChildCount - 1);
|
|
$this->processDeferredMessages();
|
|
}
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Process queued messages that were deferred due to concurrency limits.
|
|
* Called when a child process exits, freeing a slot.
|
|
*/
|
|
protected function processDeferredMessages(): void
|
|
{
|
|
while (!empty($this->deferredMessages) && $this->activeChildCount < $this->maxConcurrentChildren) {
|
|
[$connection, $channel, $message] = array_shift($this->deferredMessages);
|
|
|
|
// Verify the connection is still open before processing
|
|
if (!isset($connection->socketId) || !isset($connection->app)) {
|
|
continue;
|
|
}
|
|
|
|
$this->forkWithSocketPair($connection, $channel, $message);
|
|
}
|
|
|
|
if (!empty($this->deferredMessages)) {
|
|
Log::channel('websocket')->info('Deferred message queue: ' . count($this->deferredMessages) . ' remaining');
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Execute the controller with DB connection resilience.
|
|
* If the first attempt fails with a DB connection error (e.g., "Too many connections",
|
|
* "server has gone away"), waits briefly and retries once with a fresh connection.
|
|
*/
|
|
protected function executeControllerWithDbResilience(
|
|
$mock,
|
|
Channel $channel,
|
|
array $message,
|
|
ChannelManager $channelManager
|
|
): void {
|
|
try {
|
|
Controller::controll_message($mock, $channel, $message, $channelManager);
|
|
} catch (\Throwable $e) {
|
|
if (!$this->isDbConnectionError($e)) {
|
|
throw $e;
|
|
}
|
|
|
|
// DB connection error — wait briefly for connections to free up, then retry
|
|
Log::channel('websocket')->warning('DB connection error, retrying after 500ms', [
|
|
'error' => $e->getMessage(),
|
|
'event' => $message['event'] ?? 'unknown',
|
|
]);
|
|
|
|
usleep(500_000); // 500ms backoff
|
|
|
|
// Force a completely fresh DB connection
|
|
try {
|
|
DB::disconnect();
|
|
DB::reconnect();
|
|
} catch (\Throwable $reconnectError) {
|
|
// If reconnect also fails, throw the original error
|
|
Log::channel('websocket')->error('DB reconnect failed after retry', [
|
|
'error' => $reconnectError->getMessage(),
|
|
]);
|
|
throw $e;
|
|
}
|
|
|
|
// Retry the controller execution once
|
|
Controller::controll_message($mock, $channel, $message, $channelManager);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Check if an exception is a DB connection error (too many connections, gone away, etc.)
|
|
*/
|
|
protected function isDbConnectionError(\Throwable $e): bool
|
|
{
|
|
$message = $e->getMessage();
|
|
$previous = $e->getPrevious();
|
|
$fullMessage = $message . ($previous ? ' ' . $previous->getMessage() : '');
|
|
|
|
$dbErrorPatterns = [
|
|
'Too many connections',
|
|
'SQLSTATE[08004]',
|
|
'SQLSTATE[HY000] [1040]',
|
|
'server has gone away',
|
|
'SQLSTATE[HY000] [2006]',
|
|
'Lost connection to MySQL',
|
|
'SQLSTATE[HY000] [2002]',
|
|
'Connection refused',
|
|
'SQLSTATE[08S01]',
|
|
'no connection to the server',
|
|
];
|
|
|
|
foreach ($dbErrorPatterns as $pattern) {
|
|
if (stripos($fullMessage, $pattern) !== false) {
|
|
return true;
|
|
}
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Handle data received from child via socket pair
|
|
*/
|
|
protected function handleChildData(ConnectionInterface $connection, array $message, $data): void
|
|
{
|
|
if (!$data) {
|
|
return;
|
|
}
|
|
|
|
// Prefix-based routing: C: = connection data, B: = broadcast, W: = whisper, else regular response
|
|
// Avoids JSON decode overhead for regular responses (most common path)
|
|
if (str_starts_with($data, 'C:')) {
|
|
// Connection data operation from child process.
|
|
// Lets controllers set/clear/reset arbitrary properties on the
|
|
// parent's in-memory connection object via IPC.
|
|
$op = substr($data, 2);
|
|
|
|
if ($op === 'RESET') {
|
|
// Clear auth state
|
|
unset($connection->authLoaded);
|
|
$connection->user = null;
|
|
|
|
// Clear any custom connection data that was stored via C:SET
|
|
foreach (($connection->_connectionDataKeys ?? []) as $key => $_) {
|
|
unset($connection->$key);
|
|
}
|
|
$connection->_connectionDataKeys = [];
|
|
} elseif (str_starts_with($op, 'SET:')) {
|
|
// C:SET:key:json_value
|
|
$rest = substr($op, 4);
|
|
$pos = strpos($rest, ':');
|
|
if ($pos !== false) {
|
|
$key = substr($rest, 0, $pos);
|
|
$value = json_decode(substr($rest, $pos + 1));
|
|
$connection->$key = $value;
|
|
$connection->_connectionDataKeys ??= [];
|
|
$connection->_connectionDataKeys[$key] = true;
|
|
}
|
|
} elseif (str_starts_with($op, 'DEL:')) {
|
|
// C:DEL:key
|
|
$key = substr($op, 4);
|
|
unset($connection->$key);
|
|
if (isset($connection->_connectionDataKeys[$key])) {
|
|
unset($connection->_connectionDataKeys[$key]);
|
|
}
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
if (str_starts_with($data, 'B:')) {
|
|
$bm = json_decode(substr($data, 2), true);
|
|
$this->broadcast(
|
|
$connection->app->id,
|
|
$bm['data'] ?? null,
|
|
$bm['event'] ?? null,
|
|
$bm['channel'] ?? null,
|
|
$bm['including_self'] ?? false,
|
|
$connection
|
|
);
|
|
return;
|
|
}
|
|
|
|
if (str_starts_with($data, 'W:')) {
|
|
$bm = json_decode(substr($data, 2), true);
|
|
$this->whisper(
|
|
$connection->app->id,
|
|
$bm['data'] ?? null,
|
|
$bm['event'] ?? null,
|
|
$bm['socket_ids'] ?? [],
|
|
$bm['channel'] ?? null,
|
|
);
|
|
return;
|
|
}
|
|
|
|
// Regular response - send directly without JSON decode
|
|
$connection->send($data);
|
|
}
|
|
|
|
protected function handleMessageError(\Throwable $e): void
|
|
{
|
|
Log::channel('websocket')->error('onMessage unhandled error: ' . $e->getMessage(), [
|
|
'file' => $e->getFile(),
|
|
'line' => $e->getLine(),
|
|
'trace' => $e->getTraceAsString(),
|
|
]);
|
|
|
|
if (app()->bound('sentry')) {
|
|
app('sentry')->captureException($e);
|
|
}
|
|
}
|
|
|
|
protected function cleanupChannelConnections(ConnectionInterface $connection): void
|
|
{
|
|
$cacheUpdates = [];
|
|
$cacheDeletes = ['ws_socket_auth_' . $connection->socketId];
|
|
$socketId = $connection->socketId;
|
|
|
|
foreach ($this->channel_connections as $channel => $connections) {
|
|
if (!isset($connections[$socketId])) {
|
|
continue;
|
|
}
|
|
|
|
unset($this->channel_connections[$channel][$socketId]);
|
|
|
|
if (empty($this->channel_connections[$channel])) {
|
|
unset($this->channel_connections[$channel]);
|
|
$cacheDeletes[] = 'ws_channel_connections_' . $channel;
|
|
continue;
|
|
}
|
|
|
|
// Pre-compute array_keys once per channel
|
|
$cacheUpdates['ws_channel_connections_' . $channel] = array_keys($this->channel_connections[$channel]);
|
|
}
|
|
|
|
// Pre-compute active channels once
|
|
$activeChannels = array_keys($this->channel_connections);
|
|
$cacheUpdates['ws_active_channels'] = $activeChannels;
|
|
|
|
// Batch read authed_users - we'll update it in the same batch
|
|
$authed_users = cache()->get('ws_socket_authed_users') ?? [];
|
|
unset($authed_users[$socketId]);
|
|
$cacheUpdates['ws_socket_authed_users'] = $authed_users;
|
|
|
|
// Single batched write and delete operation - MASSIVE latency improvement
|
|
if (!empty($cacheUpdates)) {
|
|
cache()->setMultiple($cacheUpdates);
|
|
}
|
|
if (!empty($cacheDeletes)) {
|
|
cache()->deleteMultiple($cacheDeletes);
|
|
}
|
|
|
|
// Note: Removed redundant WebsocketService::clearUserAuthed() call
|
|
// as we already handle all cache operations above in a single batch
|
|
}
|
|
|
|
protected function finalizeConnectionClose(ConnectionInterface $connection): void
|
|
{
|
|
$this->channelManager
|
|
->unsubscribeFromAllChannels($connection)
|
|
->then(function (bool $unsubscribed) use ($connection): void {
|
|
if (!isset($connection->app)) {
|
|
return;
|
|
}
|
|
|
|
$this->channelManager->unsubscribeFromApp($connection->app->id);
|
|
ConnectionClosed::dispatch($connection->app->id, $connection->socketId);
|
|
cache()->forget('ws_connection_' . $connection->socketId);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Handle the websocket errors.
|
|
*
|
|
* @param WebSocketException $exception
|
|
*/
|
|
public function onError(ConnectionInterface $connection, Exception $exception): void
|
|
{
|
|
if ($exception instanceof ExceptionsWebSocketException) {
|
|
$connection->send(json_encode(
|
|
$exception->getPayload()
|
|
));
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Check if the connection can be made for the
|
|
* current server instance.
|
|
*/
|
|
protected function connectionCanBeMade(ConnectionInterface $connection): bool
|
|
{
|
|
return $this->channelManager->acceptsNewConnections();
|
|
}
|
|
|
|
/**
|
|
* Verify the app key validity.
|
|
*
|
|
* @return $this
|
|
*/
|
|
protected function verifyAppKey(ConnectionInterface $connection)
|
|
{
|
|
$query = QueryParameters::create($connection->httpRequest);
|
|
|
|
$appKey = $query->get('appKey');
|
|
|
|
if (! $app = App::findByKey($appKey)) {
|
|
throw new UnknownAppKey($appKey);
|
|
}
|
|
|
|
$app->then(function ($app) use ($connection) {
|
|
$connection->app = $app;
|
|
});
|
|
|
|
return $this;
|
|
}
|
|
|
|
/**
|
|
* Verify the origin.
|
|
*
|
|
* @return $this
|
|
*/
|
|
protected function verifyOrigin(ConnectionInterface $connection)
|
|
{
|
|
if (! $connection->app->allowedOrigins) {
|
|
return $this;
|
|
}
|
|
|
|
$header = (string) ($connection->httpRequest->getHeader('Origin')[0] ?? null);
|
|
|
|
$origin = parse_url($header, PHP_URL_HOST) ?: $header;
|
|
|
|
if (! $header || ! in_array($origin, $connection->app->allowedOrigins)) {
|
|
throw new OriginNotAllowed($connection->app->key);
|
|
}
|
|
|
|
return $this;
|
|
}
|
|
|
|
/**
|
|
* Limit the connections count by the app.
|
|
*
|
|
* @return $this
|
|
*/
|
|
protected function limitConcurrentConnections(ConnectionInterface $connection)
|
|
{
|
|
if (! is_null($capacity = $connection->app->capacity)) {
|
|
$this->channelManager
|
|
->getGlobalConnectionsCount($connection->app->id)
|
|
->then(function ($connectionsCount) use ($capacity, $connection): void {
|
|
if ($connectionsCount >= $capacity) {
|
|
$exception = new ConnectionsOverCapacity;
|
|
|
|
$payload = json_encode($exception->getPayload());
|
|
|
|
tap($connection)->send($payload)->close();
|
|
}
|
|
});
|
|
}
|
|
|
|
return $this;
|
|
}
|
|
|
|
/**
|
|
* Create a socket id.
|
|
*
|
|
* @return $this
|
|
*/
|
|
protected function generateSocketId(ConnectionInterface $connection)
|
|
{
|
|
$socketId = sprintf('%d.%d', random_int(1, 1000000000), random_int(1, 1000000000));
|
|
|
|
$connection->socketId = $socketId;
|
|
|
|
return $this;
|
|
}
|
|
|
|
/**
|
|
* Establish connection with the client.
|
|
*
|
|
* @return $this
|
|
*/
|
|
protected function establishConnection(ConnectionInterface $connection)
|
|
{
|
|
$connection->send(json_encode([
|
|
'event' => 'pusher.connection_established',
|
|
'data' => json_encode([
|
|
'socket_id' => $connection->socketId,
|
|
'activity_timeout' => 30,
|
|
]),
|
|
]));
|
|
|
|
return $this;
|
|
}
|
|
|
|
protected function get_connection_channel(&$connection, &$message): ?Channel
|
|
{
|
|
// Put channel on its place
|
|
if (! isset($message['channel']) && isset($message['data']['channel'])) {
|
|
$message['channel'] = $message['data']['channel'];
|
|
unset($message['data']['channel']);
|
|
}
|
|
|
|
$this->channelManager->findOrCreate(
|
|
$connection->app->id,
|
|
$message['channel']
|
|
);
|
|
|
|
return $this->channelManager->find(
|
|
$connection->app->id,
|
|
$message['channel']
|
|
);
|
|
}
|
|
|
|
protected function handleChannelSubscriptions($message, $connection): ?Channel
|
|
{
|
|
$channel = $this->get_connection_channel($connection, $message);
|
|
$channel_name = $channel?->getName();
|
|
|
|
if (!$channel_name || !$channel) {
|
|
return null;
|
|
}
|
|
|
|
$eventLower = strtolower($message['event']);
|
|
|
|
if ($eventLower === 'pusher.subscribe' || $eventLower === 'pusher:subscribe') {
|
|
$this->handleSubscription($channel, $channel_name, $connection, $message);
|
|
}
|
|
|
|
if (str_contains($message['event'], '.unsubscribe')) {
|
|
$this->handleUnsubscription($channel, $channel_name, $connection);
|
|
}
|
|
|
|
return $channel;
|
|
}
|
|
|
|
protected function handleSubscription(
|
|
Channel $channel,
|
|
string $channel_name,
|
|
ConnectionInterface $connection,
|
|
array $message
|
|
): void {
|
|
$socketId = $connection->socketId;
|
|
|
|
if (!isset($this->channel_connections[$channel_name])) {
|
|
$this->channel_connections[$channel_name] = [];
|
|
}
|
|
|
|
if (!isset($this->channel_connections[$channel_name][$socketId])) {
|
|
$this->channel_connections[$channel_name][$socketId] = true;
|
|
|
|
// Only update cache if connection was actually added (avoid redundant writes)
|
|
// Pre-compute array_keys once for both updates
|
|
$channelSockets = array_keys($this->channel_connections[$channel_name]);
|
|
$activeChannels = array_keys($this->channel_connections);
|
|
|
|
// Buffer these writes - they can be batched with other subscriptions
|
|
$this->bufferCacheWrite('ws_channel_connections_' . $channel_name, $channelSockets);
|
|
$this->bufferCacheWrite('ws_active_channels', $activeChannels);
|
|
}
|
|
|
|
if ($channel->hasConnection($connection)) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
$channel->subscribe($connection, (object) $message);
|
|
} catch (\Throwable $e) {
|
|
// Silently handle subscription errors
|
|
}
|
|
}
|
|
|
|
protected function handleUnsubscription(
|
|
Channel $channel,
|
|
string $channel_name,
|
|
ConnectionInterface $connection
|
|
): void {
|
|
$socketId = $connection->socketId;
|
|
|
|
if (isset($this->channel_connections[$channel_name][$socketId])) {
|
|
unset($this->channel_connections[$channel_name][$socketId]);
|
|
|
|
// Pre-compute active channels once
|
|
$activeChannels = array_keys($this->channel_connections);
|
|
|
|
if (empty($this->channel_connections[$channel_name])) {
|
|
unset($this->channel_connections[$channel_name]);
|
|
|
|
// Buffer delete and update - can be batched
|
|
$this->bufferCacheDelete('ws_channel_connections_' . $channel_name);
|
|
$this->bufferCacheWrite('ws_active_channels', $activeChannels);
|
|
} else {
|
|
// Pre-compute channel sockets once
|
|
$channelSockets = array_keys($this->channel_connections[$channel_name]);
|
|
|
|
// Buffer these writes
|
|
$this->bufferCacheWrite('ws_channel_connections_' . $channel_name, $channelSockets);
|
|
$this->bufferCacheWrite('ws_active_channels', $activeChannels);
|
|
}
|
|
}
|
|
|
|
$channel->unsubscribe($connection);
|
|
}
|
|
|
|
protected function setRequest($message, $connection)
|
|
{
|
|
foreach (request()->keys() as $key) {
|
|
request()->offsetUnset($key);
|
|
}
|
|
|
|
request()->merge($message['data'] ?? []);
|
|
}
|
|
|
|
protected function authenticateConnection(
|
|
ConnectionInterface $connection,
|
|
PrivateChannel|Channel|PresenceChannel|null $channel,
|
|
$message = []
|
|
) {
|
|
// Fast path: auth already resolved for this connection (skips cache read + DB query + cache write)
|
|
if (isset($connection->authLoaded)) {
|
|
if ($connection->user) {
|
|
Auth::login($connection->user);
|
|
}
|
|
$this->scheduleLogout();
|
|
return;
|
|
}
|
|
|
|
$this->loadCachedAuth($connection, $channel);
|
|
$this->ensureUserIsSet($connection, $channel);
|
|
$this->updateAuthState($connection);
|
|
$this->cacheAuthenticatedUser($connection);
|
|
$this->scheduleLogout();
|
|
|
|
$connection->authLoaded = true;
|
|
}
|
|
|
|
protected function loadCachedAuth(ConnectionInterface $connection, $channel): void
|
|
{
|
|
if (isset($connection->auth)) {
|
|
return;
|
|
}
|
|
|
|
if (!$connection->socketId) {
|
|
return;
|
|
}
|
|
|
|
$cached_auth = cache()->get('socket_' . $connection->socketId);
|
|
if (!$cached_auth || !isset($cached_auth['type'])) {
|
|
return;
|
|
}
|
|
|
|
$connection->user = $cached_auth['type']::find($cached_auth['id']);
|
|
|
|
if ($channel) {
|
|
$channel->saveConnection($connection);
|
|
}
|
|
}
|
|
|
|
protected function ensureUserIsSet(ConnectionInterface $connection, $channel): void
|
|
{
|
|
if (isset($connection->user) && $connection->user) {
|
|
return;
|
|
}
|
|
|
|
$connection->user = false;
|
|
if ($channel) {
|
|
$channel->saveConnection($connection);
|
|
}
|
|
}
|
|
|
|
protected function updateAuthState(ConnectionInterface $connection): void
|
|
{
|
|
$connection->user
|
|
? Auth::login($connection->user)
|
|
: Auth::logout();
|
|
}
|
|
|
|
protected function cacheAuthenticatedUser(ConnectionInterface $connection): void
|
|
{
|
|
if (!Auth::user()) {
|
|
return;
|
|
}
|
|
|
|
/** @var \App\Models\User */
|
|
$user = Auth::user();
|
|
|
|
$socketId = $connection->socketId;
|
|
|
|
// Batch all auth cache operations into a single read + single write
|
|
$authed_users = cache()->get('ws_socket_authed_users') ?? [];
|
|
$authed_users[$socketId] = $user->id;
|
|
|
|
// Single batched cache write - reduces 3 operations to 1
|
|
cache()->setMultiple([
|
|
'ws_socket_auth_' . $socketId => $user,
|
|
'ws_socket_authed_users' => $authed_users
|
|
]);
|
|
|
|
// Note: Removed redundant WebsocketService::setUserAuthed() call
|
|
// as we already handle all cache operations above in a single batch
|
|
}
|
|
|
|
protected function scheduleLogout(): void
|
|
{
|
|
$this->channelManager->loop->futureTick(function () {
|
|
Auth::logout();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Add cache operation to write buffer for batching
|
|
*/
|
|
protected function bufferCacheWrite(string $key, $value): void
|
|
{
|
|
$this->cacheWriteBuffer[$key] = $value;
|
|
$this->scheduleCacheFlush();
|
|
}
|
|
|
|
/**
|
|
* Add cache deletion to buffer for batching
|
|
*/
|
|
protected function bufferCacheDelete(string $key): void
|
|
{
|
|
$this->cacheDeleteBuffer[] = $key;
|
|
unset($this->cacheWriteBuffer[$key]); // Remove from write buffer if exists
|
|
$this->scheduleCacheFlush();
|
|
}
|
|
|
|
/**
|
|
* Schedule cache flush on next event loop tick
|
|
* Multiple rapid requests will be batched into single I/O operation
|
|
*/
|
|
protected function scheduleCacheFlush(): void
|
|
{
|
|
if ($this->cacheBufferScheduled) {
|
|
return;
|
|
}
|
|
|
|
$this->cacheBufferScheduled = true;
|
|
|
|
$this->channelManager->loop->futureTick(function () {
|
|
$this->flushCacheBuffer();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Flush cache buffer - performs all pending operations in single batch
|
|
* This is the key optimization: N operations -> 2 I/O calls (1 write, 1 delete)
|
|
*/
|
|
protected function flushCacheBuffer(): void
|
|
{
|
|
if (!empty($this->cacheWriteBuffer)) {
|
|
cache()->setMultiple($this->cacheWriteBuffer);
|
|
$this->cacheWriteBuffer = [];
|
|
}
|
|
|
|
if (!empty($this->cacheDeleteBuffer)) {
|
|
cache()->deleteMultiple(array_unique($this->cacheDeleteBuffer));
|
|
$this->cacheDeleteBuffer = [];
|
|
}
|
|
|
|
$this->cacheBufferScheduled = false;
|
|
}
|
|
|
|
/**
|
|
* Force immediate cache flush (use for critical operations)
|
|
*/
|
|
protected function flushCacheBufferImmediate(): void
|
|
{
|
|
$this->flushCacheBuffer();
|
|
$this->cacheBufferScheduled = false;
|
|
}
|
|
|
|
public function broadcast(
|
|
string $appId,
|
|
mixed $payload,
|
|
?string $event = null,
|
|
?string $channel = null,
|
|
bool $including_self = false,
|
|
$connection = null
|
|
): void {
|
|
$channel = $this->channelManager->findOrCreate($appId, $channel);
|
|
|
|
// Pre-encode once for all connections
|
|
$encoded = json_encode([
|
|
'event' => $event,
|
|
'data' => $payload,
|
|
'channel' => $channel->getName(),
|
|
]);
|
|
|
|
foreach ($channel->getConnections() as $channel_conection) {
|
|
if ($channel_conection->socketId !== $connection->socketId) {
|
|
$channel_conection->send($encoded);
|
|
}
|
|
}
|
|
|
|
if ($including_self) {
|
|
$connection->send($encoded);
|
|
}
|
|
}
|
|
|
|
public function whisper(
|
|
string $appId,
|
|
mixed $payload,
|
|
?string $event = null,
|
|
array $socketIds = [],
|
|
?string $channelName = null
|
|
): void {
|
|
$p = [
|
|
'event' => ($event ?? $event),
|
|
'data' => $payload,
|
|
'channel' => $channelName,
|
|
];
|
|
|
|
$socketIdLookup = array_flip($socketIds);
|
|
$encoded = json_encode($p);
|
|
$sentTo = [];
|
|
|
|
// Search ALL connections across ALL channels to find target socket IDs
|
|
// This is necessary because whisper targets specific sockets regardless of channel
|
|
$this->channelManager->getLocalConnections()->then(function ($connections) use ($socketIdLookup, $encoded, &$sentTo) {
|
|
foreach ($connections as $connection) {
|
|
// Skip if already sent to this socket (can appear in multiple channels)
|
|
if (isset($sentTo[$connection->socketId])) {
|
|
continue;
|
|
}
|
|
|
|
if (isset($socketIdLookup[$connection->socketId])) {
|
|
$connection->send($encoded);
|
|
$sentTo[$connection->socketId] = true;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
}
|