接受
@param $cli
@param $data
355500
public function onReceive($cli, $data)
    {
        $message_object = $this->message_read($data);
        switch ($message_object->getMessageType()) {
            case Message::CONNACK:
                Debug::Log(Debug::INFO, 'connect(): connected=1');
                # save current time for ping
                $this->connected_time = time();
                # Call connect
                $this->call_handler('connack', array($this, $message_object));
                break;
            case Message::PINGRESP:
                array_shift($this->ping_queue);
                Debug::Log(Debug::INFO, 'loop(): received PINGRESP');

                $this->last_ping_time = time();

                break;

            # Process PUBLISH
            # in: Client <- Server, Step 1
            case Message::PUBLISH:
                /**
                 * @var Message\PUBLISH $message_object
                 */

                Debug::Log(Debug::INFO, 'loop(): received PUBLISH');

                $qos = $message_object->getQoS();

                $msgid = $message_object->getMsgID();

                if ($qos == 0) {
                    Debug::Log(Debug::DEBUG, 'loop(): PUBLISH QoS=0 PASS');
                    # call handler
                    //0直接执行
                    $this->call_handler('publish', array($this, $message_object));
                    # Do nothing
                } else if ($qos == 1) {
                    # call handler
                    //1先执行再回复，可以结合飞行窗口实现顺序发送
                    $this->call_handler('publish', array($this, $message_object));
                    # PUBACK
                    $puback_bytes_written = $this->simpleCommand(Message::PUBACK, $msgid);
                    Debug::Log(Debug::DEBUG, 'loop(): PUBLISH QoS=1 PUBACK written=' . $puback_bytes_written);

                } else if ($qos == 2) {
                    # PUBREC
                    $pubrec_bytes_written = $this->simpleCommand(Message::PUBREC, $msgid);
                    Debug::Log(Debug::DEBUG, 'loop(): PUBLISH QoS=2 PUBREC written=' . $pubrec_bytes_written);
                    $this->cmdstore->addWait(
                        Message::PUBREL,
                        $msgid,
                        array(
                            'msgid' => $msgid,
                            'retry_after' => time() + $this->retry_timeout,
                        )
                    );
                    # call handler
                    //2先回复收到再执行
                    $this->call_handler('publish', array($this, $message_object));
                } else {
                    # wrong packet
                    Debug::Log(Debug::WARN, 'loop(): PUBLISH Invalid QoS');
                }
                break;

            # Process PUBACK
            # in: Client -> Server, QoS = 1, Step 2
            case Message::PUBACK:

                /**
                 * @var Message\PUBACK $message_object
                 */

                # Message has been published (QoS 1)

                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received PUBACK msgid=' . $msgid);
                # Verify Packet Identifier
                $this->call_handler('puback', array($this, $message_object));

                $this->cmdstore->delWait(Message::PUBACK, $msgid);
                break;

            # Process PUBREC, send PUBREL
            # in: Client -> Server, QoS = 2, Step 2
            case Message::PUBREC:
                /**
                 * @var Message\PUBREC $message_object
                 */

                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received PUBREC msgid=' . $msgid);

                $this->cmdstore->delWait(Message::PUBREC, $msgid);

                # PUBREL
                Debug::Log(Debug::INFO, 'loop(): send PUBREL msgid=' . $msgid);
                $pubrel_bytes_written = $this->simpleCommand(Message::PUBREL, $msgid);


                $this->cmdstore->addWait(
                    Message::PUBCOMP,
                    $msgid,
                    array(
                        'msgid' => $msgid,
                        'retry_after' => time() + $this->retry_timeout,
                    )
                );

                Debug::Log(Debug::DEBUG, 'loop(): PUBREL QoS=2 PUBREL written=' . $pubrel_bytes_written);

                $this->call_handler('pubrec', array($this, $message_object));
                break;


            # Process PUBREL
            # in: Client <- Server, QoS = 2, Step 3
            case Message::PUBREL:
                /**
                 * @var Message\PUBREL $message_object
                 */

                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received PUBREL msgid=' . $msgid);

                $this->cmdstore->delWait(Message::PUBREL, $msgid);

                # PUBCOMP
                Debug::Log(Debug::INFO, 'loop(): send PUBCOMP msgid=' . $msgid);
                $pubcomp_bytes_written = $this->simpleCommand(Message::PUBCOMP, $msgid);

                Debug::Log(Debug::DEBUG, 'loop(): PUBREL QoS=2 PUBCOMP written=' . $pubcomp_bytes_written);

                $this->call_handler('pubrel', array($this, $message_object));
                break;

            # Process PUBCOMP
            # in: Client -> Server, QoS = 2, Step 4
            case Message::PUBCOMP:

                # Message has been published (QoS 2)

                /**
                 * @var Message\PUBCOMP $message_object
                 */

                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received PUBCOMP msgid=' . $msgid);

                $this->cmdstore->delWait(Message::PUBCOMP, $msgid);

                $this->call_handler('pubcomp', array($this, $message_object));
                break;

            # Process SUBACK
            case Message::SUBACK:

                /**
                 * @var Message\SUBACK $message_object
                 */

                $return_codes = $message_object->getReturnCodes();
                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received SUBACK msgid=' . $msgid);

                if (!isset($this->subscribe_awaits[$msgid])) {
                    Debug::Log(Debug::WARN, 'loop(): SUBACK Message identifier not found: ' . $msgid);
                } else {
                    if (count($this->subscribe_awaits[$msgid]) != count($return_codes)) {
                        Debug::Log(Debug::WARN, 'loop(): SUBACK returned qos list doesn\'t match SUBSCRIBE');
                    } else {
                        # save max_qos list from suback
                        foreach ($return_codes as $k => $tqos) {
                            if ($return_codes != 0x80) {
                                $this->topics[$this->subscribe_awaits[$msgid][$k][0]] = $tqos;
                            } else {
                                Debug::Log(
                                    Debug::WARN,
                                    "loop(): Failed to subscribe '{$this->subscribe_awaits[$msgid][$k][0]}'. Request QoS={$this->subscribe_awaits[$msgid][$k][1]}"
                                );
                            }
                        }
                    }
                }

                $this->call_handler('suback', array($this, $message_object));

                break;

            # Process UNSUBACK
            case Message::UNSUBACK:
                /**
                 * @var Message\UNSUBACK $message_object
                 */

                $msgid = $message_object->getMsgID();
                Debug::Log(Debug::INFO, 'loop(): received UNSUBACK msgid=' . $msgid);

                if (!isset($this->unsubscribe_awaits[$msgid])) {
                    Debug::Log(Debug::WARN, 'loop(): UNSUBACK Message identifier not found ' . $msgid);
                } else {
                    foreach ($this->unsubscribe_awaits[$msgid] as $topic) {
                        Debug::Log(Debug::WARN, "loop(): Unsubscribe topic='{$topic}'");
                        unset($this->topics[$topic]);
                    }
                }

                $this->call_handler('unsuback', array($this, $message_object));
                break;

            default:
                return;
        }
    }
1131906
function command($cmd, $callback) {
        /**
         * 如果已经连接，直接发送数据
         */
        if ($this->client->isConnected()) {
            $this->client->send($cmd);
        } /**
         * 未连接，等待连接成功后发送数据
         */
        else {
            $this->wait_send = $cmd;
        }
        $this->callback = $callback;
        //从空闲连接池中移除，避免被其他任务使用
        $this->redis->lockConnection($this->client->sock);
    }
1338243
static void parseData(String[] args, BiConsumer<String, Boolean> data) {
        CommandLine commandLine = parseCommandLine(args);
        Object value = commandLine.optionValue("d");
        if (value != null) {
            data.accept(value.toString(), commandLine.hasOption("x"));
        } else {
            data.accept(null, commandLine.hasOption("x"));
        }
    }
721888
public static function useConfig($data): void
    {
        defined('KOLDY_START') || define('KOLDY_START', microtime(true));
        defined('KOLDY_CLI') || define('KOLDY_CLI', PHP_SAPI == 'cli');
        defined('DS') || define('DS', DIRECTORY_SEPARATOR); // this is just shorthand for Directory Separator

        if (!is_string($data) && !is_array($data)) {
            static::terminateWithError('Can not start application, expected array or string (path to config file) when useConfig is called; got ' . gettype($data));
        }

        $configInstance = new Config('application');

        if (is_string($data)) {
            $applicationConfigPath = stream_resolve_include_path($data); // this is path to /application/configs/application.php

            if ($applicationConfigPath === false || !is_file($applicationConfigPath)) {
                static::terminateWithError('Can not resolve the full path to the main application config file or file doesn\'t exists!');
            }

            try {
                $configInstance->loadFrom($applicationConfigPath);
            } catch (ConfigException $e) {
                static::terminateWithError($e->getMessage());
            }
        } else {
            $configInstance->setData($data);
        }

        static::$publicPath = dirname($_SERVER['SCRIPT_FILENAME']) . '/';

        // now, let's examine configuration we got - it's getting more strict now
        try {
            $siteUrl = $configInstance->get('site_url');

            if ($siteUrl == null) {
                static::$domain = $_SERVER['HTTP_HOST'] ?? null;
                static::$isSSL = isset($_SERVER['HTTPS']) && $_SERVER['HTTPS'] != 'off';

            } else if (is_string($siteUrl)) {
                $siteUrl = explode('//', $siteUrl);

                if (count($siteUrl) != 2) {
                    throw new ConfigException('Defined site_url in application config must contain double slash');
                }

                static::$domain = $siteUrl[1];
                static::$isSSL = substr($siteUrl[0], 0, 6) == 'https:';

            } else if (is_array($siteUrl)) {
                if (count($siteUrl) === 0) {
                    throw new ConfigException('If site_url is defined as array, then it must be non-empty array');
                }

                if (isset($_SERVER['HTTP_HOST'])) {
                    foreach ($siteUrl as $domainWithSchema) {
                        $doubleSlashPosition = strpos($domainWithSchema, '//');

                        if ($doubleSlashPosition === false) {
                            throw new ConfigException('Every defined site_url in application config must contain double slash (//)');
                        }

                        $domain = substr($domainWithSchema, $doubleSlashPosition + 2);

                        if ($domain == $_SERVER['HTTP_HOST']) {

                            static::$domain = $domain;
                            static::$isSSL = substr($domainWithSchema, 0, 6) == 'https:';
                        }
                    }
                } else {
                    // we are probably in CLI mode, so let's take the first value from 'site_url' config
                    $domainWithSchema = $siteUrl[0];

                    $doubleSlashPosition = strpos($domainWithSchema, '//');

                    if ($doubleSlashPosition === false) {
                        throw new ConfigException('Every defined site_url in application config must contain double slash (//)');
                    }

                    $domain = substr($domainWithSchema, $doubleSlashPosition + 2);

                    static::$domain = $domain;
                    static::$isSSL = substr($domainWithSchema, 0, 6) == 'https:';
                }

                // beware then static::$domain might stay null, in this case, in run(), redirect to first domain if site_url is array
            }
        } catch (ConfigException $e) {
            static::terminateWithError($e->getMessage());
        }


        $paths = $configInstance->get('paths');

        if (!is_array($paths)) {
            static::terminateWithError('Application config doesn\'t have paths array defined');
        }

        ///// APPLICATION PATH /////
        if (!isset($paths['application'])) {
            static::terminateWithError('Path to \'application\' in \'paths\' is not defined');
        } else if (substr($paths['application'], -1) != '/') {
            static::terminateWithError('Path to \'application\' in \'paths\' must end with slash (/)');
        }

        static::$applicationPath = $paths['application'];


        ///// STORAGE PATH /////
        if (!isset($paths['storage'])) {
            static::terminateWithError('Path to \'storage\' in \'paths\' is not defined');
        } else if (substr($paths['storage'], -1) != '/') {
            static::terminateWithError('Path to \'storage\' in \'paths\' must end with slash (/)');
        }

        static::$storagePath = $paths['storage'];


        ///// VIEW PATH /////
        if (isset($paths['view']) && substr($paths['storage'], -1) != '/') {
            static::terminateWithError('Path to \'storage\' in \'paths\' must end with slash (/)');
        }

        static::$viewPath = $paths['view'] ?? (static::$applicationPath . 'views/');


        ///// MODULES PATH /////
        if (isset($paths['modules']) && substr($paths['modules'], -1) != '/') {
            static::terminateWithError('Path to \'modules\' in \'paths\' must end with slash (/)');
        }

        static::$modulePath = $paths['modules'] ?? (static::$applicationPath . 'modules/');


        ///// CONFIGS PATH /////
        if (isset($paths['configs']) && substr($paths['configs'], -1) != '/') {
            static::terminateWithError('Path to \'configs\' in \'paths\' must end with slash (/)');
        }

        static::$configsPath = $paths['configs'] ?? (static::$applicationPath . 'configs/');


        ///// SCRIPTS PATH /////
        if (isset($paths['scripts']) && substr($paths['scripts'], -1) != '/') {
            static::terminateWithError('Path to \'scripts\' in \'paths\' must end with slash (/)');
        }

        static::$scriptsPath = $paths['scripts'] ?? (static::$applicationPath . 'scripts/');


        // check the environment
        $env = $configInstance->get('env');
        if ($env === null || !($env == static::DEVELOPMENT || $env == static::PRODUCTION || $env == static::TEST)) {
            static::terminateWithError('Invalid ENV parameter in main application config');
        }
        static::$env = $env;

        $key = $configInstance->get('key');
        if (!isset($key) || !is_string($key) || strlen($key) > 32) {
            static::terminateWithError('Invalid unique key in main application config. It has to be max 32 chars long.');
        }

        $timezone = $configInstance->get('timezone');
        if (!isset($timezone)) {
            static::terminateWithError('Timezone is not set in main application config');
        }

        static::$isLive = $configInstance->get('live') === true;
        static::$configs['application'] = $configInstance;
	    static::$appName = $configInstance->get('app_name');
    }
225093
public static function output($data)
    {
        if (!self::$output || !is_string($data)) {
            return;
        }

        if (php_sapi_name() != 'cli') {
            echo sprintf(self::$style['called_from_format'], Debugger::getCalledFrom(2));
            echo sprintf(self::$style['output_format'], $data);
            echo '<style type="text/css">.xicrow-php-debug-timer:hover{ ' . self::$style['hover_style'] . ' }</style>';
        } else {
            echo Debugger::getCalledFrom(2);
            echo "\n";
            echo $data;
        }
    }
861349
public static function show(array $data, array $opts = []): void
    {
        static $counter = 0;
        static $started = 1;

        if ($started) {
            $started = 0;
            $opts    = \array_merge([
                // 'char' => Cli::isSupportColor() ? '─' : '-', // ——
                'char'        => '-',
                'prefix'      => Cli::isSupportColor() ? '├' : '|',
                'leftPadding' => '',
            ], $opts);

            $opts['_level']   = 1;
            $opts['_is_main'] = true;

            Console::startBuffer();
        }

        foreach ($data as $key => $value) {
            if (\is_scalar($value)) {
                $counter++;
                $leftString = $opts['leftPadding'] . \str_pad($opts['prefix'], $opts['_level'] + 1, $opts['char']);

                Console::write($leftString . ' ' . FormatUtil::typeToString($value));
            } elseif (\is_array($value)) {
                $newOpts             = $opts;
                $newOpts['_is_main'] = false;
                $newOpts['_level']++;

                self::show($value, $newOpts);
            }
        }

        if ($opts['_is_main']) {
            Console::write('node count: ' . $counter);
            // var_dump('f');
            Console::flushBuffer();

            // reset.
            $counter = $started = 0;
        }
    }
34566
function route($f3,$params) {
        if (PHP_SAPI!='cli' && !$this->web)
            $f3->error(404);
        $exec=isset($params['job'])?
            array($params['job']=>$this->execute($params['job'],FALSE)):
            $this->run();
        if (!$this->silent) {
            if (PHP_SAPI!='cli')
                header('Content-Type: text/plain');
            if (!$exec)
                die('Nothing to do');
            foreach($exec as $job=>$ok)
                echo sprintf('%s [%s]',$job,$ok?'OK':'async')."\r\n";
        }
    }
787340
public function insert(array $data, $options = [])
    {
        // 分析并处理数据
        $data = $this->parseData($data, $options);
        $bulk = new BulkWrite;
        if ($insertId = $bulk->insert($data)) {
            $this->insertId = $insertId;
        }
        $this->log('insert', $data, $options);
        return $bulk;
    }
467902
public static function addCommands(Application $cli)
    {
        $cli->addCommands([
            new Command\PublishMessageCommand(),
            new Command\PurgeQueueCommand(),
            new Command\SetupFabricCommand(),
            new Command\ShowCommand(),
            new Command\StartCallbackConsumerCommand(),
            new Command\StartJsonRpcServerCommand(),
        ]);
    }
165657
public function sendPluginCmd($plugin, $data)
  {
    $this->execute("plugincmd", array("name" => $plugin, "data" => $data, "targetmode" => TeamSpeak3::PLUGINCMD_CLIENT, "target" => $this->getId()));
  }
1107457
private function init($data=null)
    {
        if (!is_null($this->stack)) return $this;
        
        if (!is_null($data))  $this->data = $data;
        if (!is_array($this->data) || empty($this->data)) return $this;
        
        // 先转为 id 为键的数据
        $t_data = array();
        $pids   = array();
        foreach ($this->data as $k => $v) {
            if (!isset($v[$this->fields[0]])) {
                throw new \InvalidArgumentException('tree fields error:' . implode(',', $this->fields));
                break; // 如果数组结构不符合，跳过
            }
            $t_data[$v[$this->fields[0]]] = $v;
            $pids[$v[$this->fields[1]]] = $v[$this->fields[1]];
        }
        $this->data = & $t_data;
        
        // 修正root，当parent_id=root不存在时取最小的parent_id作为root
        if (!isset($pids[$this->root])) {
            ksort($pids);
            $this->root = array_keys($pids)[0];
        }
        
        // 先生成父子级的二维的数组
        foreach ($this->data as $node) {
            $this->stack[$node[$this->fields[1]]][$node[$this->fields[0]]] = $node;
        }
        
        
        // 递归遍历生成 完整的带有 items 的数组
        foreach ($this->stack as $k => $v) {
            $this->handleStackElement($k);
        }
        //print_r($this->stack);exit;
    }
1105596
public function put($id = null, $data = array())
    {

        $args = func_get_args();
        if(count($args) == 2){
            $args = $id;
        }else{
            $data = array_pop($args);
        }

        try {
            $this->client->request($this, 'put', $args, $data);
        } catch(Exception $ex) {
            $this->dispatchException($ex);
        }

        return true;
    }
872037
public function post($path, $data, $callback)
    {
        // 解决swoole底层http_build_query failed的问题
        if (empty($data)) {
            $data = '';
        }
        $this->client->post($path, $data, $callback);
    }
66676
static public function addCommands(Application $cli)
    {
        $cli->addCommands(array(
            // DBAL Commands
            new \Doctrine\DBAL\Tools\Console\Command\RunSqlCommand(),
            new \Doctrine\DBAL\Tools\Console\Command\ImportCommand(),

            // ORM Commands
            new \Doctrine\ORM\Tools\Console\Command\ClearCache\MetadataCommand(),
            new \Doctrine\ORM\Tools\Console\Command\ClearCache\ResultCommand(),
            new \Doctrine\ORM\Tools\Console\Command\ClearCache\QueryCommand(),
            new \Doctrine\ORM\Tools\Console\Command\SchemaTool\CreateCommand(),
            new \Doctrine\ORM\Tools\Console\Command\SchemaTool\UpdateCommand(),
            new \Doctrine\ORM\Tools\Console\Command\SchemaTool\DropCommand(),
            new \Doctrine\ORM\Tools\Console\Command\EnsureProductionSettingsCommand(),
            new \Doctrine\ORM\Tools\Console\Command\ConvertDoctrine1SchemaCommand(),
            new \Doctrine\ORM\Tools\Console\Command\GenerateRepositoriesCommand(),
            new \Doctrine\ORM\Tools\Console\Command\GenerateEntitiesCommand(),
            new \Doctrine\ORM\Tools\Console\Command\GenerateProxiesCommand(),
            new \Doctrine\ORM\Tools\Console\Command\ConvertMappingCommand(),
            new \Doctrine\ORM\Tools\Console\Command\RunDqlCommand(),
            new \Doctrine\ORM\Tools\Console\Command\ValidateSchemaCommand(),
            new \Doctrine\ORM\Tools\Console\Command\InfoCommand(),
            new \Doctrine\ORM\Tools\Console\Command\MappingDescribeCommand(),
        ));
    }
1351553
static function setEnvironment($data = []){
	
		if($data && is_array($data)){
			self::$env = $data;
			return;
		}
		
		// Check environment is set?
		if(count(self::$env ) == 0){
			self::$env = (require APP_PATH.'config' . DS . 'Environment.php');
		}
	}
815099
public static function trigger($event, $data = null)
	{
		if (!is_array($data)) {
			$data = array($data);
		}

		if (empty(self::$events[$event])) {
			return true;
		}
		
		foreach (self::$events[$event] as $callback) {
			if (!is_callable($callback)) {
				continue;
			}
			call_user_func_array($callback, $data);
		}
		
		return true;
	}
332599
function addData($data, $col_id = 0, $row_id = 0) {
        foreach ($data as $row) {
            if ($row === self::CONSOLE_TABLE_HORIZONTAL_RULE) {
                $this->_data[$row_id] = self::CONSOLE_TABLE_HORIZONTAL_RULE;
                $row_id++;
                continue;
            }
            $starting_col = $col_id;
            foreach ($row as $cell) {
                $this->_data[$row_id][$starting_col++] = $cell;
            }
            $this->_updateRowsCols();
            $this->_max_cols = max($this->_max_cols, $starting_col);
            $row_id++;
        }
    }