http handshake process
@param string $buffer
@param Connection $connection
@throws \Exception
@return int
488598
public function handshakeInput($buffer, $connection) 
    {
        $len = parent::input($buffer, $connection);
        
        if(!$len)  return 0;
        
        $req = new HttpReqeust();
        $req->loadRequest($buffer);
        
        if($req->headers->get('Connection') !== 'Upgrade'){
            throw new \Exception("Bad Connection", 412);
        }
        
        if($req->headers->get('Upgrade') !== 'websocket'){
            throw new \Exception("Bad Upgrade", 412);
        }
        
        $seckey = $req->headers->get('Sec-Websocket-Key');
        if(!$seckey ||  base64_encode(base64_decode($seckey)) !== $seckey){
            throw new \Exception("Bad Websocket-Key", 412);
        }
        
        return $len;
    }
476982
public function removeConnection(Connection $connection): void
    {
        $socket = $connection->getSocket();

        if ($socket->getResource()) {
            $index = $socket->getResourceId();
        } else {
            $index = array_search($connection, $this->connections);
        }

        if (!$index) {
            $this->logger->warning('Could not remove connection: not found');
        }

        unset($this->connections[$index]);
        unset($this->resources[$index]);

        $this->server->notify(
            Server::EVENT_SOCKET_DISCONNECT,
            [$connection->getSocket(), $connection]
        );
    }
299922
final public function run(Connection $connection)
    {
        $start = microtime(true);

        foreach (QueryBuilder::getBeforeRun() as $callback) {

            // function(\UniMapper\Query $query)
            $callback($this);
        }

        $result = $this->onExecute($connection);
        foreach (QueryBuilder::getAfterRun() as $callback) {

            // function(\UniMapper\Query $query, mixed $result, int $elapsed)
            $callback($this, $result, microtime(true) - $start);
        }

        return $result;
    }
964125
protected function initializeConnection($remote, $connection)
    {
        try {
            $context = $this->getContext($connection);
            $socket = $this->getSocket($remote, $context);
            $stream = $this->getStream($socket);
            $this->initializeStream($stream, $connection);
            $this->addActiveConnection($connection);
        } catch (\Exception $e) {
            $this->emitConnectionError($e, $connection);
        }

        $this->emit('connect.after.each', array($connection, $connection->getData('write')));
    }
56681
public function getConnection($connection, $server = null)
    {
        $connectionKey = $this->credentials->sanitizeConnection($connection, $server);
        $handle = $connectionKey->toHandle();

        $connections = $this->getConnections();
        if (!$connections->has($handle)) {
            throw new ConnectionException('Invalid connection: '.$handle);
        }

        return $connections[$handle];
    }
1307386
public static function sendHandshake($connection)
    {
        if (!empty($connection->handshakeStep)) {
            return;
        }
        // Get Host.
        $port = $connection->getRemotePort();
        $host = $port === 80 ? $connection->getRemoteHost() : $connection->getRemoteHost() . ':' . $port;
        // Handshake header.
        $connection->websocketSecKey = base64_encode(md5(mt_rand(), true));
        $user_header = isset($connection->headers) ? $connection->headers :
            (isset($connection->wsHttpHeader) ? $connection->wsHttpHeader : null);
        $user_header_str = '';
        if (!empty($user_header)) {
            if (is_array($user_header)){
                foreach($user_header as $k=>$v){
                    $user_header_str .= "$k: $v\r\n";
                }
            } else {
                $user_header_str .= $user_header;
            }
            $user_header_str = "\r\n".trim($user_header_str);
        }
        $header = 'GET ' . $connection->getRemoteURI() . " HTTP/1.1\r\n".
        (!preg_match("/\nHost:/i", $user_header_str) ? "Host: $host\r\n" : '').
        "Connection: Upgrade\r\n".
        "Upgrade: websocket\r\n".
        "Origin: ". (isset($connection->websocketOrigin) ? $connection->websocketOrigin : '*') ."\r\n".
        (isset($connection->WSClientProtocol)?"Sec-WebSocket-Protocol: ".$connection->WSClientProtocol."\r\n":'').
        "Sec-WebSocket-Version: 13\r\n".
        "Sec-WebSocket-Key: " . $connection->websocketSecKey . $user_header_str . "\r\n\r\n";
        $connection->send($header, true);
        $connection->handshakeStep               = 1;
        $connection->websocketCurrentFrameLength = 0;
        $connection->websocketDataBuffer         = '';
        $connection->tmpWebsocketData            = '';
    }
324037
public function onWebsocketConnect($connection, $http_buffer)
    {
        $this->sendToWorker(GatewayProtocol::CMD_ON_WEBSOCKET_CONNECT, $connection, array('get' => $_GET, 'server' => $_SERVER, 'cookie' => $_COOKIE));
    }
1063492
public function input($buffer, $connection)
    {
        if(empty($this->seckey)) {
            $this->seckey = $connection->listener->getHandler()->secKey;
        }
        
        if(!$connection->hasAttribute( $this->seckey )) {
            return $this->handshakeInput($buffer, $connection);
        }
        
        return $this->streamInput($buffer, $connection);
    }
1225650
public function getByConnection(ConnectionInterface $connection)
    {
        $hash = spl_object_hash($connection);

        foreach ($this->slaves as $slave) {
            if ($slave->getConnection() && $hash === spl_object_hash($slave->getConnection())) {
                return $slave;
            }
        }

        throw new \Exception("Slave connection not registered.");
    }
858751
protected function acceptConnection($server, $request, $httpRequest, $handler)
    {
        $id = $request->fd;

        $connection = $this->createWebSocketConnection($server, $id);

        $this->fireOpenEvent($handler, $connection, $httpRequest);

        $this->connections[$id] = compact('connection', 'handler');
    }
92209
public function listen($connection, $queue, $delay, $memory, $timeout = 60)
    {
        $process = $this->makeProcess($connection, $queue, $delay, $memory, $timeout);

        while(true) {
            $this->runProcess($process, $memory);
        }
    }
490395
public function onHandshakeRequest(
        Connection $connection,
        string $path,
        string $origin,
        string $key,
        array $extensions
    ): void {
        if (!$this->isAllowed($origin)) {
            $connection->close(Protocol::CLOSE_NORMAL, 'Not allowed origin during handshake request');
        }
    }
680306
public function initConnection(Connection $connection)
    {
        /** @var $pdo PDO */
        $pdo = $connection->getWrappedConnection();
        $pdo->setAttribute(PDO::ATTR_EMULATE_PREPARES, true);
        $params = $connection->getParams();
        if (!isset($params['charset'])) {
            throw new RuntimeException("No connection charset specified");
        }
        $pdo->exec('SET NAMES ' . $pdo->quote($params['charset']));
    }
186605
protected function loadSession(Connection $connection)
    {
        $sessionId = $connection->session = $this->getSessionIdFromCookie($connection);
        
        $session = clone $this->Session;
        $session->setId($sessionId);
        $session->start();
        
        $readonly = (new ReadOnly(config('session.cookie')))->initialize($session->all(), $sessionId);
        WsSession::swap($readonly);
        Session::flush();
        
        unset($session);
    }
760656
public function onMessage($connection, $data)
    {
        $uri  = parse_url($_SERVER['REQUEST_URI']);
        $path = isset($uri['path']) ? $uri['path'] : '/';

        $file = $this->root . $path;

        if (!is_file($file)) {
            $this->app->worker($connection, $data);
        } else {
            $this->sendFile($connection, $file);
        }
    }
1135723
protected function handleAuthentication(ConnectionInterface $connection, Payload $payload)
    {
        if (! $this->connectionManager->authenticate($connection, $payload->getData())) {
            $connection->emit(new Payload(ConnectionEvent::SOCKET_AUTH_FAILURE, 'Invalid access token.'));

            $this->eventDispatcher->dispatch(ConnectionEvent::SOCKET_AUTH_FAILURE, new ConnectionEvent($connection));

            $this->logger->notice(
                sprintf(
                    'Authentication error <info>#%s</info> (<comment>%s</comment>)',
                    $connection->getId(),
                    $connection->getRemoteAddress()
                )
            );

            return;
        }

        $response = new Payload(ConnectionEvent::SOCKET_AUTH_SUCCESS, $connection->getClient()->jsonSerialize());
        $connection->emit($response);

        $this->eventDispatcher->dispatch(ConnectionEvent::SOCKET_AUTH_SUCCESS, new ConnectionEvent($connection));

        $this->logger->notice(
            sprintf(
                'Authenticated <info>#%s</info> (<comment>%s</comment>)',
                $connection->getId(),
                $connection->getRemoteAddress()
            )
        );
    }
947192
public function log($connection, $queue, $payload, $exception)
    {
        $result = $this->failedJobs->insert(
            [
                'connection' => $connection,
                'queue' => $queue,
                'payload' => $payload,
                'exception' => (string) $exception,
                'failed_at' => new UTCDateTime(),
            ]
        );

        return (string) $result->getInsertedId();
    }