批量发送消息.

@param $uids
@param $data
@param $fromDispatch

@throws SwooleException
@throws \Exception
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);
            }
        }
    }
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;
    }
749520
public function update($data = null): IResult
    {
        $query = static::query($this);
        $query = $this->parseWhereId($query);
        if(null === $data)
        {
            $data = static::parseSaveData($this);
        }
        else if(!$data instanceof \ArrayAccess)
        {
            $data = new LazyArrayObject($data);
        }

        // 更新前
        $this->trigger(ModelEvents::BEFORE_UPDATE, [
            'model' => $this,
            'data'  => $data,
            'query' => $query,
        ], $this, \Imi\Model\Event\Param\BeforeUpdateEventParam::class);

        if(!isset($query->getOption()->where[0]))
        {
            throw new \RuntimeException('use Model->update(), primary key can not be null');
        }

        $result = $query->update($data);

        // 更新后
        $this->trigger(ModelEvents::AFTER_UPDATE, [
            'model' => $this,
            'data'  => $data,
            'result'=> $result,
        ], $this, \Imi\Model\Event\Param\AfterUpdateEventParam::class);

        // 子模型更新
        ModelRelationManager::updateModel($this);

        return $result;
    }
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);
    }
1086702
function createDifferenceObject( $fromData, $toData )
    {
        $changes = new eZXMLTextDiff();
        $contentINI = eZINI::instance( 'content.ini' );
        $useSimplifiedXML = $contentINI->variable( 'ContentVersionDiffSettings', 'UseSimplifiedXML' );
        $diffSimplifiedXML = ( $useSimplifiedXML == 'enabled' );

        $oldXMLTextObject = $fromData->content();
        $newXMLTextObject = $toData->content();

        $oldXML = $oldXMLTextObject->attribute( 'xml_data' );
        $newXML = $newXMLTextObject->attribute( 'xml_data' );

        $simplifiedXML = new eZSimplifiedXMLEditOutput();

        $domOld = new DOMDocument( '1.0', 'utf-8' );
        $domOld->preserveWhiteSpace = false;
        $domOld->loadXML( $oldXML );

        $domNew = new DOMDocument( '1.0', 'utf-8' );
        $domNew->preserveWhiteSpace = false;
        $domNew->loadXML( $newXML );

        $old = $simplifiedXML->performOutput( $domOld );
        $new = $simplifiedXML->performOutput( $domNew );

        if ( !$diffSimplifiedXML )
        {
            $old = trim( strip_tags( $old ) );
            $new = trim( strip_tags( $new ) );

            $pattern = array( '/[ ][ ]+/',
                              '/ \n( \n)+/',
                              '/^ /m',
                              '/(\n){3,}/' );
            $replace = array( ' ',
                              "\n",
                              '',
                              "\n\n" );

            $old = preg_replace( $pattern, $replace, $old );
            $new = preg_replace( $pattern, $replace, $new );
        }

        $oldArray = explode( "\n", $old );
        $newArray = explode( "\n", $new );

        $oldSums = array();
        foreach( $oldArray as $paragraph )
        {
            $oldSums[] = crc32( $paragraph );
        }

        $newSums = array();
        foreach( $newArray as $paragraph )
        {
            $newSums[] = crc32( $paragraph );
        }

        $textDiffer = new eZDiffTextEngine();

        $pre = $textDiffer->preProcess( $oldSums, $newSums );
        $out = $textDiffer->createOutput( $pre, $oldArray, $newArray );
        $changes->setChanges( $out );
        return $changes;
    }
360270
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        if (!array_key_exists('_type', $data)) {
            throw new Exceptions\Parser("Missing '_type' value for URLAliasCreate.");
        }

        if ($data['_type'] == 'LOCATION') {
            if (!array_key_exists('location', $data)) {
                throw new Exceptions\Parser("Missing 'location' value for URLAliasCreate.");
            }

            if (!is_array($data['location']) || !array_key_exists('_href', $data['location'])) {
                throw new Exceptions\Parser("Missing 'location' > '_href' attribute for URLAliasCreate.");
            }
        } else {
            if (!array_key_exists('resource', $data)) {
                throw new Exceptions\Parser("Missing 'resource' value for URLAliasCreate.");
            }
        }

        if (!array_key_exists('path', $data)) {
            throw new Exceptions\Parser("Missing 'path' value for URLAliasCreate.");
        }

        if (!array_key_exists('languageCode', $data)) {
            throw new Exceptions\Parser("Missing 'languageCode' value for URLAliasCreate.");
        }

        if (array_key_exists('alwaysAvailable', $data)) {
            $data['alwaysAvailable'] = $this->parserTools->parseBooleanValue($data['alwaysAvailable']);
        } else {
            $data['alwaysAvailable'] = false;
        }

        if (array_key_exists('forward', $data)) {
            $data['forward'] = $this->parserTools->parseBooleanValue($data['forward']);
        } else {
            $data['forward'] = false;
        }

        return $data;
    }
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));
      }
    }
  }
180042
public function fromDispatcher(JsUtils $js,$dispatcher,$startIndex=0){
		$this->startIndex=$startIndex;
		return $this->addElements($js->fromDispatcher($dispatcher));
	}
884126
public function th_syncData($node_name, $datas, $type)
    {
        switch ($type) {
            case ClusterProcess::TYPE_UID:
                if (!isset($this->map[$node_name])) {
                    $this->map[$node_name] = new Set($datas);
                } else {
                    $this->map[$node_name]->add(...$datas);
                }
                secho("CLUSTER", "同步$node_name uid信息");
                break;
            case ClusterProcess::TYPE_ACTOR:
                if (!isset($this->actorMap[$node_name])) {
                    $this->actorMap[$node_name] = new Set($datas);
                } else {
                    $this->actorMap[$node_name]->add(...$datas);
                }
                secho("CLUSTER", "同步$node_name actor信息");
                break;
        }
    }
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);
        }
    }
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);
            }
        }
    }
523043
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        $sessionInput = new SessionInputValue();

        if (!array_key_exists('login', $data)) {
            throw new Exceptions\Parser("Missing 'login' attribute for SessionInput.");
        }

        $sessionInput->login = $data['login'];

        if (!array_key_exists('password', $data)) {
            throw new Exceptions\Parser("Missing 'password' attribute for SessionInput.");
        }

        $sessionInput->password = $data['password'];

        return $sessionInput;
    }
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());
        }
    }
1138238
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        if (!array_key_exists('identifier', $data)) {
            throw new Exceptions\Parser("Missing 'identifier' attribute for ObjectStateGroupCreate.");
        }

        $objectStateGroupCreateStruct = $this->objectStateService->newObjectStateGroupCreateStruct($data['identifier']);

        if (!array_key_exists('defaultLanguageCode', $data)) {
            throw new Exceptions\Parser("Missing 'defaultLanguageCode' attribute for ObjectStateGroupCreate.");
        }

        $objectStateGroupCreateStruct->defaultLanguageCode = $data['defaultLanguageCode'];

        if (!array_key_exists('names', $data) || !is_array($data['names'])) {
            throw new Exceptions\Parser("Missing or invalid 'names' element for ObjectStateGroupCreate.");
        }

        if (!array_key_exists('value', $data['names']) || !is_array($data['names']['value'])) {
            throw new Exceptions\Parser("Missing or invalid 'names' element for ObjectStateGroupCreate.");
        }

        $objectStateGroupCreateStruct->names = $this->parserTools->parseTranslatableList($data['names']);

        if (array_key_exists('descriptions', $data) && is_array($data['descriptions'])) {
            $objectStateGroupCreateStruct->descriptions = $this->parserTools->parseTranslatableList($data['descriptions']);
        }

        return $objectStateGroupCreateStruct;
    }
307335
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        // Since ContentTypeGroupInput is used both for creating and updating ContentTypeGroup and identifier is not
        // required when updating ContentTypeGroup, we need to rely on PAPI to throw the exception on missing
        // identifier when creating a ContentTypeGroup
        // @todo Bring in line with XSD which says that identifier is required always

        $contentTypeGroupIdentifier = null;
        if (array_key_exists('identifier', $data)) {
            $contentTypeGroupIdentifier = $data['identifier'];
        }

        $contentTypeGroupCreateStruct = $this->contentTypeService->newContentTypeGroupCreateStruct($contentTypeGroupIdentifier);

        if (array_key_exists('modificationDate', $data)) {
            $contentTypeGroupCreateStruct->creationDate = new DateTime($data['modificationDate']);
        }

        // @todo mainLanguageCode, names, descriptions?

        if (array_key_exists('User', $data) && is_array($data['User'])) {
            if (!array_key_exists('_href', $data['User'])) {
                throw new Exceptions\Parser("Missing '_href' attribute for User element in ContentTypeGroupInput.");
            }

            $contentTypeGroupCreateStruct->creatorId = $this->requestParser->parseHref($data['User']['_href'], 'userId');
        }

        return $contentTypeGroupCreateStruct;
    }
638719
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        if (!array_key_exists('identifier', $data)) {
            throw new Exceptions\Parser("Missing 'identifier' attribute for ObjectStateCreate.");
        }

        $objectStateCreateStruct = $this->objectStateService->newObjectStateCreateStruct($data['identifier']);

        if (!array_key_exists('priority', $data)) {
            throw new Exceptions\Parser("Missing 'priority' attribute for ObjectStateCreate.");
        }

        $objectStateCreateStruct->priority = (int)$data['priority'];

        if (!array_key_exists('defaultLanguageCode', $data)) {
            throw new Exceptions\Parser("Missing 'defaultLanguageCode' attribute for ObjectStateCreate.");
        }

        $objectStateCreateStruct->defaultLanguageCode = $data['defaultLanguageCode'];

        if (!array_key_exists('names', $data) || !is_array($data['names'])) {
            throw new Exceptions\Parser("Missing or invalid 'names' element for ObjectStateCreate.");
        }

        if (!array_key_exists('value', $data['names']) || !is_array($data['names']['value'])) {
            throw new Exceptions\Parser("Missing or invalid 'names' element for ObjectStateCreate.");
        }

        $objectStateCreateStruct->names = $this->parserTools->parseTranslatableList($data['names']);

        // @todo XSD says that descriptions field is mandatory. Does that make sense?
        if (array_key_exists('descriptions', $data) && is_array($data['descriptions'])) {
            $objectStateCreateStruct->descriptions = $this->parserTools->parseTranslatableList($data['descriptions']);
        }

        return $objectStateCreateStruct;
    }
940920
public function parse(array $data, ParsingDispatcher $parsingDispatcher)
    {
        $fieldDefinitionReferences = array();

        foreach ($data['FieldDefinition'] as $fieldDefinitionData) {
            $fieldDefinitionReferences[] = $this->parserTools->parseObjectElement(
                $fieldDefinitionData,
                $parsingDispatcher
            );
        }

        return new Values\FieldDefinitionList(
            $this->contentTypeService,
            $fieldDefinitionReferences
        );
    }