WebSocket on message

@param Swoole\WebSocket\Server $server
@param [type] $frame
@return void
551740
public function onMessage(\Swoole\WebSocket\Server $server, \Swoole\WebSocket\Frame $frame)
    {
        // echo "[websocket][onmessage]receive from {$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}\n";
        // $server->push($frame->fd, "this is server");
        // print_r(self::$websocketFrameContext);

        if ( empty(self::$websocketFrameContext[$frame->fd]) ) {
            if ( getenv('STDOUT_LOG') ) {
                echo date('Y-m-d H:i:s') . " |\033[31m [ERROR][onMessage] WebSocket has been stoped before frame sending data\033[0m \n";
            }
            return;
        }

        // Save websocket connection Context
        $context = self::$websocketFrameContext[$frame->fd];

        // Restore global data
        $_POST   = $_SERVER = [];
        $_GET    = $_REQUEST = $context['get'];
        $_COOKIE = $context['cookie'];
        $controller_class = $context['controller_class'];

        call_user_func([new $controller_class(), 'onMessage'], $server, $frame);
    }
324037
public function onWebsocketConnect($connection, $http_buffer)
    {
        $this->sendToWorker(GatewayProtocol::CMD_ON_WEBSOCKET_CONNECT, $connection, array('get' => $_GET, 'server' => $_SERVER, 'cookie' => $_COOKIE));
    }
1018372
private void callOnFrameError(WebSocketException cause, WebSocketFrame frame)
    {
        mWebSocket.getListenerManager().callOnFrameError(cause, frame);
    }
1168181
void onReadingThreadFinished(WebSocketFrame closeFrame)
    {
        synchronized (mThreadsLock)
        {
            mReadingThreadFinished = true;
            mServerCloseFrame = closeFrame;

            if (mWritingThreadFinished == false)
            {
                // Wait for the writing thread to finish.
                return;
            }
        }

        // Both the reading thread and the writing thread have finished.
        onThreadsFinished();
    }
1212284
public function execute($server, $frame)
    {
        $skip = false;

        foreach ($this->strategies as $strategy) {
            $result = App::invokeMethod(
                [$strategy, 'handle'],
                [
                    'server' => $server,
                    'frame'  => $frame,
                ]
            );
            if ($result === true) {
                $skip = true;
                break;
            }
        }

        return $skip;
    }
15662
public function message(Server $server, Frame $frame)
    {
        $fd = $frame->fd;

        try {
            if (!$path = WebSocketContext::getMeta('path', $fd)) {
                throw new ContextLostException("The connection info has lost of the fd#$fd, on message");
            }

            $className = $this->getHandler($path)[0];

            /** @var HandlerInterface $handler */
            $handler = \bean($className);
            $handler->onMessage($server, $frame);
        } catch (\Throwable $e) {
            /** @see \Swoft\Event\EventManager::hasListenerQueue() */
            if (App::hasBean('eventManager') && \bean('eventManager')->hasListenerQueue(WsEvent::ON_ERROR)) {
                App::trigger(WsEvent::ON_ERROR, $frame, $e);
            } else {
                App::error($e->getMessage(), ['fd' => $fd, 'data' => $frame->data]);
                // close connection
                $server->close($fd);
            }
        }
    }
4905
public function pushMessage($server, array $data)
    {
        $pusher = Pusher::make($data, $server);
        $pusher->push($this->payloadParser->encode(
            $pusher->getEvent(),
            $pusher->getMessage()
        ));
    }
468734
protected function sendFrame($frame)
    {
        $c = &$this->attrs->get['c'];
        if (!is_string($c)) {
            $this->header('400 Bad Request');
            $this->finish();
            return;
        }
        $this->outputFrame($c . '(' . json_encode($frame, JSON_UNESCAPED_SLASHES) . ");\r\n");
        parent::sendFrame($frame);
    }
733602
protected function fireMessageEvent($handler, $connection, $frame)
    {
        $isBinary = $frame->opcode == WEBSOCKET_OPCODE_BINARY;

        $this->fireWebSocketHandler($handler, 'onMessage', $connection, $isBinary, $frame->data);
    }
168679
def on_send(self, frame):
        
        print('on_send %s %s %s' % (frame.cmd, frame.headers, frame.body))
769235
private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof CloseWebSocketFrame) {
            this.handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            CompletableFuture.completedFuture(new WebSocketContext(this.session,this.handler))
                    .thenAcceptAsync(this.handler::onDisConnect);
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException("unsupported frame type: " + frame.getClass().getName());
        }
        CompletableFuture.completedFuture(new WebSocketContext(this.session,this.handler,((TextWebSocketFrame) frame).text()))
                .thenAcceptAsync(this.handler::onText,ctx.executor());
    }
72061
public function onWorkerStart($server)
    {
        if ($this->container->config->get('swoole.enable_coroutine', false)) {
            Runtime::enableCoroutine(true);
        }

        $this->clearCache();

        $this->container->event->trigger('swoole.workerStart', func_get_args());

        // don't init app in task workers
        if ($server->taskworker) {
            $this->setProcessName('task process');

            return;
        }

        $this->setProcessName('worker process');

        $this->prepareApplication();

        if ($this->isServerWebsocket) {
            $this->prepareWebsocketHandler();
            $this->loadWebsocketRoutes();
        }
    }
499649
private function enable(Frame $frame)
    {
        $this->onHeartbeatFrame($frame, $this->getHeartbeats($frame));
        if ($this->intervalServer && $this->intervalClient) {
            $intervalAgreement = $this->calculateInterval(max($this->intervalClient, $this->intervalServer));
            $this->intervalUsed = $intervalAgreement / 1000; // milli to micro
            if ($intervalAgreement) {
                $this->enabled = true;
                $this->rememberActivity();
            }
        }
    }
499497
public function close(Server $server, int $fd)
    {
        try {
            if (!$path = WebSocketContext::getMeta('path', $fd)) {
                throw new ContextLostException(
                    "The connection info has lost of the fd#$fd, on connection closed"
                );
            }

            $className = $this->getHandler($path)[0];

            /** @var HandlerInterface $handler */
            $handler = \bean($className);

            if (\method_exists($handler, 'onClose')) {
                $handler->onClose($server, $fd);
            }
        } catch (\Throwable $e) {
            App::error($e->getMessage(), ['fd' => $fd]);
        }
    }
1309509
public function onFrame($data, $type)
    {
        $ws = $this;
        $req = json_decode($data, true);
        if ($req['type'] === 'subscribe') {
            $eventName = $req['event'];
            $this->appInstance->pubsub->sub($req['event'], $this, function ($data) use ($ws, $eventName) {
                $ws->sendObject([
                    'type' => 'event',
                    'event' => $eventName,
                    'data' => $data,
                ]);
            });
        }
    }
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);
    }
1064768
public function receiveFrame(Frame $frame)
    {
        if ('ERROR' === $frame->command) {
            throw new ServerErrorException($frame);
        }

        if ($this->state->isConnecting()) {
            if ('CONNECTED' !== $frame->command) {
                throw new InvalidFrameException(sprintf("Received frame with command '%s', expected 'CONNECTED'.", $frame->command));
            }

            $this->state->doneConnecting(
                $frame->getHeader('session'),
                $frame->getHeader('server')
            );

            return new ConnectionEstablishedCommand();
        }

        if ('CONNECTED' === $frame->command) {
            throw new InvalidFrameException(sprintf("Received 'CONNECTED' frame outside a connecting window."));
        }

        if ($this->state->isDisconnecting()) {
            if ('RECEIPT' === $frame->command && $this->state->isDisconnectionReceipt($frame->getHeader('receipt-id'))) {
                $this->state->doneDisconnecting();

                return new CloseCommand();
            }
        }

        return new NullCommand();
    }