@param swoole_server $server
@param $fd
@param $data
@param $from_id

@return int|mixed

@throws \Exception
@throws \FastD\Packet\Exceptions\PacketException
504692
public function doWork(swoole_server $server, $fd, $data, $from_id)
    {
        if ('quit' === $data) {
            $server->close($fd);

            return 0;
        }
        $data = Json::decode($data);
        $request = new ServerRequest($data['method'], $data['path']);
        if (isset($data['args'])) {
            if ('GET' === $request->getMethod()) {
                $request->withQueryParams($data['args']);
            } else {
                $request->withParsedBody($data['args']);
            }
        }
        $response = app()->handleRequest($request);
        if (null !== $response->getFileDescriptor()) {
            $fd = $response->getFileDescriptor();
        }
        if (false === $server->connection_info($fd)) {
            return -1;
        }
        $server->send($fd, (string) $response->getBody());
        app()->shutdown($request, $response);

        return 0;
    }
1325045
private function handleSubscription(ConnectionInterface $from, $data) {
    if (isset($data['query'])) {
      // Request is for a query.

      $this->handleSubscriptionQuery($from, $data);
    } elseif (isset($data['uid']) && is_string($data['uid'])) {
      // Request is for a UID.

      $this->handleSubscriptionUid($from, $data);
    }
  }
1272495
public function onSwooleTask($serv, $task_id, $from_id, $data)
    {
        $type = $data['type'] ?? '';
        $message = $data['message'] ?? '';
        switch ($type) {
            case SwooleMarco::MSG_TYPE_SEND_BATCH://发送消息
                foreach ($message['fd'] as $fd) {
                    $this->send($fd, $message['data'], true);
                }

                return null;
            case SwooleMarco::MSG_TYPE_SEND_ALL://发送广播
                foreach ($this->uid_fd_table as $row) {
                    $this->send($row['fd'], $message['data'], true);
                }

                return null;
            case SwooleMarco::MSG_TYPE_SEND_ALL_FD://发送广播
                foreach ($serv->connections as $fd) {
                    $this->send($fd, $message['data'], true);
                }

                return null;
            case SwooleMarco::SERVER_TYPE_TASK://task任务
                $task_name = $message['task_name'];
                $task = $this->loader->task($task_name, $this);
                $task_fuc_name = $message['task_fuc_name'];
                $task_data = $message['task_fuc_data'];
                $task_context = $message['task_context'];
                $call = [$task->getProxy(), $task_fuc_name];
                if (is_callable($call)) {
                    //给task做初始化操作
                    $task->initialization($task_id, $from_id, $this->server->worker_pid, $task_name, $task_fuc_name, $task_context);
                    $result = sd_call_user_func_array($call, $task_data);
                } else {
                    throw new SwooleException("method $task_fuc_name not exist in $task_name");
                }
                $task->destroy();

                return $result;
            default:
                return parent::onSwooleTask($serv, $task_id, $from_id, $data);
        }
    }
1022380
public function onTask($server, $taskId, $srcWorkerId, $data)
    {
        $this->container->make('events')->dispatch('swoole.task', func_get_args());

        try {
            // push websocket message
            if ($this->isWebsocketPushPayload($data)) {
                $this->pushMessage($server, $data['data']);
            // push async task to queue
            } elseif ($this->isAsyncTaskPayload($data)) {
                (new SwooleTaskJob($this->container, $server, $data, $taskId, $srcWorkerId))->fire();
            }
        } catch (Throwable $e) {
            $this->logServerError($e);
        }
    }
517247
public function onTask(Server $server, int $taskId, $fromId, $data)
    {
        $this->log('task worker received a new task', [
            'taskId'   => $taskId,
            'fromId'   => $fromId,
            'workerId' => $server->worker_id,
            'data'     => $data
        ], 'debug');

        // 返回任务执行的结果(finish操作是可选的，也可以不返回任何结果)
        return 'OK';
    }
700081
public function doPacket(swoole_server $server, $data, $clientInfo)
    {
        $data = Json::decode($data);
        $request = new ServerRequest($data['method'], $data['path']);
        if (isset($data['args'])) {
            if ('GET' === $request->getMethod()) {
                $request->withQueryParams($data['args']);
            } else {
                $request->withParsedBody($data['args']);
            }
        }
        $response = app()->handleRequest($request);
        $server->sendto($clientInfo['address'], $clientInfo['port'], (string) $response->getBody());
        app()->shutdown($request, $response);

        return 0;
    }
61044
public function write(int $id, int $offset, $data, int $flags = 0): int
    {
        $this->logger->beginProfile("write_{$id}");

        $storeValue = null;

        $stateFlags = 0;
        if (is_object($data)) {
            $storeValue = Serializer::serialize($data);
            $stateFlags |= self::F_SERIALIZED;
            $this->logger->debug(sprintf("Set F_SERIALIZED 0x%02x to process result value for id %d",
                self::F_SERIALIZED,
                $id));
        } else if (is_int($data)) {
            $storeValue = pack(self::P_INT, $data);
            $stateFlags |= self::F_IS_INT | self::F_PACKED;
            $this->logger->debug(sprintf("Set F_IS_INT | F_PACKED 0x%02x to process result value for id %d",
                self::F_IS_INT | self::F_PACKED, $id));
        } else if (is_float($data)) {
            $storeValue = pack(self::P_FLOAT, $data);
            $stateFlags |= self::F_IS_FLOAT | self::F_PACKED;
            $this->logger->debug(sprintf("Set F_IS_FLOAT | F_PACKED 0x%02x to process result value for id %d",
                self::F_IS_FLOAT | self::F_PACKED, $id));
        } else if (is_array($data)) {
            $storeValue = Serializer::serialize($data);
            $stateFlags |= self::F_IS_ARRAY | self::F_SERIALIZED;
            $this->logger->debug(sprintf("Set F_IS_ARRAY | F_SERIALIZED 0x%02x to process result value for id %d",
                self::F_IS_ARRAY | self::F_SERIALIZED, $id));
        } else if (is_bool($data)) {
            $stateFlags |= self::F_IS_BOOL;
            $this->logger->debug(sprintf("Set F_IS_BOOL 0x%02x to process result value for id %d", self::F_IS_BOOL,
                $id));
            $storeValue = $data;
        } else if (is_null($data)) {
            $stateFlags |= self::F_IS_NULL;
            $this->logger->debug(sprintf("Set F_IS_NULL 0x%02x to process result value for id %d", self::F_IS_NULL,
                $id));
            $storeValue = 0;
        } else if (is_string($data)) {
            $stateFlags |= self::F_IS_STRING;
            $this->logger->debug(sprintf("Set F_IS_STRING 0x%02x to process result value for id %d", self::F_IS_STRING,
                $id));
            $storeValue = $data;
        } else {
            throw new \InvalidArgumentException('Unsupported result type: ' . gettype($data));
        }

        $stateFlags |= $flags;

        $size = $this->sizeof($storeValue);
        $this->logger->debug(sprintf("Size of result value for id %d: 0x%08x", $id, $size));
        $header = $this->packHeader($id, $offset, $size, $stateFlags);

        if ($this->exists($id)) {
            $this->logger->debug("New shared block already exists for id {$id}. Deleting...");
            $this->delete($id);
        }

        try {
            $sh = shmop_open($this->getSharedItemKey($id), 'c', 0644, self::HEADER_SIZE + $size);

            if (!$sh) {
                $this->logger->error(sprintf("Cannot open shared memory block with size 0x%08x for id %d", $size, $id));

                return 0;
            }
        } catch (\Throwable $e) {
            $this->getLogger()->error($e);

            return 0;
        }


        $written = shmop_write($sh, $header, 0);
        $written += shmop_write($sh, $storeValue, self::HEADER_SIZE);
        $this->logger->debug(sprintf("Written memory bytes for id %d: 0x%08x", $id, $written));
        shmop_close($sh);
        $this->logger->endProfile("write_{$id}", "Writing {$id} data");

        return $written;
    }
810846
public function onClose(\Swoole\WebSocket\Server $server, int $fd)
    {
        // echo "[websocket][onclose]client fd{$fd} closed\n";

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

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

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

        // Clear websocket cache
        unset(self::$websocketFrameContext[$fd]);

        call_user_func([new $controller_class(), 'onClose'], $server, $fd);
    }
820268
public function processSpringbotAbandonedCart($from, $data)
		{
			$form_data = array(
				'from' => $from,
				'data' => $data
			);
			$this->call('springbot_process_abandoned_cart', $form_data);

			return TRUE;
		}
733183
public function onClose(\swoole_websocket_server $server, $fd) {
        $user_id = $this->fidToUserId($fd);
        $fid = $this->userIdToFid($user_id);
        Cache::redis()->hDel('user_id_#_fid', $user_id);
        Cache::redis()->hDel('fid_#_user_id', $fid);
        Cache::release();
        CoContext::getContext()->release();
    }
1220145
@Override
    public UniversalIdIntQueueMessage fromBytes(byte[] data) {
        try {
            UniversalIdIntQueueMessage other = BaseUniversalQueueMessage.fromBytes(data,
                    UniversalIdIntQueueMessage.class);
            fromMap(other.toMap());
            return this;
        } catch (InstantiationException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }
981679
public function _send($client_id, $data)
    {
        $length = \strlen($data);
        for ($written = 0; $written < $length; $written += $fwrite) {
            $fwrite = \stream_socket_sendto($client_id, substr($data, $written));
            if ($fwrite <= 0 or $fwrite === false) {
                return $written;
            }
        }
        return $written;
    }
471243
private function handlePublishUid(ConnectionInterface $from, $data) {
    $this->logger->notice(
        "Received a UID publish! (".
          (
            isset($data['name'])
                ? $data['name']
                : "{$data['oldName']} => {$data['newName']}"
          ).
          ", {$data['event']}, {$from->resourceId})"
    );

    foreach ($data as $key => $name) {
      if (!in_array($key, ['name', 'newName', 'oldName'])
          || !key_exists($name, $this->uidSubs)) {
        continue;
      }
      foreach ($this->uidSubs[$name] as $curClient) {
        $this->logger->notice(
            "Notifying client of {$data['event']}! ".
              "($name, {$curClient->resourceId})"
        );
        $payload = [
          'uid' => $name,
          'event' => $data['event']
        ];
        if (isset($data['value'])) {
          $payload['value'] = $data['value'];
        }
        $curClient->send(json_encode($payload));
      }
    }
  }
925585
public function messageBroadcast($fromUserId,$objectName,$content) {
        try{
            if(empty($fromUserId))
                throw new Exception('发送人用户 Id 不能为空');
            if(empty($objectName))
                throw new Exception('消息类型不能为空');
            if(empty($content))
                throw new Exception('发送消息内容不能为空');
            $ret = $this->curl(
                '/message/broadcast',
                array(
                    'fromUserId' => $fromUserId,
                    'objectName' => $objectName,
                    'content' => $content
                )
            );
            if(empty($ret))
                throw new Exception('请求失败');
            return $ret;
        }catch (Exception $e) {
            print_r($e->getMessage());
        }
    }
1067909
public function getJsonRpcResponseFromException(
        \Exception $exception,
        JsonRpcRequest $fromRequest = null
    ) : JsonRpcResponse {
        $event = new ActionEvent\OnExceptionEvent($exception, $fromRequest);
        $this->dispatchJsonRpcEvent($event::EVENT_NAME, $event);

        return $this->responseCreator->createErrorResponse($event->getException(), $fromRequest);
    }
1133004
public function serverGetByUid($uid)
  {
    foreach($this->serverList() as $server)
    {
      if($server["virtualserver_unique_identifier"] == $uid) return $server;
    }

    throw new Exception("invalid serverID", 0x400);
  }
864104
public synchronized void send(String data, String event, String id, EventCallback callback) {
        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, data, event, id, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(event, data, id, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }