PipeMessage

@param \swoole_server $serv server实例
@param int $fromWorkerId worker id
@param string $message 消息内容
777073
public function onPipeMessage($serv, $fromWorkerId, $message)
    {
        parent::onPipeMessage($serv, $fromWorkerId, $message);
        $data = unserialize($message);
        switch ($data['type']) {
            case Macro::MSG_TYPR_ASYN:
                $this->asynPoolManager->distribute($data['message']);
                break;
        }
    }
115832
public function onReceive(SwServer $server, int $fd, int $fromId, string $data)
    {
        // {{ log
        $this->get('logger')->info('接收数据', [
            'type' => 'TCP',
            'client' => $server->getClientInfo($fd),
            'data' => $data
        ]);
        // }}

        $this->callProtocolMethod('onReceive', $server, $fd, $fromId, $data);
    }
133603
private void replaceHeaders(final Message from, final Message to) {
        to.getHeaders().clear();
        to.getHeaders().putAll(from.getHeaders());
    }
701361
public function log($level, $message, $context = [], $fromErrorLogger = false)
    {
        // 将较高级别的日志,引到error.logger服务
        if ($fromErrorLogger
            || (isset($this->levels[$level]) && $this->levels[$level] < $this->levels[$this->proxyLevel])
        ) {
            return parent::log($level, $message, $context);
        } else {
            return $this->errorLogger->log($level, $message, $context, true);
        }
    }
867220
function (message) {
  var newMessage = "";
  if (requestNamespace) {
    if (requestNamespace.get('reqId')) {
      newMessage += "[" + requestNamespace.get('reqId') + "]";
    }
    if (requestNamespace.get('workerId')) {
      newMessage += "[Worker - " + requestNamespace.get('workerId') + "]";
    }
    const hrTime = process.hrtime();
    newMessage += "[" + timeInMilli(hrTime) + "]";
  }
  newMessage += message;
  return newMessage;
}
718060
public static Builder from(ServiceMessage message) {
    return ServiceMessage.builder().data(message.data()).headers(message.headers());
  }
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;
    }
607890
public function onTask($serv, $taskId, $fromId, $data)
    {
        if (is_string($data)) {
            $unserializeData = unserialize($data);
        } else {
            $unserializeData = $data;
        }
        $type    = $unserializeData['type'] ?? '';
        $message = $unserializeData['message'] ?? '';
        $result  = false;
        $status  = 200;
        switch ($type) {
            case Macro::SERVER_TYPE_TASK:
                try {
                    $taskName      = $message['task_name'];
                    $taskFucName   = $message['task_fuc_name'];
                    $taskData      = $message['task_fuc_data'];
                    /**
                     * @var Context $taskContext
                     */
                    $taskContext   = $message['task_context'];
                    $taskConstruct = $message['task_construct'];

                    // 构造请求日志对象
                    $PGLog                            = clone getInstance()->log;
                    $PGLog->logId                     = $taskContext->getLogId();
                    $PGLog->accessRecord['beginTime'] = microtime(true);
                    $PGLog->accessRecord['uri']       = $taskContext->getInput()->getPathInfo();
                    $PGLog->pushLog('task', $taskName);
                    $PGLog->pushLog('method', $taskFucName);
                    defined('SYSTEM_NAME') && $PGLog->channel = SYSTEM_NAME . '-task';
                    $PGLog->init();
                    $taskContext->setLogId($PGLog->logId);
                    $taskContext->setLog($PGLog);

                    if (empty($taskName) || empty($taskFucName)) {
                        $status = 500;
                        $PGLog->error('Task Not Found');
                        $PGLog->pushLog('status', $status);
                        $PGLog->appendNoticeLog();
                        return $result;
                    }

                    $objectPool    = AOPFactory::getObjectPool($this->objectPool, $this);
                    /**
                     * @var Task $task
                     */
                    $task          = $objectPool->get($taskName, $taskConstruct);

                    // 运行方法
                    if (method_exists($task, $taskFucName)) {
                        //给task做初始化操作
                        $task->__initialization($taskId, $this->server->worker_pid, $taskName, $taskFucName, $taskContext, $objectPool);
                        $result = $task->$taskFucName(...$taskData);
                    } else {
                        throw new Exception("method $taskFucName not exist in $taskName");
                    }
                } catch (\Throwable $e) {
                    $status = 500;
                    $PGLog->error(dump($e, false, true));
                } finally {
                    $PGLog->pushLog('status', $status);

                    if ($status === 200) {
                        (mt_rand(0, 99) < $this->taskLogRate) && $PGLog->appendNoticeLog();
                    } else {
                        $PGLog->appendNoticeLog();
                    }

                    //销毁对象
                    foreach ($this->objectPoolBuckets as $k => $obj) {
                        $objectPool->push($obj);
                        $this->objectPoolBuckets[$k] = null;
                        unset($this->objectPoolBuckets[$k]);
                    }
                }
                return $result;
            default:
                return parent::onTask($serv, $taskId, $fromId, $data);
        }
    }
273970
public function sendToAll($data, $fromDispatch = false)
    {
        $send_data = $this->packServerMessageBody(SwooleMarco::MSG_TYPE_SEND_ALL, ['data' => $data]);
        if ($this->isTaskWorker()) {
            $this->onSwooleTask($this->server, 0, 0, $send_data);
        } else {
            if ($this->task_num > 0) {
                $this->server->task($send_data);
            } else {
                foreach ($this->uid_fd_table as $row) {
                    $this->send($row['fd'], $data, true);
                }
            }
        }
        if ($fromDispatch) {
            return;
        }
        if ($this->isCluster()) {
            ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_sendToAll($data);
        }
    }
27257
public boolean sendInternalMessage(String worker, final String messageId, String eventName,
            final String eventMessage, boolean checkAvailableThreads) throws JMSException {
        // Spring Injection
        if (mdwMessageProducer != null) {
            mdwMessageProducer.sendMessage(eventMessage, messageId);

        }
        else {
            InternalEventDriver command = new InternalEventDriver(messageId, eventMessage);
            ThreadPoolProvider thread_pool = ApplicationContext.getThreadPoolProvider();
            if (checkAvailableThreads && !thread_pool.execute(worker, eventName, command)) {
                String msg = worker + " has no thread available for event: " + eventName
                        + " message:\n" + eventMessage;
                // make this stand out
                logger.warnException(msg, new Exception(msg));
                logger.info(thread_pool.currentStatus());
                return false;
            }
            else if (!checkAvailableThreads)
                thread_pool.execute(worker, eventName, command);

        }
        return true;
    }
1116538
public function onPipeMessage(Server $server, int $srcWorkerId, string $data): void
    {
        $this->log("worker #{$server->worker_id} received message from #$srcWorkerId, data: $data");
    }
255468
public static function onWorkerStart($serv, $worker_id) {
		Yesf::app()->initInWorker();
		//根据类型，设置不同的进程名
		if ($serv->taskworker) {
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' task ' . $worker_id);
		} else {
			self::initHotReload($serv);
			self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' worker ' . $worker_id);
		}
		//清除opcache
		if (function_exists('opcache_reset')) {
			opcache_reset();
		}
		//标记一下
		Swoole::$isTaskWorker = $serv->taskworker;
		//回调
		Internal::onWorkerStart();
		Plugin::trigger('workerStart', [$serv->taskworker, $worker_id]);
	}
151056
public function onSwooleWorkerError($serv, $worker_id, $worker_pid, $exit_code)
    {
        $data = ['worker_id' => $worker_id,
            'worker_pid' => $worker_pid,
            'exit_code' => $exit_code];
        $log = "WORKER Error ";
        $log .= json_encode($data);
        $this->log->alert($log);
        if ($this->onErrorHandel != null) {
            sd_call_user_func($this->onErrorHandel, '【！！！】服务器进程异常退出', $log);
        }
    }
1223190
public function onReceive(SwServer $server, $fd, $fromId, $data): void
    {
        $data = trim($data);
        $this->log("Receive data [$data] from client [FD:$fd].");
        $server->send($fd, "I have been received your message.\n");

        // $this->onTaskReceive($server, $fd, $fromId, $data);

        // 群发收到的消息
        // $this->reloadWorker->write($data);
    }
508901
Sink newMessageSink(int formatOpcode, long contentLength) {
    if (activeWriter) {
      throw new IllegalStateException("Another message writer is active. Did you call close()?");
    }
    activeWriter = true;

    // Reset FrameSink state for a new writer.
    frameSink.formatOpcode = formatOpcode;
    frameSink.contentLength = contentLength;
    frameSink.isFirstFrame = true;
    frameSink.closed = false;

    return frameSink;
  }
536402
protected function registerPipeMessageCallback()
    {
        $this->swooleServer->on('PipeMessage', function ($server, $srcWorkerId, $data) {
            $this->events->dispatch(new Events\PipeMessageReceived($this, $srcWorkerId, $data));

            $this->dispatchEvent('PipeMessage', $this, $srcWorkerId, $data);
        });
    }
347625
protected function createExchangeForMessageFromURI(MessageInterface $message, $from)
    {
        $version = $message->getContext()->get(Context::FLOWS_VERSION);
        $params = $this->itineraryResolver->getItineraryParams($from, $version);
        $itinerary = $params[InternalRouter::KEY_ITINERARY];

        $exchange = new Exchange($message, clone $itinerary);
        $exchange->setHeader(Exchange::HEADER_HANDLER, $this->getId());
        $exchange->setHeader(Exchange::HEADER_FROM, $from);

        foreach ($this->itineraryResolver->filterItineraryParamsToPropagate($params) as $key => $value) {
            $exchange->setHeader($key, $value);
        }

        return $exchange;
    }