Helix\Socket\WebSocket\WebSocketServer

Complexity

Total Complexity

17

Size/Duplication

Total Lines	141
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	accept()	0	7	1
A	broadcast()	0	7	3
A	broadcastBinary()	0	3	1
A	broadcastPing()	0	3	1
A	broadcastText()	0	3	1
A	close()	0	12	3
A	count()	0	3	1
A	getClients()	0	3	1
A	newClient()	0	3	1
A	onOutOfBand()	0	2	1
A	onReadable()	0	3	1
A	remove()	0	4	1

<?php

namespace Helix\Socket\WebSocket;

use Countable;
use Exception;
use Helix\Socket\ReactiveInterface;
use Helix\Socket\Reactor;
use Helix\Socket\StreamServer;

/**
 * A WebSocket server.
 *
 * https://tools.ietf.org/html/rfc6455
 */
class WebSocketServer extends StreamServer implements Countable, ReactiveInterface {

    /**
     * @var WebSocketClient[]
     */
    protected $clients = [];

    /**
     * @var Reactor
     */
    protected $reactor;

    /**
     * @param $resource
     * @param Reactor $reactor
     */
    public function __construct ($resource, Reactor $reactor) {
        parent::__construct($resource);
        $reactor->add($this);
        $this->reactor = $reactor;
    }

    /**
     * @return WebSocketClient
     */
    public function accept () {
        /** @var WebSocketClient $client */
        $client = parent::accept();
        $this->clients[$client->getId()] = $client;
        $this->reactor->add($client);
        return $client;
    }

    /**
     * Sends a payload to all clients in the OK state.
     *
     * @param int $opCode
     * @param string $payload
     */
    public function broadcast (int $opCode, string $payload) {
        foreach ($this->clients as $client) {
            if ($client->isOk()) {
                $client->getFrameHandler()->write($opCode, $payload);
            }
        }
    }

    /**
     * @param string $payload
     */
    public function broadcastBinary (string $payload) {
        $this->broadcast(Frame::OP_BINARY, $payload);
    }

    /**
     * Sends a ping to all clients in the OK state.
     *
     * @param string $payload
     */
    public function broadcastPing (string $payload = '') {
        $this->broadcast(Frame::OP_PING, $payload);
    }

    /**
     * Sends a message to all clients in the OK state.
     *
     * @param string $text
     */
    public function broadcastText (string $text) {
        $this->broadcast(Frame::OP_TEXT, $text);
    }

    /**
     * Closes and removes all clients.
     *
     * @param int $code
     * @param string $reason
     * @return StreamServer
     */
    public function close (int $code = Frame::CLOSE_INTERRUPT, $reason = '') {
        foreach ($this->clients as $client) {
            try {
                $client->close($code, $reason);
            }
            catch (Exception $e) {
                continue;
            }
        }
        $this->reactor->remove($this);
        return parent::close();
    }

    /**
     * The number of clients attached.
     *
     * @return int
     */
    public function count (): int {
        return count($this->clients);
    }

    /**
     * @return WebSocketClient[]
     */
    public function getClients () {
        return $this->clients;
    }

    /**
     * @param resource $resource
     * @return WebSocketClient
     */
    protected function newClient ($resource) {
        return new WebSocketClient($resource, $this);
    }

    /**
     * Servers never get OOB data.
     *
     * @inheritDoc
     */
    final public function onOutOfBand (): void {
    }

    /**
     * @inheritDoc
     */
    public function onReadable (): void {
        $this->accept();
    }

    /**
     * Removes the client from the server and reactor.
     *
     * @param WebSocketClient $client
     */
    public function remove (WebSocketClient $client) {
        unset($this->clients[$client->getId()]);
        $this->reactor->remove($client);
    }
}