处理管理消息

@param string   $msg
@param resource $socket
335981
private function onMessage($msg, $socket) {
        try {
            $payload = @json_decode($msg, true);
            if (!$payload) {
                $this->response($socket, ['error' => 403, 'message' => 'error request']);

                return;
            }
            $command = isset($payload['command']) ? $payload['command'] : 'status';
            $service = isset($payload['args']['service']) ? $payload['args']['service'] : '';
            switch ($command) {
                case 'stop':
                    if ($service) {
                        $rst = $this->stopService($service);
                        $this->response($socket, ['status' => $rst ? 'done' : 'fail']);
                    } else {
                        $this->response($socket, ['status' => 'done']);
                        $this->shutdown = true;
                        $this->rSignal  = SIGTERM;
                    }
                    break;
                case 'start':
                    if ($service) {
                        $rst = $this->startService($service);
                        $this->response($socket, ['status' => $rst ? 'done' : 'fail']);
                    } else {
                        $this->response($socket, ['status' => 'fail']);
                    }
                    break;
                case 'reload':
                    $rst = $this->reloadConfig(null, $service);
                    $this->response($socket, ['status' => $rst ? 'done' : 'fail']);
                    break;
                case 'ps':
                    if ($service) {
                        if (isset($this->services[ $service ])) {
                            $pids = isset($this->services[ $service ]['pids']) ? $this->services[ $service ]['pids'] : [];
                            $this->response($socket, [
                                'ps'   => [$service => $pids],
                                'ssid' => posix_getpid()
                            ]);
                        } else {
                            $this->response($socket, ['ps' => [], 'ssid' => posix_getpid()]);
                        }
                    } else {
                        $ps = [];
                        foreach ($this->services as $s => $ser) {
                            $ps[ $s ] = isset($ser['pids']) ? $ser['pids'] : [];
                        }
                        $this->response($socket, ['ps' => $ps, 'ssid' => posix_getpid()]);
                    }
                    break;
                case 'status':
                default:
                    if ($service) {
                        if (isset($this->services[ $service ])) {
                            $this->response($socket, [
                                'status' => $this->services[ $service ]['status'],
                                'detail' => $this->services[ $service ]
                            ]);
                        } else {
                            $this->response($socket, ['status' => 'unknown', 'detail' => []]);
                        }
                    } else {
                        $this->response($socket, ['status' => 'running', 'services' => $this->services]);
                    }
            }
        } catch (\Exception $e) {
            $this->response($socket, ['error' => 500, 'msg' => $e->getMessage()]);
        }
    }
432048
protected function writeToSocket($socket, $message)
    {
        socket_sendto($socket, $message, strlen($message), 0, PAPERTRAIL_HOST, PAPERTRAIL_PORT);
    }
1036167
public function pack($msg)
    {
        if (!$msg instanceof Message) {
            trigger_error("Given parameter is not a message instance.",
                          E_USER_ERROR);
            return;
        }

        // Set value using serialzed message.
        $this->value = $msg->serializeToString();

        // Set type url.
        $pool = \Google\Protobuf\Internal\DescriptorPool::getGeneratedPool();
        $desc = $pool->getDescriptorByClassName(get_class($msg));
        $fully_qualifed_name = $desc->getFullName();
        $this->type_url = GPBUtil::TYPE_URL_PREFIX.substr(
            $fully_qualifed_name, 1, strlen($fully_qualifed_name));
    }
1250433
protected function sendTextMsgToJsonString(WxSendTextMsg $msg)
    {
        $formatStr = '{
                        "touser":"%s",
                        "msgtype":"%s",
                        "text":
                        {
                             "content":"%s"
                        }
                    }';
        $result = sprintf($formatStr,   $msg->getToUserName(),
                                        $msg->getMsgType(),
                                        $msg->getContent()
                        );


        return $result;
    }
369505
public static function checkMinLength($field, $minLength, $msg)
    {
        if ((StringUtils::isEmpty($field) === false) && mb_strlen($field) < $minLength) {
            throw new MsgException($msg);
        }
    }
412667
public function record($msg, $type = 'info', array $context = [])
    {
        if (is_string($msg)) {
            $replace = [];
            foreach ($context as $key => $val) {
                $replace['{' . $key . '}'] = $val;
            }

            $msg = strtr($msg, $replace);
        }

        $this->log[$type][] = $msg;

        if (PHP_SAPI == 'cli') {
            // 命令行日志实时写入
            $this->save();
        }

        return $this;
    }
1019228
protected function log($level, $msg, $context = [])
    {
        $this->log[$level][] = $msg;
        \Log::log($level, $msg, $context);
    }
1202359
private function readErrorFromSocket($socket) {
        $error_info = "";
        while (!empty($res = socket_read($socket, 8192))) {
            $error_info .= $res;
        }
        $error_info = unserialize($error_info);
        if (is_array($error_info)) {
            return sprintf('[%s] %s:%d %s', self::$ERROR_CODES[$error_info['type']], $error_info['file'], $error_info['line'], $error_info['message']);
        }

        return null;
    }
1095749
public static function t($msg, $args = array())
    {
        assert('is_string($msg)');
        if (extension_loaded('gettext')) {
            bindtextdomain("phptoolkit", dirname(dirname(__DIR__)).'/locale');
            textdomain('phptoolkit');

            $translatedMsg = gettext($msg);
        } else {
            $translatedMsg = $msg;
        }
        if (!empty($args)) {
            $params = array_merge(array($translatedMsg), $args);
            $translatedMsg = call_user_func_array('sprintf', $params);
        }
        return $translatedMsg;
    }
298622
private function response($socket, array $data) {
        $msg = json_encode($data, JSON_UNESCAPED_SLASHES) . "\r\n\r\n";
        @socket_write($socket, $msg, strlen($msg));
    }
795296
public function send($openid, $msg)
    {
        // 获取消息类型
        $msg_type = '';
        if (gettype($msg)=='string') {
            $msg_type = 'text_simple';
        } elseif (gettype($msg)=='array') {
            $msg_type = $msg['type'];
        }

        switch ($msg_type) {
            /**
             * 1.1 发送文本消息(简洁输入)
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', 'hello world!');
             * ```
             */
            case 'text_simple':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"text",'.
                        '"text":{'.
                            '"content":"%s"'.
                        '}}',
                        $openid,
                        $msg);
                break;

            /**
             * 1.2 发送文本消息
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', array(
             *  'type' => 'text',
             *  'content' => 'hello world!'
             * ));
             * ```
             */
            case 'text':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"text",'.
                        '"text":{'.
                            '"content":"%s"'.
                        '}%s}', 
                        $openid,
                        $msg['content'],
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 2 发送图片消息
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', array(
             *  'type' => 'image',
             *  'media_id' => 'Uq7OczuEGEyUu--dYjg7seTm-EJTa0Zj7UDP9zUGNkVpjcEHhl7tU2Mv8mFRiLKC'
             * ));
             * ```
             */
            case 'image':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"image",'.
                        '"image":{'.
                            '"media_id":"%s"'.
                        '}%s}', 
                        $openid,
                        $msg['media_id'],
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 3 发送语音消息
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', array(
             *  'type' => 'voice',
             *  'media_id' => 'rVT43tfDwjh4p1BV2gJ5D7Zl2BswChO5L_llmlphLaTPytcGcguBAEJ1qK4cg4r_'
             *  ));
             * ```
             */
            case 'voice':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"voice",'.
                        '"voice":{'.
                            '"media_id":"%s"'.
                        '}%s}', 
                        $openid,
                        $msg['media_id'],
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 4 发送视频消息
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', array(
             *  'type' => 'video',
             *  'media_id' => 'yV0l71NL0wtpRA8OMX0-dBRQsMVyt3fspPUzurIS3psi6eWOrb_WlEeO39jasoZ8',
             *  'thumb_media_id' => '7ct_DvuwZXIO9e9qbIf2ThkonUX_FzLAoqBrK-jzUboTYJX0ngOhbz6loS-wDvyZ',     // 可选(无效, 官方文档好像写错了)
             *  'title' => '视频消息的标题',           // 可选
             *  'description' => '视频消息的描述'      // 可选
             * ));
             * ```
             */         
            case 'video':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"video",'.
                        '"video":{'.
                            '"media_id":"%s",'.
                            '"thumb_media_id":"%s",'.
                            '"title":"%s",'.
                            '"description":"%s"'.
                        '}%s}', 
                        $openid,
                        $msg['media_id'],
                        $msg['thumb_media_id'],
                        isset($msg['title']) ? $msg['title'] : '',
                        isset($msg['description']) ? $msg['description'] : '',
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 5 发送音乐消息
             *
             * Examples:
             * ```
             * $api->send('ocNtAt_K8nRlAdmNEo_R0WVg_rRw', array(
             *  'type' => 'music',
             *  'title' => '音乐标题',                      //可选
             *  'description' => '音乐描述',                //可选
             *  'music_url' => 'http://me.diary8.com/data/music/2.mp3',     //可选
             *  'hqmusic_url' => 'http://me.diary8.com/data/music/2.mp3',   //可选
             *  'thumb_media_id' => 'O39wW0ZsXCb5VhFoCgibQs5PupFb6VZ2jH5A8gHUJCJz2Qmkrb7objoTue7bGTGQ',
             * ));
             * ```
             */         
            case 'music':
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"music",'.
                        '"music":{'.
                            '"title":"%s",'.
                            '"description":"%s",'.
                            '"musicurl":"%s",'.
                            '"hqmusicurl":"%s",'.
                            '"thumb_media_id":"%s"'.
                        '}%s}', 
                        $openid,
                        isset($msg['title']) ? $msg['title'] : '',
                        isset($msg['description']) ? $msg['description'] : '',
                        isset($msg['music_url']) ? $msg['music_url'] : '',
                        isset($msg['hqmusic_url']) ? $msg['hqmusic_url'] : '',
                        $msg['thumb_media_id'],
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 6 发送图文消息
             *
             * Examples:
             * ```
             * $api->send($msg->FromUserName, array(
             *  'type' => 'news',
             *  'articles' => array(
             *      array(
             *          'title' => '图文消息标题1',                           //可选
             *          'description' => '图文消息描述1',                     //可选
             *          'picurl' => 'http://me.diary8.com/data/img/demo1.jpg',  //可选
             *          'url' => 'http://www.example.com/'                      //可选
             *      ),
             *      array(
             *          'title' => '图文消息标题2',
             *          'description' => '图文消息描述2',
             *          'picurl' => 'http://me.diary8.com/data/img/demo2.jpg',
             *          'url' => 'http://www.example.com/'
             *      ),
             *      array(
             *          'title' => '图文消息标题3',
             *          'description' => '图文消息描述3',
             *          'picurl' => 'http://me.diary8.com/data/img/demo3.jpg',
             *          'url' => 'http://www.example.com/'
             *      )
             *  ),
             *  'kf_account' => 'test1@kftest'      // 可选(指定某个客服发送, 会显示这个客服的头像)
             * ));
             * ```
             */
            case 'news':
                $articles = array();
                foreach ($msg['articles'] as $article) {
                    array_push($articles, sprintf('{'.
                        '"title":"%s",'.
                        '"description":"%s",'.
                        '"url":"%s",'.
                        '"picurl":"%s"'.
                        '}',
                        $article['title'],
                        $article['description'],
                        $article['url'],
                        $article['picurl']));
                }
                $articles = implode(",", $articles);
                $xml = sprintf('{'.
                        '"touser":"%s",'.
                        '"msgtype":"news",'.
                        '"news":{"articles": [%s]}%s}',
                        $openid,
                        $articles,
                        isset($msg['kf_account']) ? ',"customservice":{"kf_account": "'.$msg['kf_account'].'"}' : '');
                break;

            /**
             * 0 异常消息处理
             *
             */ 
            default:
                return Error::code('ERR_MEG_TYPE');
                break;
        }

        $url = self::API_DOMAIN . 'cgi-bin/message/custom/send?access_token=' . $this->get_access_token();
        $res = HttpCurl::post($url, $xml, 'json');
        // 异常处理: 获取access_token网络错误
        if ($res === false) {
            return Error::code('ERR_GET');
        }
        // 判断是否调用成功
        if ($res->errcode == 0) {
            return array(null, true);
        } else {
            return array($res, null);
        }
    }
1348120
private static function listenOnBoundSocket(
        $socket,
        Socket\ServerListenContext $listenContext,
        Socket\ServerTlsContext $tlsContext = null
    ): Server {
        if ($tlsContext) {
            $context = \array_merge(
                $listenContext->toStreamContextArray(),
                $tlsContext->toStreamContextArray()
            );
        } else {
            $context = $listenContext->toStreamContextArray();
        }

        \socket_listen($socket, $context["socket"]["backlog"] ?? 0);

        /** @var resource $stream */
        /** @noinspection PhpVoidFunctionResultUsedInspection */
        $stream = \socket_export_stream($socket);
        \stream_context_set_option($stream, $context); // put eventual options like ssl back (per worker)

        return new Server($stream);
    }
700528
public function handleConnect($server, $socket)
    {
        $socket->conn = new NetworkConnection($socket);

        try
        {
            $this->component->handleConnect($socket->conn);

            $socket->on('data',  [ $this, 'handleData' ]);
            $socket->on('error', [ $this, 'handleError' ]);
            $socket->on('close', [ $this, 'handleDisconnect' ]);
        }
        catch (Error $ex)
        {
            $this->close($socket);
        }
        catch (Exception $ex)
        {
            $this->close($socket);
        }
    }
913700
protected function socket_receive($socket)
	{
		// initially read to the length of the header size, then
		// expand to read more
		$bytes_total = self::SOCKET_HEADER_SIZE;
		$bytes_read = 0;
		$have_header = false;
		$socket_message = '';
		while ($bytes_read < $bytes_total)
		{
			$read = @socket_read($socket, $bytes_total - $bytes_read);
			if ($read === false)
			{
				$this->log('socket_receive error: ' . socket_strerror(socket_last_error()), self::LOG_LEVEL_CRIT);
				return false;
			}

			// blank socket_read means done
			if ($read == '')
			{
				break;
			}

			$bytes_read += strlen($read);
			$socket_message .= $read;

			if (!$have_header && $bytes_read >= self::SOCKET_HEADER_SIZE)
			{
				$have_header = true;
				list($bytes_total) = array_values(unpack('N', $socket_message));
				$bytes_read = 0;
				$socket_message = '';
			}
		}

		$message = @unserialize($socket_message);

		return $message;
	}
1338153
public static function setTypeMsg($type, $msg=null)
    {
        if (is_array($type)) {
            self::$typeMsg=array_merge(self::$typeMsg, $type);
        } else {
            self::$typeMsg[$type]=$msg;
        }
    }
1325656
public function onSocketConnect($socket, Connection $connection): void
    {
        $this->checkConnections($connection);
        $this->checkConnectionsPerIp($connection);
    }
752522
public function sendToUser($user_id, array $msg) {
        //检查redis ping 防止掉线
        $fid = $this->userIdToFid($user_id);
        if (!$fid) {
            return null;
        }
        $fid_server = explode('@', $fid);
        $host_name = $fid_server[ 0 ];
        $fid = $fid_server[ 1 ];
        if ($host_name == $this->host_name) {
            if (!$this->server->exist($fid)) {
                //断开
                $this->server->close($fid);
                $redis = Cache::redis();
                $redis->hDel('user_id_#_fid', $user_id);
                $redis->hDel('fid_#_user_id', $fid);
                Cache::release();
                return false;
            }
            $this->server->push($fid, json_encode($msg));
        } else {
            try {
                //通过集群中的其他服务器推送
                Http::put('http://' . $host_name . ':' . $this->config[ 'port' ] . '/open/push', [], ['fid' => $fid,
                                                                                                      'msg' => json_encode($msg),
                                                                                                      'secret' => $this->config[ 'secret' ]]);
            } catch (\Exception $exception) {
            } catch (\Error $exception) {
            }
        }
        return null;
    }