WebSocket Support for Real-time Communication

[Thavarshan]

Summary

Add WebSocket support to enable real-time bidirectional communication, extending Fetch PHP beyond traditional HTTP to include persistent connections.

Motivation

WebSocket support enables:

• Real-time Applications: Chat, notifications, live updates
• Bidirectional Communication: Two-way data flow
• Persistent Connections: Avoid connection overhead
• Event-driven Architecture: Real-time event handling
• Modern Web Standards: Support for WebSocket protocol
• Unified API: Consistent interface with HTTP client

Proposed API

// Basic WebSocket connection
$websocket = fetch_client()
    ->websocket('wss://api.example.com/ws')
    ->onOpen(function () {
        echo "Connected!\n";
    })
    ->onMessage(function (WebSocketMessage $message) {
        echo "Received: " . $message->getData() . "\n";
    })
    ->onClose(function ($code, $reason) {
        echo "Closed: $code - $reason\n";
    })
    ->onError(function (\Throwable $error) {
        echo "Error: " . $error->getMessage() . "\n";
    })
    ->connect();

// Send messages
$websocket->send('Hello, WebSocket!');
$websocket->sendJson(['type' => 'chat', 'message' => 'Hello']);

// Auto-reconnection
$websocket = fetch_client()
    ->websocket('wss://api.example.com/ws')
    ->withReconnection([
        'enabled' => true,
        'max_attempts' => 5,
        'delay' => 1000, // 1 second
        'backoff_multiplier' => 2,
    ])
    ->connect();

Implementation Details

interface WebSocketInterface
{
    public function connect(): PromiseInterface;
    public function send(string $data): void;
    public function sendJson(array $data): void;
    public function close(int $code = 1000, string $reason = ''): void;
    public function ping(string $data = ''): void;
    public function getReadyState(): WebSocketState;
}

enum WebSocketState: int
{
    case CONNECTING = 0;
    case OPEN = 1;
    case CLOSING = 2;
    case CLOSED = 3;
}

class WebSocketMessage
{
    public function __construct(
        private string $data,
        private WebSocketOpcode $opcode,
        private bool $fin = true
    ) {}
    
    public function getData(): string { return $this->data; }
    public function getOpcode(): WebSocketOpcode { return $this->opcode; }
    public function isText(): bool { return $this->opcode === WebSocketOpcode::TEXT; }
    public function isBinary(): bool { return $this->opcode === WebSocketOpcode::BINARY; }
    public function getJson(): array { return json_decode($this->data, true) ?? []; }
}

class WebSocketConnection implements WebSocketInterface
{
    private WebSocketState $state = WebSocketState::CLOSED;
    private array $eventListeners = [];
    private ?ReconnectionManager $reconnectionManager = null;
    
    public function __construct(
        private string $uri,
        private array $options = []
    ) {
        if ($this->options['reconnection']['enabled'] ?? false) {
            $this->reconnectionManager = new ReconnectionManager($this->options['reconnection']);
        }
    }
    
    public function connect(): PromiseInterface
    {
        return async(function () {
            $this->state = WebSocketState::CONNECTING;
            
            // Perform WebSocket handshake
            $handshake = $this->performHandshake();
            
            if ($handshake->successful()) {
                $this->state = WebSocketState::OPEN;
                $this->emit('open');
                $this->startMessageLoop();
            } else {
                throw new WebSocketException('Handshake failed');
            }
        });
    }
    
    private function performHandshake(): ResponseInterface
    {
        $key = base64_encode(random_bytes(16));
        
        return fetch($this->uri, [
            'method' => 'GET',
            'headers' => [
                'Upgrade' => 'websocket',
                'Connection' => 'Upgrade',
                'Sec-WebSocket-Key' => $key,
                'Sec-WebSocket-Version' => '13',
            ],
        ]);
    }
    
    public function send(string $data): void
    {
        if ($this->state !== WebSocketState::OPEN) {
            throw new WebSocketException('WebSocket is not open');
        }
        
        $frame = $this->createFrame(WebSocketOpcode::TEXT, $data);
        $this->writeFrame($frame);
    }
    
    private function startMessageLoop(): void
    {
        async(function () {
            while ($this->state === WebSocketState::OPEN) {
                try {
                    $frame = $this->readFrame();
                    $this->handleFrame($frame);
                } catch (\Throwable $e) {
                    $this->handleError($e);
                    break;
                }
            }
        });
    }
    
    private function handleFrame(WebSocketFrame $frame): void
    {
        match ($frame->getOpcode()) {
            WebSocketOpcode::TEXT, WebSocketOpcode::BINARY => 
                $this->emit('message', new WebSocketMessage($frame->getPayload(), $frame->getOpcode())),
            WebSocketOpcode::CLOSE => $this->handleClose($frame),
            WebSocketOpcode::PING => $this->handlePing($frame),
            WebSocketOpcode::PONG => $this->handlePong($frame),
            default => null,
        };
    }
}

Advanced Features

Auto-reconnection

class ReconnectionManager
{
    private int $attempts = 0;
    private int $maxAttempts;
    private int $baseDelay;
    private float $backoffMultiplier;
    
    public function shouldReconnect(): bool
    {
        return $this->attempts < $this->maxAttempts;
    }
    
    public function getDelay(): int
    {
        return (int) ($this->baseDelay * pow($this->backoffMultiplier, $this->attempts));
    }
    
    public function scheduleReconnection(WebSocketConnection $connection): PromiseInterface
    {
        $delay = $this->getDelay();
        $this->attempts++;
        
        return async(function () use ($connection, $delay) {
            await(sleep($delay / 1000));
            return $connection->connect();
        });
    }
}

Message Types and Protocols

// Chat protocol
$chat = fetch_client()
    ->websocket('wss://chat.example.com/ws')
    ->protocol('chat.v1')
    ->onMessage(function (WebSocketMessage $message) {
        $data = $message->getJson();
        
        match ($data['type']) {
            'chat' => $this->handleChatMessage($data),
            'join' => $this->handleUserJoin($data),
            'leave' => $this->handleUserLeave($data),
            default => null,
        };
    });

// Custom message handlers
$websocket->onMessageType('notification', function ($data) {
    $this->showNotification($data['title'], $data['body']);
});

Use Cases

Real-time Dashboard

$dashboard = fetch_client()
    ->websocket('wss://api.example.com/dashboard')
    ->onMessage(function (WebSocketMessage $message) {
        $update = $message->getJson();
        
        match ($update['type']) {
            'metric' => $this->updateMetric($update['data']),
            'alert' => $this->showAlert($update['data']),
            'user_activity' => $this->updateUserCount($update['data']),
        };
    })
    ->withHeartbeat(30) // Send ping every 30 seconds
    ->connect();

Live Chat

$chat = fetch_client()
    ->websocket('wss://chat.example.com/room/123')
    ->onMessage(function (WebSocketMessage $message) {
        $chatMessage = $message->getJson();
        $this->displayMessage($chatMessage['user'], $chatMessage['text']);
    })
    ->connect();

// Send chat message
$chat->sendJson([
    'type' => 'message',
    'text' => 'Hello everyone!',
    'user' => $this->getCurrentUser(),
]);

Benefits

• Real-time Capability: Enable live, interactive applications
• Efficiency: Persistent connections reduce overhead
• Modern Standards: Support for WebSocket protocol
• Unified API: Consistent with existing HTTP client interface
• Event-driven: Natural fit for reactive applications

Priority

Future Consideration - Valuable for real-time applications but requires significant implementation effort and may be outside core HTTP client scope.

[github-actions]

This issue has been automatically marked as stale due to inactivity.