Swoole 异步任务响应事件

@param  \Swoole\Server   $server
@param  int              $taskId
@param  int              $fromId
@param  array            $data

@return array
@throws \RuntimeException
328351
public function onTask(Server $server, int $taskId, int $fromId, array $data): array
    {
        $result = [
            'name' => $data['name'],
            'result' => []
        ];

        array_walk($this->handlers[$data['name']], function ($name) use ($result) {
            $handler = $this->createHandler($name);
            $result[$name][] = $handler->handle($data['parameters']);
            unset($handler);
        });

        return $result;
    }
186902
public function sendToUids($uids, $data, $fromDispatch = false)
    {
        $current_fds = [];
        foreach ($uids as $key => $uid) {
            if ($this->uid_fd_table->exist($uid)) {
                $current_fds[] = $this->uid_fd_table->get($uid)['fd'];
                unset($uids[$key]);
            }
        }
        if (count($current_fds) > $this->send_use_task_num && $this->task_num > 0) {//过多人就通过task
            $task_data = $this->packServerMessageBody(SwooleMarco::MSG_TYPE_SEND_BATCH, ['data' => $data, 'fd' => $current_fds]);
            if ($this->isTaskWorker()) {
                $this->onSwooleTask($this->server, 0, 0, $task_data);
            } elseif ($this->isWorker()) {
                $this->server->task($task_data);
            } else {
                foreach ($current_fds as $fd) {
                    $this->send($fd, $data, true);
                }
            }
        } else {
            foreach ($current_fds as $fd) {
                $this->send($fd, $data, true);
            }
        }
        if ($fromDispatch) {
            return;
        }
        //本机处理不了的发给dispatch
        if ($this->isCluster()) {
            if (count($uids) > 0) {
                ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_sendToUids(array_values($uids), $data);
            }
        }
    }
999476
public static function runImmediateTask($taskId)
    {
        $command = new ProcessBuilder();
        $command->setWorkingDirectory(Yii::getAlias('@app'));
        $command->setPrefix(self::getPhpBinary());
        if (strncasecmp(PHP_OS, 'WIN', 3) === 0) {
            $command->add('yii');
        } else {
            $command->add('./yii');
        }
        $command->add('deferred/index')->add("$taskId");
        $process = $command->getProcess();
        $process->setWorkingDirectory(Yii::getAlias('@app'));
        $process->disableOutput();
        if (strncasecmp(PHP_OS, 'WIN', 3) === 0) {
            $process->setCommandLine('start /b "deferred" ' . $process->getCommandLine());
        } else {
            $process->setCommandLine($process->getCommandLine() . ' &');
        }
        if (isset(Yii::$app->params['deferred.env'])) {
            $process->setEnv(ArrayHelper::merge(Yii::$app->params['deferred.env'], $process->getEnv()));
        }
        $process->mustRun();
    }
1132841
public function dequeue($taskId = null)
    {
        $idList = $this->getIds();
        if ((null === $taskId) && (false === ($taskId = current($idList)))) {
            return null;
        }

        if (!in_array($taskId, $idList)) {
            // Not in list, get out.
            return null;
        }

        $this->getConfig()->remove($taskId);

        return $this->getTask($taskId);
    }
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);
    }
  }
910778
public function onPacket(SwServer $server, string $data, array $clientInfo)
    {
        // {{ log
        $this->get('logger')->info('接收数据', [
            'type' => 'UDP',
            'client' => $clientInfo,
            'data' => $data
        ]);
        // }}

        $this->callProtocolMethod('onPacket', $server, $data, $clientInfo);
    }
704189
private function handlePublishEntity(ConnectionInterface $from, $data) {
    $this->logger->notice(
        "Received an entity publish! ".
          "({$data['guid']}, {$data['event']}, {$from->resourceId})"
    );

    $etype = $data['etype'];

    if (!key_exists($etype, $this->querySubs)) {
      return;
    }

    foreach ($this->querySubs[$etype] as $curQuery => $curClients) {
      $updatedClients = new \SplObjectStorage();

      if ($data['event'] === 'delete' || $data['event'] === 'update') {
        // Check if it is in any client's currents.
        foreach ($curClients as $curClient) {
          $curData = $curClients[$curClient];
          if (in_array($data['guid'], $curData['current'])) {
            // Update currents list.
            $queryArgs = unserialize($curQuery);
            $this->prepareSelectors($queryArgs);
            $queryArgs[0]['source'] = 'client';
            $queryArgs[] = ['&', 'guid' => $data['guid']];
            $token = null;
            if ($this->sessions->contains($curClient)) {
              $token = $this->sessions[$curClient];
            }
            if (class_exists('\Tilmeld\Tilmeld') && isset($token)) {
              $user = \Tilmeld\Tilmeld::extractToken($token);
              if ($user) {
                // Log in the user for access controls.
                \Tilmeld\Tilmeld::fillSession($user);
              }
            }
            $current = call_user_func_array(
                "\Nymph\Nymph::getEntity",
                $queryArgs
            );

            if (isset($current)) {
              // Notify subscriber.
              $this->logger->notice(
                  "Notifying client of update! ".
                    "({$curClient->resourceId})"
              );
              if (is_callable([$current, 'updateDataProtection'])) {
                $current->updateDataProtection();
              }
              $curClient->send(json_encode([
                'query' => $curData['query'],
                'updated' => $data['guid'],
                'data' => $current
              ]));
            } else {
              $curData['current'] = array_diff(
                  $curData['current'],
                  [$data['guid']]
              );
              $curClients[$curClient] = $curData;

              // Notify subscriber.
              $this->logger->notice(
                  "Notifying client of removal! ".
                    "({$curClient->resourceId})"
              );
              $curClient->send(json_encode([
                'query' => $curData['query'],
                'removed' => $data['guid']
              ]));
            }

            if (class_exists('\Tilmeld\Tilmeld') && isset($token)) {
              // Clear the user that was temporarily logged in.
              \Tilmeld\Tilmeld::clearSession();
            }

            $updatedClients->attach($curClient);
          }
        }
      }

      if ($data['event'] === 'create' || $data['event'] === 'update') {
        // Check if it matches the query.
        $selectors = unserialize($curQuery);
        $this->prepareSelectors($selectors);
        $options = $selectors[0];
        unset($selectors[0]);
        $entityData = $data['entity']['data'];
        $entityData['cdate'] = $data['entity']['cdate'];
        $entityData['mdate'] = $data['entity']['mdate'];
        $entitySData = [];

        if (class_exists($options['class'])
            && class_exists($data['entity']['class'])
            && $options['class']::ETYPE === $data['entity']['class']::ETYPE
            && \Nymph\Nymph::checkData(
                $entityData,
                $entitySData,
                $selectors,
                $data['guid'],
                $data['entity']['tags']
            )
          ) {
          // It does match the query.
          foreach ($curClients as $curClient) {
            if ($updatedClients->contains($curClient)) {
              // The user was already notified. (Of an update.)
              continue;
            }

            // Check that the user can access the entity.
            $queryArgs = unserialize($curQuery);
            $this->prepareSelectors($queryArgs);
            $queryArgs[0]['source'] = 'client';
            $queryArgs[] = ['&', 'guid' => $data['guid']];
            $token = null;
            if ($this->sessions->contains($curClient)) {
              $token = $this->sessions[$curClient];
            }
            if (class_exists('\Tilmeld\Tilmeld') && isset($token)) {
              $user = \Tilmeld\Tilmeld::extractToken($token);
              if ($user) {
                // Log in the user for access controls.
                \Tilmeld\Tilmeld::fillSession($user);
              }
            }
            $current = call_user_func_array(
                "\Nymph\Nymph::getEntity",
                $queryArgs
            );

            if (isset($current)) {
              // Update the currents list.
              $curData = $curClients[$curClient];
              $curData['current'][] = $data['guid'];
              $curClients[$curClient] = $curData;

              // Notify client.
              $this->logger->notice(
                  "Notifying client of new match! ".
                    "({$curClient->resourceId})"
              );
              if (is_callable([$current, 'updateDataProtection'])) {
                $current->updateDataProtection();
              }
              $curClient->send(json_encode([
                'query' => $curData['query'],
                'added' => $data['guid'],
                'data' => $current
              ]));
            }

            if (class_exists('\Tilmeld\Tilmeld') && isset($token)) {
              // Clear the user that was temporarily logged in.
              \Tilmeld\Tilmeld::clearSession();
            }
          }
        }
      }
    }
  }
954047
public function sendToUid($uid, $data, $fromDispatch = false)
    {
        if ($this->uid_fd_table->exist($uid)) {//本机处理
            $fd = $this->uid_fd_table->get($uid)['fd'];
            $this->send($fd, $data, true);
        } else {
            if ($fromDispatch) {
                $this->onUidSendFail($uid, $data, null);

                return;
            }
            if ($this->isCluster()) {
                ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_sendToUid($uid, $data);
            } else {
                $this->onUidSendFail($uid, $data, null);
            }
        }
    }
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());
        }
    }
221504
public function onPipeMessage(SwServer $server, int $fromId, string $message)
    {
        // {{ log
        $this->get('logger')->info('接收管道消息', [
            'fromWorkerId' => $fromId,
            'message' => $message
        ]);
        // }}

        $this->callProtocolMethod('onPipeMessage', $server, $fd, $fromId, $data);
    }
977317
public function kickUid($uid, $fromDispatch = false)
    {
        if ($this->uid_fd_table->exist($uid)) {//本机处理
            $fd = $this->uid_fd_table->get($uid)['fd'];
            $this->close($fd);
        } else {
            if ($fromDispatch) {
                return;
            }
            if ($this->isCluster()) {
                ProcessManager::getInstance()->getRpcCall(ClusterProcess::class, true)->my_kickUid($uid);
            }
        }
    }
112580
private function handleSubscriptionUid(ConnectionInterface $from, $data) {
    if ($data['action'] === 'subscribe') {
      // Client is subscribing to a UID.
      if (!key_exists($data['uid'], $this->uidSubs)) {
        $this->uidSubs[$data['uid']] = new \SplObjectStorage();
      }
      $this->uidSubs[$data['uid']][$from] = [
        'count' => !!$data['count']
      ];
      $this->logger->notice(
          "Client subscribed to a UID! ".
            "({$data['uid']}, {$from->resourceId})"
      );

      if (Server::$config['broadcast_counts']) {
        // Notify clients of the subscription count.
        $count = count($this->uidSubs[$data['uid']]);
        foreach ($this->uidSubs[$data['uid']] as $key) {
          $curData = $this->uidSubs[$data['uid']][$key];
          if ($curData['count']) {
            $key->send(json_encode([
              'uid' => $data['uid'],
              'count' => $count
            ]));
          }
        }
      }
    }

    if ($data['action'] === 'unsubscribe') {
      // Client is unsubscribing from a UID.
      if (!key_exists($data['uid'], $this->uidSubs)) {
        return;
      }
      if (!$this->uidSubs[$data['uid']]->contains($from)) {
        return;
      }
      unset($this->uidSubs[$data['uid']][$from]);
      $this->logger->notice(
          "Client unsubscribed from a UID! ".
            "({$data['uid']}, {$from->resourceId})"
      );

      $count = count($this->uidSubs[$data['uid']]);

      if ($count === 0) {
        // No more subscribed clients.
        unset($this->uidSubs[$data['uid']]);
        return;
      }

      if (Server::$config['broadcast_counts']) {
        // Notify clients of the subscription count.
        $count = count($this->uidSubs[$data['uid']]);
        foreach ($this->uidSubs[$data['uid']] as $key) {
          $curData = $this->uidSubs[$data['uid']][$key];
          if ($curData['count']) {
            $key->send(json_encode([
              'uid' => $data['uid'],
              'count' => $count
            ]));
          }
        }
      }
    }
  }
812959
public function onReceive($server, $fd, $from_id, $data) {
		$this->handle(__METHOD__, [$fd, $from_id, $data]);
	}
361349
public function callBack($appId, $fromClient, $to, $fromSerNum = null, $toSerNum = null, $type = 'json')
    {
        $url = self::BaseUrl . self::SoftVersion . '/Accounts/' . $this->accountSid . '/Calls/callBack?sig=' . $this->getSigParameter();
        if ($type === 'json') {
            $body_json = array('callback' => array(
                'appId'      => $appId,
                'fromClient' => $fromClient,
                'fromSerNum' => $fromSerNum,
                'to'         => $to,
                'toSerNum'   => $toSerNum,
            ));
            $body = json_encode($body_json);
        } elseif ($type === 'xml') {
            $body_xml = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
                        <callback>
                            <fromClient>' . $fromClient . '</clientNumber>
                            <fromSerNum>' . $fromSerNum . '</chargeType>
                            <to>' . $to . '</charge>
                            <toSerNum>' . $toSerNum . '</toSerNum>
                            <appId>' . $appId . '</appId>
                        </callback>';
            $body = trim($body_xml);
        } else {
            throw new Exception('只能json或xml，默认为json');
        }
        $data = $this->getResult($url, $body, $type, 'post');

        return $data;
    }
120550
public static function sendToUDP(string $data, $addr, $port = 0, $from = -1) {
		self::$server->sendto($addr, $port, $data);
	}
1266320
public function onTask($server, $taskId, $srcWorkerId, $data)
    {
        $this->container->event->trigger('swoole.task', func_get_args());

        try {
            // push websocket message
            if ($this->isWebsocketPushPayload($data)) {
                $this->pushMessage($server, $data['data']);
                // push async task to queue
            }
        } catch (Throwable $e) {
            $this->logServerError($e);
        }
    }
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;
        }
    }