自定义文本协议，服务端丢消息

poisonbian 更新于2019-10-30

自定义的文本格式协议，格式很简单，用\r\n\r\n来分隔数据。代码见最下方。

现在的情况是这样子的，我通过tcp客户端发起一次请求，传输5个文本包，服务端的日志来看，这5个包确实都接收到了（protocol的input中的日志被打印出来了），但是并没有触发onMessage的操作。

进一步定位，protocol中的decode函数压根都没有被调用，下面输出的内容中可以看到，5个包都正常收到了，解析length也正确，就是没有触发decode操作。。。

++++++++++++++++++++++++++++ 输出 +++++++++++++++++++++++++++

----------------------- WORKERMAN -----------------------------
Workerman version:3.3.1          PHP version:5.5.25
------------------------ WORKERS -------------------------------
user          worker            listen                  processes status
root          workerman_logger  orzText://0.0.0.0:8300   4          
----------------------------------------------------------------
Press Ctrl-C to quit. Start success.

string(11) "length: 223"
string(9) "decode..."
string(12) "onMessage..."
string(11) "length: 223"
string(9) "decode..."
string(12) "onMessage..."
string(11) "length: 223"
string(11) "length: 223"
string(11) "length: 223"

++++++++++++++++++++++++++++ 代码 +++++++++++++++++++++++++++

<?php
namespace Workerman\Protocols;

use Workerman\Connection\TcpConnection;

class OrzText
{
    public static $SEPARATOR = "\r\n\r\n";
    public static $SEPARATOR_LENGTH = 4;

    public static function input($buffer, TcpConnection $connection)
    {
        if (strlen($buffer) >= TcpConnection::$maxPackageSize) {
            $connection->close();
            return 0;
        }
        $pos = strpos($buffer, self::$SEPARATOR);
        if ($pos === false) {
            return 0;
        }
        $length = $pos + self::$SEPARATOR_LENGTH;
        var_dump("length: " . $length);
        return $length;
    }

    public static function encode($buffer)
    {
        return $buffer . self::$SEPARATOR;
    }

    public static function decode($buffer)
    {
        var_dump("decode...");
        return trim($buffer);
    }
}

 4188  4 0

4个回答

walkor 2016-05-27

打赏

贴完整代码。

暂无评论

poisonbian 2016-05-27

服务端代码

require_once(dirname(__FILE__) . '/CommandBaseManager.class.php');
Vendor('Workerman.Autoloader', '', '.php');
Vendor('GatewayWorker.Lib.Db', '', '.php');

use \Workerman\Worker;
use \GatewayWorker\Lib\Db;

abstract class WorkermanBaseManager extends CommandBaseManager
{
    protected static $config = null;
    protected static $dbs = array();
    protected static $db = null;

    public function _initialize()
    {
        parent::_initialize();

        Db::set_config(0, array(
            'host'  => C('DB_HOST'),
            'port'  => C('DB_PORT'),
            'user'  => C('DB_USER'),
            'password'  => C('DB_PWD'),
            'dbname'    => C('DB_NAME'),
        ));
        self::$dbs = Db::instance(0);
        self::$db = self::$dbs;
    }

    public function onWorkerStart($task)
    {
        $this->_logger()->debug('name: ' . $task->name);
        return true;
    }

    public function onConnect($connection)
    {
        $this->_logger()->debug('from ip ' . $connection->getRemoteIp());
        return true;
    }

    public function onMessage($connection, $data)
    {
        static $i = 0;
        var_dump("onMessage..." . (++$i));
    }

    public function onClose($connection)
    {
        $this->_logger()->debug('from ip ' . $connection->getRemoteIp());
        return true;
    }

    protected function _work()
    {
        $config = array(
            'listen'        => 'OrzText://0.0.0.0:8300',
            'name'          => 'workerman_logger',
            'count'         => 4,
            'logFile'       => 'Log/Command/logger_run.log',
        );

        $worker = new Worker(self::$config);
        Worker::$logFile = REAL_ROOT_PATH . self::$config;
        $worker->count = self::$config;
        $worker->name = self::$config;

        $worker->onMessage = array($this, 'onMessage');
        $worker->onConnect = array($this, 'onConnect');
        $worker->onClose = array($this, 'onClose');
        $worker->onWorkerStart = array($this, 'onWorkerStart');
        $worker->onWorkStop = array($this, 'onWorkerStop');
        $worker->onBufferFull = array($this, 'onBufferFull');
        $worker->onBufferDrain = array($this, 'onBufferDrain');
        $worker->onError = array($this, 'onError');

        Worker::runAll();
    }

    // 一般不需要做修改的一些回调
    public function onBufferFull($connection)
    {
        $this->_logger()->warn('do not send again');
    }
    public function onBufferDrain($connection)
    {
        $this->_logger()->warn('send again');
    }
    public function onWorkerStop($worker)
    {
        $log = sprintf('worker stop: %s', $worker->id, $worker->name);
        $this->_logger()->debug($log);
    }

    public function onError($connection, $code, $msg)
    {
        $log = sprintf('from ip %s, error: ', $connection->getRemoteIp(), $code, $msg);
        $this->_logger()->error($log);
    }
}

客户端调用的代码

<?php 
$host = 'localhost';
$port = 8300;
$timeout = 1;
$eof = "\r\n\r\n";

$socket = fsockopen($host, $port, $errno, $errstr, $timeout);

for ($i = 0; $i < 20; $i++)
{
    fwrite($socket, 'testing' . $eof);
}

fclose($socket);
echo "done\n";