websocket握手时触发

@param $connection
@param $http_buffer
324037
public function onWebsocketConnect($connection, $http_buffer)
    {
        $this->sendToWorker(GatewayProtocol::CMD_ON_WEBSOCKET_CONNECT, $connection, array('get' => $_GET, 'server' => $_SERVER, 'cookie' => $_COOKIE));
    }
224085
def on_connect(self, connection):
        ""
        self._stream = connection._reader
        self._buffer = SocketBuffer(self._stream, self._read_size)
        if connection.decode_responses:
            self.encoding = connection.encoding
1325656
public function onSocketConnect($socket, Connection $connection): void
    {
        $this->checkConnections($connection);
        $this->checkConnectionsPerIp($connection);
    }
1301038
protected void jointMessageHandler(DirectBuffer buffer, int offset, int length, Header header) {
        byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        // deserialize message
        val message = VoidMessage.fromBytes(data);

        // we're checking if this is known connection or not, and add it if not
        if (!remoteConnections.containsKey(message.getOriginatorId()))
            addConnection(message.getOriginatorId());

        log.debug("Got [{}] message from [{}]", message.getClass().getSimpleName(), message.getOriginatorId());

        // we're just putting deserialized message into the buffer
        try {
            messageQueue.put(message);
        } catch (InterruptedException e) {
            // :(
            throw new RuntimeException(e);
        }
    }
361380
public function onHandShake($server, $serverSocket, $request, $response)
    {
        $secWebSocketAccept = base64_encode(
            sha1($request->header['sec-websocket-key'].'258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true)
        );

        $response->header('Upgrade', 'websocket');
        $response->header('Connection', 'Upgrade');
        $response->header('Sec-WebSocket-Accept', $secWebSocketAccept);
        $response->header('Sec-WebSocket-Version', '13');

        // WebSocket connection to 'ws://127.0.0.1:9502/'
        // failed: Error during WebSocket handshake:
        // Response must not include 'Sec-WebSocket-Protocol' header if not present in request: websocket
        if (isset($request->header['sec-websocket-protocol'])) {
            $response->header('Sec-WebSocket-Protocol', $request->header['sec-websocket-protocol']);
        }

        $response->status(101);
        $response->end();

        if (method_exists($this, 'onOpen')) {
            $this->onOpen($server, $serverSocket, $request);
        }
    }
941731
public static synchronized void websocket(String url, BiConsumer<WebSocketChannel, BufferedTextMessage> onMessage) {
        checkStarted();
        instance().endpoints.add(HandlerUtil.websocket(url, new WebsocketEndpoint() {
            @Override
            public void onConnect(WebSocketHttpExchange exchange, WebSocketChannel channel) {

            }

            @Override
            protected void onFullTextMessage(WebSocketChannel channel, BufferedTextMessage message) throws IOException {
                onMessage.accept(channel, message);
            }
        }));
    }
1159513
protected function writeToRemote($data, $connection)
    {
        $buffer = serialize($data);
        $buffer = pack('N', 4 + strlen($buffer)) . $buffer;
        $len = fwrite($connection, $buffer);
        if ($len !== strlen($buffer)) {
            throw new \Exception('writeToRemote fail');
        }
    }
614531
public function onMessage($connection, $data)
    {
        Base::getLog()->debug(__METHOD__ . ' receive http request', [
            'uri'  => $_SERVER['REQUEST_URI'],
            'ip'   => $connection->getRemoteIp(),
            'port' => $connection->getRemotePort(),
            'data' => $data,
        ]);

        // 请求的文件
        $urlInfo = parse_url($_SERVER['REQUEST_URI']);
        if ( ! $urlInfo)
        {
            Base::getHttp()->header('HTTP/1.1 400 Bad Request');
            Base::getLog()->warning(__METHOD__ . ' receive bad request', [
                'uri'  => $_SERVER['REQUEST_URI'],
                'ip'   => $connection->getRemoteIp(),
                'port' => $connection->getRemotePort(),
            ]);
            return $connection->close($this->error400);
        }

        $path = $urlInfo['path'];

        $pathInfo = pathinfo($path);
        $extension = isset($pathInfo['extension']) ? $pathInfo['extension'] : '';
        if ($extension === '')
        {
            $path = ($len = strlen($path)) && $path[$len - 1] === '/'
                ? $path . $this->indexFile
                : $path . '/' . $this->indexFile;
            $extension = 'php';
        }

        $serverName = Arr::get($_SERVER, 'SERVER_NAME');

        $rootDir = isset($this->serverRoot[$serverName])
            ? $this->serverRoot[$serverName]
            : current($this->serverRoot);

        $file = "$rootDir/$path";

        // 对应的php文件不存在，而且支持rewrite
        if ( ! is_file($file) && $this->rewrite)
        {
            $file = is_string($this->rewrite)
                ? $rootDir . '/' . $this->rewrite
                : $rootDir . '/' . $this->indexFile;
            $extension = 'php';
            $_SERVER['PATH_INFO'] = $_SERVER['REQUEST_URI'];
        }

        // 请求的文件存在
        if (is_file($file))
        {
            Base::getLog()->debug(__METHOD__ . ' request file existed', [
                'file'      => $file,
                'extension' => $extension,
            ]);

            // 判断是否是站点目录里的文件
            if (( ! ($requestRealPath = realpath($file)) || ! ($rootDirRealPath = realpath($rootDir))) || 0 !== strpos($requestRealPath, $rootDirRealPath))
            {
                Base::getHttp()->header('HTTP/1.1 400 Bad Request');
                Base::getLog()->warning(__METHOD__ . ' receive bad request', [
                    'uri'  => $_SERVER['REQUEST_URI'],
                    'ip'   => $connection->getRemoteIp(),
                    'port' => $connection->getRemotePort(),
                ]);
                return $connection->close('<h1>400 Bad Request</h1>');
            }

            $file = realpath($file);

            // 如果请求的是php文件
            // PHP文件需要include
            if ($extension === 'php')
            {
                Base::getLog()->debug(__METHOD__ . ' handle request', [
                    'uri'  => $_SERVER['REQUEST_URI'],
                    'ip'   => $connection->getRemoteIp(),
                    'port' => $connection->getRemotePort(),
                    'file' => $file,
                ]);

                Base::getLog()->debug(__METHOD__ . ' clean components - start');
                Base::cleanComponents();
                Base::getLog()->debug(__METHOD__ . ' clean components - end');

                $cwd = getcwd();
                chdir($rootDir);
                ini_set('display_errors', 'off');
                // 缓冲输出
                ob_start();
                // 载入php文件
                try
                {
                    // $_SERVER变量
                    $_SERVER['HOME'] = $_SERVER['DOCUMENT_ROOT'] = dirname($file);
                    $_SERVER['SCRIPT_FILENAME'] = $file;
                    Base::getLog()->debug(__METHOD__ . ' dispatch client info', [
                        'ip'   => $_SERVER['REMOTE_ADDR'],
                        'port' => $_SERVER['REMOTE_PORT'],
                    ]);

                    include $file;
                }
                catch (Exception $e)
                {
                    Base::getLog()->error($e->getMessage(), [
                        'code' => $e->getCode(),
                        'file' => $e->getFile(),
                        'line' => $e->getLine(),
                    ]);
                    // 如果不是exit
                    if ($e->getMessage() != 'jump_exit')
                    {
                        echo $e;
                    }
                }

                Patch::applyShutdownFunction();

                $content = ob_get_clean();
                ini_set('display_errors', 'on');
                $result = $connection->close($content);
                chdir($cwd);
                return $result;
            }
            // 请求的是静态资源文件
            else
            {
                $contentType = Mime::getMimeFromExtension($extension, self::$defaultMimeType);
                Base::getLog()->debug(__METHOD__ . ' get static file content type', [
                    'extension'   => $extension,
                    'contentType' => $contentType,
                ]);
                Base::getHttp()->header('Content-Type: ' . $contentType);

                // 获取文件信息
                $info = stat($file);

                $modifiedTime = $info ? date('D, d M Y H:i:s', Arr::get($info, 'mtime')) . ' GMT' : '';

                // 如果有$_SERVER['HTTP_IF_MODIFIED_SINCE']
                if ( ! empty($_SERVER['HTTP_IF_MODIFIED_SINCE']) && $info)
                {
                    // 文件没有更改则直接304
                    if ($modifiedTime === $_SERVER['HTTP_IF_MODIFIED_SINCE'])
                    {
                        Base::getLog()->debug(__METHOD__ . ' no modified, return 304');
                        // 304
                        Base::getHttp()->header('HTTP/1.1 304 Not Modified');
                        // 发送给客户端
                        return $connection->close('');
                    }
                }

                if ($modifiedTime)
                {
                    Base::getLog()->debug(__METHOD__ . ' set last modified time', [
                        'time' => $modifiedTime,
                    ]);
                    Base::getHttp()->header("Last-Modified: $modifiedTime");
                }
                // 发送给客户端
                return $connection->close(file_get_contents($file));
            }
        }
        else
        {
            Base::getLog()->warning(__METHOD__ . ' requested file not found', [
                'file' => $file,
            ]);

            // 404
            Base::getHttp()->header("HTTP/1.1 404 Not Found");
            return $connection->close($this->error404);
        }
    }
957879
public static function encode($content, TcpConnection $connection)
    {
        // Default http-code.
        if (!isset(HttpCache::$header['Http-Code'])) {
            $header = "HTTP/1.1 200 OK\r\n";
        } else {
            $header = HttpCache::$header['Http-Code'] . "\r\n";
            unset(HttpCache::$header['Http-Code']);
        }

        // Content-Type
        if (!isset(HttpCache::$header['Content-Type'])) {
            $header .= "Content-Type: text/html;charset=utf-8\r\n";
        }

        // other headers
        foreach (HttpCache::$header as $key => $item) {
            if ('Set-Cookie' === $key && is_array($item)) {
                foreach ($item as $it) {
                    $header .= $it . "\r\n";
                }
            } else {
                $header .= $item . "\r\n";
            }
        }
		if(HttpCache::$gzip && isset($connection->gzip) && $connection->gzip){
			$header .= "Content-Encoding: gzip\r\n";
			$content = gzencode($content,$connection->gzip);
		}
        // header
        $header .= "Server: workerman/" . Worker::VERSION . "\r\nContent-Length: " . strlen($content) . "\r\n\r\n";

        // save session
        self::sessionWriteClose();

        // the whole http package
        return $header . $content;
    }
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']));
    }
1366275
public static function encode($content, TcpConnection $connection)
    {
        $headers = HttpCache::$header;

        Base::getLog()->debug(__METHOD__ . ' encode header and send', [
            'headers' => $headers,
        ]);

        $sendHeader = [];

        // 没有http-code默认给个，content-type也是
        $hasStatus = false;
        $hasContentType = false;
        foreach ($headers as $header)
        {
            if (substr($header, 0, strlen('HTTP/')) == 'HTTP/')
            {
                $hasStatus = true;
                $sendHeader[] = $header;
                Base::getLog()->debug(__METHOD__ . ' custom http code', [
                    'header' => $header,
                ]);
            }
            elseif (strtolower(substr($header, 0, strlen('content-type:'))) == 'content-type:')
            {
                $hasContentType = true;
            }
        }
        if ( ! $hasStatus)
        {
            $sendHeader[] = "HTTP/1.1 200 OK";
            Base::getLog()->debug(__METHOD__ . ' default http code');
        }
        if ( ! $hasContentType)
        {
            Base::getLog()->debug(__METHOD__ . ' missing content type, set default value');
            $sendHeader[] = 'Content-Type: text/html;charset=utf-8';
        }

        // 合并
        $sendHeader = Arr::merge($sendHeader, $headers);

        // 一些额外的header信息
        if (Base::$expose)
        {
            $sendHeader[] = "Server: tourze/" . Base::VERSION;
        }
        $sendHeader[] = "Content-Length: " . strlen($content);

        // 记录要输出的header

        Base::getLog()->debug(__METHOD__ . ' final send header', $sendHeader);
        $sendHeader = implode("\r\n", $sendHeader);
        $sendHeader .= "\r\n\r\n";

        // save session
        self::sessionWriteClose();

        HttpCache::$header = [];
        // the whole http package
        return $sendHeader . $content;
    }
1178774
protected function sendResponse($response, $http_code = 200)
    {
        $http_code = min(max($http_code, 100), 505);
        http_response_code($http_code);
        header('Content-type: application/json');
        echo json_encode($response);
    }
1290549
public function pushed($connection, $queue, JobPayload $payload)
    {
        $this->connection()->pipeline(function ($pipe) use ($connection, $queue, $payload) {
            $this->storeJobReference($pipe, 'recent_jobs', $payload);

            $time = str_replace(',', '.', microtime(true));

            $pipe->hmset(
                $payload->id(), [
                    'id' => $payload->id(),
                    'connection' => $connection,
                    'queue' => $queue,
                    'name' => $payload->decoded['displayName'],
                    'status' => 'pending',
                    'payload' => $payload->value,
                    'created_at' => $time,
                    'updated_at' => $time,
                ]
            );

            $pipe->expireat(
                $payload->id(), Chronos::now()->addMinutes($this->recentJobExpires)->getTimestamp()
            );
        });
    }
1205000
public function onMessage($server, $frame) {
        $data = json_decode($frame->data, true);
        $method = $data[ 'method' ];
        unset($data[ 'method' ]);
        $user_id = $this->fidToUserId($frame->fd);
        /* @var $service WebSocketService */
        $service = Ioc::get($this->config[ 'service' ]);
        $service->$method($user_id, $data);
        CoContext::getContext()->release();
    }
190130
public function log( $connection, $queue, $payload )
	{
		$failed_at = Carbon::now();

		return $this->getTable()->insertGetId( compact( 'connection', 'queue', 'payload', 'failed_at' ) );
	}
703537
public function onHandShake($server, $serverSocket, $request, $response)
    {
        $swapHttpRequest = $this->app['request'];

        $httpRequest = $this->createHandShakeRequest($request);

        $respondent = $this->createRespondent($response);

        $httpRequest->attributes->add([
            'request'    => $request,
            'response'   => $response,
            'respondent' => $respondent
        ]);

        $httpResponse = $this->handleRequest($httpRequest, $respondent);

        if ($httpResponse instanceof HandShakeResponse && $httpResponse->isAccepted()) {
            $this->acceptConnection($server, $request, $httpRequest, $httpResponse->getHandler());
        }

        $this->sendResponse($respondent, $httpResponse);

        $this->app->instance('request', $swapHttpRequest);
    }
628133
public static function decode($buffer, ConnectionInterface $connection)
    {
        $masks = $data = $decoded = '';
        $len = ord($buffer[1]) & 127;
        if ($len === 126) {
            $masks = substr($buffer, 4, 4);
            $data  = substr($buffer, 8);
        } else {
            if ($len === 127) {
                $masks = substr($buffer, 10, 4);
                $data  = substr($buffer, 14);
            } else {
                $masks = substr($buffer, 2, 4);
                $data  = substr($buffer, 6);
            }
        }
        for ($index = 0; $index < strlen($data); $index++) {
            $decoded .= $data[$index] ^ $masks[$index % 4];
        }
        if ($connection->websocketCurrentFrameLength) {
            $connection->websocketDataBuffer .= $decoded;
            return $connection->websocketDataBuffer;
        } else {
            if ($connection->websocketDataBuffer !== '') {
                $decoded                         = $connection->websocketDataBuffer . $decoded;
                $connection->websocketDataBuffer = '';
            }
            return $decoded;
        }
    }