WebSocketServer::looping() - Code Metrics - Inspection of "#3. Fix 5.5 -> 7.0 + refactor, shorten casts boole..." - arthurkushman/php-wss - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( 2d58e0...41ca8e )

by Arthur
created 2018-02-10 08:18 UTC
WebSocketServer::looping() F

↳ Parent: WebSocketServer
Complexity

Conditions	24
Paths	489
Size

Total Lines	102
Code Lines	56
Duplication

Lines	25
Ratio	24.51 %
Importance

Changes
Metric	Value
dl	25
loc	102
rs	3.0503
c	0
b	0
f	0
cc	24
eloc	56
nc	489
nop	1
How to fix Long Method Complexity

<?php

namespace WSSC;

use WSSC\Components\Connection;
use WSSC\Contracts\CommonsContract;
use WSSC\Contracts\WebSocketMessageContract;
use WSSC\Contracts\WebSocketServerContract;
use WSSC\Exceptions\WebSocketException;

/**
 * Create by Arthur Kushman
 */
class WebSocketServer implements WebSocketServerContract, CommonsContract
{

    private $clients = [];
    // set any template You need ex.: GET /subscription/messenger/token
    private $pathParams = [];
    private $config;
    private $handshakes = [];
    private $headersUpgrade = [];
    private $totalClients = 0;
    private $maxClients = 1;
    private $handler;
    private $connImpl;
    private $cureentConn;
    // for the very 1st time must be true
    private $stepRecursion = true;

    const MAX_BYTES_READ = 8192;
    const HEADER_BYTES_READ = 1024;
    // must be the time for interaction between each client
    const STREAM_SELECT_TIMEOUT = 3600;
    // stream non-blocking 
    const NON_BLOCK = 0;
    // max clients to fork another process
    const MAX_CLIENTS_REMAINDER_FORK = 1000;
    const PROC_TITLE = 'php-wss';

    /**
     * WebSocketServer constructor.
     * @param WebSocketMessageContract $handler
     * @param array $config
     */
    public function __construct(
        WebSocketMessageContract $handler,
        $config = [
            'host' => self::DEFAULT_HOST,
            'port' => self::DEFAULT_PORT,
        ]
    )
    {
        ini_set('default_socket_timeout', 5); // this should be >= 5 sec, otherwise there will be broken pipe - tested
        $this->handler = $handler;
        $this->config = $config;
        $this->connImpl = new Connection();
    }

    /**
     * Runs main process - Anscestor with server socket on TCP
     */
    public function run()
    {
        $errno = NULL;
        $errorMessage = '';

        $server = stream_socket_server("tcp://{$this->config['host']}:{$this->config['port']}", $errno, $errorMessage);
        if ($server === false) {
            die('Could not bind to socket: ' . $errno . ' - ' . $errorMessage . PHP_EOL);
        }
        @cli_set_process_title(self::PROC_TITLE);
// For example instead of
@mkdir($dir);

// Better use
if (@mkdir($dir) === false) {
    throw new \RuntimeException('The directory '.$dir.' could not be created.');
}
        $this->eventLoop($server);
    }

    /**
     * Recursive event loop that input intu recusion by remainder = 0 - thus when N users,
     * and when forks equals true which prevents it from infinite recursive iterations
     *
     * @param resource $server server connection
     * @param bool $fork flag to fork or run event loop
     */
    private function eventLoop($server, bool $fork = false)
    {
        if ($fork === true) {
            $pid = pcntl_fork();

            if ($pid) { // run eventLoop in parent        
                @cli_set_process_title(self::PROC_TITLE);
// For example instead of
@mkdir($dir);

// Better use
if (@mkdir($dir) === false) {
    throw new \RuntimeException('The directory '.$dir.' could not be created.');
}
                $this->eventLoop($server);
            }
        } else {
            $this->looping($server);
        }
    }

    /**
     * @param resource $server
     */
    private function looping($server)
    {
        while (true) {
            $this->totalClients = count($this->clients) + 1;

            // maxClients prevents process fork on count down
            if ($this->totalClients > $this->maxClients) {
                $this->maxClients = $this->totalClients;
            }

            if ($this->totalClients !== 0 // avoid 0 process creation
                && $this->totalClients % self::MAX_CLIENTS_REMAINDER_FORK === 0 // only when N is there
                && true === $this->stepRecursion // only once
                && $this->maxClients === $this->totalClients // only if stack grows
            ) {
                $this->stepRecursion = false;
                $this->eventLoop($server, true);
            }

            if ($this->totalClients !== 0 && $this->totalClients % self::MAX_CLIENTS_REMAINDER_FORK === 0 && $this->maxClients > $this->totalClients) { // there is less connection for amount of processes at this moment
                exit(1);
            }

            //prepare readable sockets
            $readSocks = $this->clients;
            $readSocks[] = $server;

            //start reading and use a large timeout
            if (!stream_select($readSocks, $write, $except, self::STREAM_SELECT_TIMEOUT)) {
                die('something went wrong while selecting');
            }

            //new client
            if (in_array($server, $readSocks)) {
                $newClient = stream_socket_accept($server, 0); // must be 0 to non-block
                if ($newClient) {
                    // print remote client information, ip and port number
                    //                        $socketName = stream_socket_get_name($newClient, true);

                    // important to read from headers here coz later client will change and there will be only msgs on pipe
                    $headers = fread($newClient, self::HEADER_BYTES_READ);
                    if (empty($this->handler->pathParams[0]) === false) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                        $this->setPathParams($headers);
                    }
                    $this->clients[] = $newClient;
                    $this->stepRecursion = true; // set on new client coz of remainder % is always 0
                    // trigger OPEN event
                    $this->handler->onOpen($this->connImpl->getConnection($newClient));
                    $this->handshake($newClient, $headers);
                }
                //delete the server socket from the read sockets
                unset($readSocks[array_search($server, $readSocks)]);
            }

            //message from existing client
            foreach ($readSocks as $kSock => $sock) {
                $data = $this->decode(fread($sock, self::MAX_BYTES_READ));
                $dataType = $data['type'];
                $dataPayload = $data['payload'];
                // to manipulate connection through send/close methods via handler, specified in IConnection
                $this->cureentConn = $this->connImpl->getConnection($sock);
                if (empty($data) || $dataType === self::EVENT_TYPE_CLOSE) { // close event triggered from client - browser tab or close socket event
                    // trigger CLOSE event
                    try {
                        $this->handler->onClose($this->cureentConn);
                    } catch (WebSocketException $e) {
                        $e->printStack();
                    }
                    // to avoid event leaks
                    unset($this->clients[array_search($sock, $this->clients)], $readSocks[$kSock]);
                    continue;
                }

                if ($dataType === self::EVENT_TYPE_TEXT) {

                    // trigger MESSAGE event
                    try {
                        echo 'trigger MESSAGE event';
                        $this->handler->onMessage($this->cureentConn, $dataPayload);
                    } catch (WebSocketException $e) {
                        $e->printStack();
                    }
                }

                if ($dataType === self::EVENT_TYPE_PING) {

                    // trigger PING event
                    try {
                        $this->handler->onPing($this->cureentConn, $dataPayload);
                    } catch (WebSocketException $e) {
                        $e->printStack();
                    }
                }

                if ($dataType === self::EVENT_TYPE_PONG) {

                    // trigger PONG event
                    try {
                        $this->handler->onPong($this->cureentConn, $dataPayload);
                    } catch (WebSocketException $e) {
                        $e->printStack();
                    }
                }
            }
        }
    }

    /**
     * Message frames decoder
     *
     * @param string $data
     * @return mixed null on empty data|false on improper data|array - on success
     */
    private function decode($data)
    {
        if (empty($data)) {
            return NULL; // close has been sent
        }

        $unmaskedPayload = '';
        $decodedData = [];

        // estimate frame type:
        $firstByteBinary = sprintf('%08b', ord($data[0]));
        $secondByteBinary = sprintf('%08b', ord($data[1]));
        $opcode = bindec(substr($firstByteBinary, 4, 4));
        $isMasked = $secondByteBinary[0] === '1';
        $payloadLength = ord($data[1]) & self::MASK_127;

        // unmasked frame is received:
        if (!$isMasked) {
            return ['type' => '', 'payload' => '', 'error' => self::ERR_PROTOCOL];
        }

        switch ($opcode) {
            // text frame:
            case self::DECODE_TEXT:
                $decodedData['type'] = self::EVENT_TYPE_TEXT;
                break;
            case self::DECODE_BINARY:
                $decodedData['type'] = self::EVENT_TYPE_BINARY;
                break;
            // connection close frame:
            case self::DECODE_CLOSE:
                $decodedData['type'] = self::EVENT_TYPE_CLOSE;
                break;
            // ping frame:
            case self::DECODE_PING:
                $decodedData['type'] = self::EVENT_TYPE_PING;
                break;
            // pong frame:
            case self::DECODE_PONG:
                $decodedData['type'] = self::EVENT_TYPE_PONG;
                break;
            default:
                return ['type' => '', 'payload' => '', 'error' => self::ERR_UNKNOWN_OPCODE];
        }

        if ($payloadLength === self::MASK_126) {
            $mask = substr($data, 4, 4);
            $payloadOffset = self::PAYLOAD_OFFSET_8;
            $dataLength = bindec(sprintf('%08b', ord($data[2])) . sprintf('%08b', ord($data[3]))) + $payloadOffset;
        } elseif ($payloadLength === self::MASK_127) {
            $mask = substr($data, 10, 4);
            $payloadOffset = self::PAYLOAD_OFFSET_14;
            $tmp = '';
            for ($i = 0; $i < 8; $i++) {
                $tmp .= sprintf('%08b', ord($data[$i + 2]));
            }
            $dataLength = bindec($tmp) + $payloadOffset;
            unset($tmp);
        } else {
            $mask = substr($data, 2, 4);
            $payloadOffset = self::PAYLOAD_OFFSET_6;
            $dataLength = $payloadLength + $payloadOffset;
        }

        /**
         * We have to check for large frames here. socket_recv cuts at 1024 bytes
         * so if websocket-frame is > 1024 bytes we have to wait until whole
         * data is transferd.
         */
        if (strlen($data) < $dataLength) {
            return false;
        }

        if ($isMasked) {
            for ($i = $payloadOffset; $i < $dataLength; $i++) {
                $j = $i - $payloadOffset;
                if (isset($data[$i])) {
                    $unmaskedPayload .= $data[$i] ^ $mask[$j % 4];
                }
            }
            $decodedData['payload'] = $unmaskedPayload;
        } else {
            $payloadOffset -= 4;
            $decodedData['payload'] = substr($data, $payloadOffset);
        }

        return $decodedData;
    }

    /**
     * Handshakes/upgrade and key parse
     *
     * @param resource $client Source client socket to write
     * @param string $headers Headers that client has been sent
     * @return string   socket handshake key (Sec-WebSocket-Key)| false on parse error
     */
    private function handshake($client, string $headers) : string
    {
        $match = [];
        $key = empty($this->handshakes[(int)$client]) ? 0 : $this->handshakes[(int)$client];
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}
        preg_match(self::SEC_WEBSOCKET_KEY_PTRN, $headers, $match);
        if (empty($match[1])) {
            return false;
        }

        $key = $match[1];
        $this->handshakes[(int)$client] = $key;

        // sending header according to WebSocket Protocol
        $secWebSocketAccept = base64_encode(sha1(trim($key) . self::HEADER_WEBSOCKET_ACCEPT_HASH, true));
        $this->setHeadersUpgrade($secWebSocketAccept);
        $upgradeHeaders = $this->getHeadersUpgrade();

        fwrite($client, $upgradeHeaders);
        return $key;
    }

    /**
     * Sets an array of headers needed to upgrade server/client connection
     *
     * @param string $secWebSocketAccept base64 encoded Sec-WebSocket-Accept header
     */
    private function setHeadersUpgrade($secWebSocketAccept)
    {
        $this->headersUpgrade = [
            self::HEADERS_UPGRADE_KEY              => self::HEADERS_UPGRADE_VALUE,
            self::HEADERS_CONNECTION_KEY           => self::HEADERS_CONNECTION_VALUE,
            self::HEADERS_SEC_WEBSOCKET_ACCEPT_KEY => ' ' . $secWebSocketAccept // the space before key is really important
        ];
    }

    /**
     * Retreives headers from an array of headers to upgrade server/client connection
     *
     * @return string   Headers to Upgrade communication connection
     */
    private function getHeadersUpgrade() : string
    {
        $handShakeHeaders = self::HEADER_HTTP1_1 . self::HEADERS_EOL;
        if (empty($this->headersUpgrade)) {
            die('Headers array is not set' . PHP_EOL);
        }
        foreach ($this->headersUpgrade as $key => $header) {
            $handShakeHeaders .= $key . ':' . $header . self::HEADERS_EOL;
            if ($key === self::HEADERS_SEC_WEBSOCKET_ACCEPT_KEY) { // add additional EOL fo Sec-WebSocket-Accept
                $handShakeHeaders .= self::HEADERS_EOL;
            }
        }
        return $handShakeHeaders;
    }

    /**
     * Parses parameters from GET on web-socket client connection before handshake
     *
     * @param string $headers
     */
    private function setPathParams(string $headers) : void
    {
        /** @var WebSocketMessageContract $handler */
        if (empty($this->handler->pathParams) === false) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
            $matches = [];
            preg_match('/GET\s(.*?)\s/', $headers, $matches);
            $left = $matches[1];
            foreach ($this->handler->pathParams as $k => $param) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                if (empty($this->handler->pathParams[$k + 1]) && strpos($left, '/', 1) === false) {
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                    // do not eat last char if there is no / at the end
                    $this->handler->pathParams[$param] = substr($left, strpos($left, '/') + 1);
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                } else {
                    // eat both slashes
                    $this->handler->pathParams[$param] = substr($left, strpos($left, '/') + 1, strpos($left, '/', 1) - 1);
interface SomeInterface { }
class SomeClass implements SomeInterface {
    public $a;
}

function someFunction(SomeInterface $object) {
    if ($object instanceof SomeClass) {
        $a = $object->a;
    }
}
                }
                // clear the declaration of parsed param
                unset($this->handler->pathParams[array_search($param, $this->handler->pathParams, false)]);
                $left = substr($left, strpos($left, '/', 1));
            }
        }
    }

}

Push — master ( 2d58e0...41ca8e )

WebSocketServer::looping() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

arthurkushman / php-wss

Push — master ( 2d58e0...41ca8e )

WebSocketServer::looping() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Available Fixes

Duplication Side-by-Side

Filter issues like
