book_admin/application/common/library/Ssdb.php

<?php
/**
 * Created by PhpStorm.
 * User: wangfanchang
 * Date: 18/1/10
 * Time: 下午2:13
 */

namespace app\common\library;

use think\Config;
use think\Log;

/**
 * Class Ssdb
 * @package app\common\library
 * @method bool hexists(string $name, string $key) 判断指定的 key 是否存在于 hashmap 中.
 * @method bool hset(string $name, string $key, string $value) 设置 hashmap 中指定 key 对应的值内容.
 * @method null|bool|string hget(string $name, string $key) 获取 hashmap 中指定 key 的值内容.
 * @method bool|array hkeys(string $name, string $key_start, string $key_end, int $limit) 列出 hashmap 中处于区间 (key_start, key_end] 的 key 列表.
 * @method bool hdel(string $name, string $key) 获取 hashmap 中的指定 key.
 * @method bool|array hscan(string $name, string $key_start, string $key_end, int $limit) 列出 hashmap 中处于区间 (key_start, key_end] 的 key-value 列表.
 * @method bool|array hgetall(string $name) 返回整个 hashmap.
 * @method bool|int hclear(string $name) 删除 hashmap 中的所有 key.
 * @method bool|int hsize(string $name) 返回 hashmap 中的元素个数.
 * @method bool|null|string get(string $name) 获取指定 key 的值内容.
 * @method bool exists(string $name) 判断指定的 key 是否存在.
 * @method bool|int incr(string $key, int $num = 1) 使 key 对应的值增加 num. 参数 num 可以为负数. 如果原来的值不是整数(字符串形式的整数), 它会被先转换成整数.
 * @method bool|int hincr(string $name, string $key, int $num = 1) 使 hashmap 中的 key 对应的值增加 num. 参数 num 可以为负数. 如果原来的值不是整数(字符串形式的整数), 它会被先转换成整数.
 * @method bool setx(string $key, string $value, int $ttl) 设置指定 key 的值内容, 同时设置存活时间.
 * @method bool|array scan(string $key_start, string $key_end, int $limit) 列出处于区间 (key_start, key_end] 的 key-value 列表.
 * @method bool|array hlist(string $name_start, string $name_end, int $limit) 列出名字处于区间 (name_start, name_end] 的 hashmap.
 * @method bool del(string $key) 删除指定的key
 * @method bool multi_del(array $keys) 批量删除一批 key 和其对应的值内容.
 * @method bool multi_hdel(string $name, array $keys) 批量删除 hashmap 中的 key.
 * @method mixed dbsize($first = [], $second = [])
 * @method mixed ping($first = [], $second = [])
 * @method mixed qset($first = [], $second = [])
 * @method mixed getbit($first = [], $second = [])
 * @method mixed setbit($first = [], $second = [])
 * @method mixed countbit($first = [], $second = [])
 * @method mixed strlen($first = [], $second = [])
 * @method mixed set($first = [], $second = [])
 * @method mixed setnx($first = [], $second = [])
 * @method mixed zset($first = [], $second = [])
 * @method mixed qpush($first = [], $second = [])
 * @method mixed qpush_front($first = [], $second = [])
 * @method mixed qpush_back($first = [], $second = [])
 * @method mixed qtrim_front($first = [], $second = [])
 * @method mixed qtrim_back($first = [], $second = [])
 * @method mixed zdel($first = [], $second = [])
 * @method mixed zsize($first = [], $second = [])
 * @method mixed qsize($first = [], $second = [])
 * @method mixed zclear($first = [], $second = [])
 * @method mixed qclear($first = [], $second = [])
 * @method mixed multi_set($first = [], $second = [])
 * @method mixed multi_hset($first = [], $second = [])
 * @method mixed multi_zset($first = [], $second = [])
 * @method mixed multi_zdel($first = [], $second = [])
 * @method mixed decr($first = [], $second = [])
 * @method mixed zincr($first = [], $second = [])
 * @method mixed zdecr($first = [], $second = [])
 * @method mixed hdecr($first = [], $second = [])
 * @method mixed zget($first = [], $second = [])
 * @method mixed zrank($first = [], $second = [])
 * @method mixed zrrank($first = [], $second = [])
 * @method mixed zcount($first = [], $second = [])
 * @method mixed zsum($first = [], $second = [])
 * @method mixed zremrangebyrank($first = [], $second = [])
 * @method mixed zremrangebyscore($first = [], $second = [])
 * @method mixed ttl($first = [], $second = [])
 * @method mixed expire($first = [], $second = [])
 * @method mixed zavg($first = [], $second = [])
 * @method mixed substr($first = [], $second = [])
 * @method mixed getset($first = [], $second = [])
 * @method mixed qget($first = [], $second = [])
 * @method mixed qfront($first = [], $second = [])
 * @method mixed qback($first = [], $second = [])
 * @method mixed qpop($first = [], $second = [])
 * @method mixed qpop_front($first = [], $second = [])
 * @method mixed qpop_back($first = [], $second = [])
 * @method mixed keys($first = [], $second = [])
 * @method mixed zkeys($first = [], $second = [])
 * @method mixed zlist($first = [], $second = [])
 * @method mixed qslice($first = [], $second = [])
 * @method mixed auth($first = [], $second = [])
 * @method mixed zexists($first = [], $second = [])
 * @method mixed multi_exists($first = [], $second = [])
 * @method mixed multi_hexists($first = [], $second = [])
 * @method mixed multi_zexists($first = [], $second = [])
 * @method mixed rscan($first = [], $second = [])
 * @method mixed zscan($first = [], $second = [])
 * @method mixed zrscan($first = [], $second = [])
 * @method mixed zrange($first = [], $second = [])
 * @method mixed zrrange($first = [], $second = [])
 * @method mixed hrscan($first = [], $second = [])
 * @method mixed multi_hsize($first = [], $second = [])
 * @method mixed multi_zsize($first = [], $second = [])
 * @method mixed multi_get($first = [], $second = [])
 * @method mixed multi_hget($first = [], $second = [])
 * @method mixed multi_zget($first = [], $second = [])
 * @method mixed zpop_front($first = [], $second = [])
 * @method mixed zpop_back($first = [], $second = [])
 */
class Ssdb
{

    /**
     * 单例对象
     */
    protected static $instance;

    /**
     * @var \SimpleSSDB
     */
    protected $handler;

    /**
     * SSDB 连接池
     * @var array
     */
    protected static $connections = [];

    /**
     * ssdb 区分链接的规则
     * @var array
     */
    protected static $rules = null;

    /**
     * 初始化
     *
     * @access public
     * @return Ssdb
     */
    public static function instance()
    {
        if (is_null(self::$instance)) {
            self::$instance = new static();
        }
        return self::$instance;
    }

    /**
     * @param string $key
     * @return SimpleSSDB
     */
    private static function getSsdb($key = '')
    {
        if (is_null(self::$rules)) {
            $rules = Config::get('ssdb.rules');
            $list = explode(';', $rules);
            foreach ($list as $item) {
                $con = explode('=', $item); // 0 编号  1 配置
                if (count($con) >= 2) {
                    $c = explode(',', $con[1]); //规则
                    foreach ($c as $it) {
                        if ($it) {
                            self::$rules[$con[0]][] = $it;
                        }
                    }
                }
            }
        }
        if ($key && count(self::$rules)) {
            foreach (self::$rules as $idx => $item) {
                foreach ($item as $v) {
                    if (strpos($key, $v) === 0) { //查找到，且必须为第一个0
                        if (isset(self::$connections[$idx])) {
                            return self::$connections[$idx];
                        }
                    }
                }
            }
        }
        $default = Config::get('ssdb.default');
        if (isset(self::$connections[$default])) {
            return self::$connections[$default];
        } else {
            return self::$connections[0];
        }
    }

    /**
     * 构造函数
     *
     * @param array $options
     */
    public function __construct()
    {
        $config = Config::get('ssdb');

        $list = explode(';', $config['list']);
        foreach ($list as $item) {
            $conGroup = explode('=', $item);
            if (count($conGroup) >= 2) {
                $conGroupList = explode(':', $conGroup[1]);
                if (count($conGroupList) % 3 > 0) {
                    Log::error('SSDB config error!冒号分隔的属性个数不是3的倍数');
                    break;
                }
                for ($i = 0; $i < count($conGroupList); $i += 3) {
                    $host = $conGroupList[$i];
                    $port = $conGroupList[$i + 1];
                    $pwd = $conGroupList[$i + 2];
                    Log::info(sprintf('SSDB连接,host:%s,port:%s,pwd:%s'
                        , $host, $port, $pwd));
                    try {
                        self::$connections[$conGroup[0]] = new SimpleSSDB($host, $port);
                        self::$connections[$conGroup[0]]->easy();
                        self::$connections[$conGroup[0]]->auth($pwd);
                        break;
                    } catch (\Exception $e) {
                        Log::error(sprintf('SSDB连接失败,host:%s,port:%s,pwd:%s'
                            , $host, $port, $pwd));
                        continue;
                    }
                }

                if (empty(self::$connections[$conGroup[0]])) {
                    $errMsg = 'SSDB config error! num:' . $conGroup[0];
                    Log::error($errMsg);
                    throw new \Exception($errMsg);
                }
            }
        }

        if (count(self::$connections) == 0) {
            throw new \Exception('SSDB config error!');
        }
    }

    function __call($name, $arguments)
    {
        if (count($arguments)) {
            $ssdb = self::getSsdb($arguments[0]);
        } else {
            $ssdb = self::getSsdb();
        }
        return call_user_func_array([$ssdb, $name], $arguments);
    }
}



class SSDBException extends \Exception
{
}

class SSDBTimeoutException extends SSDBException
{
}

class SSDB_Response
{
    public $cmd;
    public $code;
    public $data = null;
    public $message;

    function __construct($code='ok', $data_or_message=null){
        $this->code = $code;
        if($code == 'ok'){
            $this->data = $data_or_message;
        }else{
            $this->message = $data_or_message;
        }
    }

    function __toString(){
        if($this->code == 'ok'){
            $s = $this->data === null? '' : json_encode($this->data);
        }else{
            $s = $this->message;
        }
        return sprintf('%-13s %12s %s', $this->cmd, $this->code, $s);
    }

    function ok(){
        return $this->code == 'ok';
    }

    function not_found(){
        return $this->code == 'not_found';
    }
}

class SimpleSSDB
{
    private $debug = false;
    public $sock = null;
    private $_closed = false;
    private $recv_buf = '';
    private $_easy = false;
    public $last_resp = null;

    function __construct($host, $port, $timeout_ms=2000){
        $timeout_f = (float)$timeout_ms/1000;
        $this->sock = @stream_socket_client("$host:$port", $errno, $errstr, $timeout_f);
        if(!$this->sock){
            throw new SSDBException("$errno: $errstr");
        }
        $timeout_sec = intval($timeout_ms/1000);
        $timeout_usec = ($timeout_ms - $timeout_sec * 1000) * 1000;
        @stream_set_timeout($this->sock, $timeout_sec, $timeout_usec);
        if(function_exists('stream_set_chunk_size')){
            @stream_set_chunk_size($this->sock, 1024 * 1024);
        }
    }

    function set_timeout($timeout_ms){
        $timeout_sec = intval($timeout_ms/1000);
        $timeout_usec = ($timeout_ms - $timeout_sec * 1000) * 1000;
        @stream_set_timeout($this->sock, $timeout_sec, $timeout_usec);
    }

    /**
     * After this method invoked with yesno=true, all requesting methods
     * will not return a SSDB_Response object.
     * And some certain methods like get/zget will return false
     * when response is not ok(not_found, etc)
     */
    function easy(){
        $this->_easy = true;
    }

    function close(){
        if(!$this->_closed){
            @fclose($this->sock);
            $this->_closed = true;
            $this->sock = null;
        }
    }

    function closed(){
        return $this->_closed;
    }

    private $batch_mode = false;
    private $batch_cmds = array();

    function batch(){
        $this->batch_mode = true;
        $this->batch_cmds = array();
        return $this;
    }

    function multi(){
        return $this->batch();
    }

    function exec(){
        $ret = array();
        foreach($this->batch_cmds as $op){
            list($cmd, $params) = $op;
            $this->send_req($cmd, $params);
        }
        foreach($this->batch_cmds as $op){
            list($cmd, $params) = $op;
            $resp = $this->recv_resp($cmd, $params);
            $resp = $this->check_easy_resp($cmd, $resp);
            $ret[] = $resp;
        }
        $this->batch_mode = false;
        $this->batch_cmds = array();
        return $ret;
    }

    function request(){
        $args = func_get_args();
        $cmd = array_shift($args);
        return $this->__call($cmd, $args);
    }

    private $async_auth_password = null;

    function auth($password){
        $this->async_auth_password = $password;
        return null;
    }

    function __call($cmd, $params=array()){
        $cmd = strtolower($cmd);
        if($this->async_auth_password !== null){
            $pass = $this->async_auth_password;
            $this->async_auth_password = null;
            $auth = $this->__call('auth', array($pass));
            if($auth !== true){
                throw new Exception("Authentication failed");
            }
        }

        if($this->batch_mode){
            $this->batch_cmds[] = array($cmd, $params);
            return $this;
        }

        try{
            if($this->send_req($cmd, $params) === false){
                $resp = new SSDB_Response('error', 'send error');
            }else{
                $resp = $this->recv_resp($cmd, $params);
            }
        }catch(SSDBException $e){
            if($this->_easy){
                throw $e;
            }else{
                $resp = new SSDB_Response('error', $e->getMessage());
            }
        }

        if($resp->code == 'noauth'){
            $msg = $resp->message;
            throw new Exception($msg);
        }

        $resp = $this->check_easy_resp($cmd, $resp);
        return $resp;
    }

    private function check_easy_resp($cmd, $resp){
        $this->last_resp = $resp;
        if($this->_easy){
            if($resp->not_found()){
                return NULL;
            }else if(!$resp->ok() && !is_array($resp->data)){
                return false;
            }else{
                return $resp->data;
            }
        }else{
            $resp->cmd = $cmd;
            return $resp;
        }
    }

    function multi_set($kvs=array()){
        $args = array();
        foreach($kvs as $k=>$v){
            $args[] = $k;
            $args[] = $v;
        }
        return $this->__call(__FUNCTION__, $args);
    }

    function multi_hset($name, $kvs=array()){
        $args = array($name);
        foreach($kvs as $k=>$v){
            $args[] = $k;
            $args[] = $v;
        }
        return $this->__call(__FUNCTION__, $args);
    }

    function multi_zset($name, $kvs=array()){
        $args = array($name);
        foreach($kvs as $k=>$v){
            $args[] = $k;
            $args[] = $v;
        }
        return $this->__call(__FUNCTION__, $args);
    }

    function incr($key, $val=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    function decr($key, $val=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    function zincr($name, $key, $score=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    function zdecr($name, $key, $score=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    function zadd($key, $score, $value){
        $args = array($key, $value, $score);
        return $this->__call('zset', $args);
    }

    function zRevRank($name, $key){
        $args = func_get_args();
        return $this->__call("zrrank", $args);
    }

    function zRevRange($name, $offset, $limit){
        $args = func_get_args();
        return $this->__call("zrrange", $args);
    }

    function hincr($name, $key, $val=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    function hdecr($name, $key, $val=1){
        $args = func_get_args();
        return $this->__call(__FUNCTION__, $args);
    }

    private function send_req($cmd, $params){
        $req = array($cmd);
        foreach($params as $p){
            if(is_array($p)){
                $req = array_merge($req, $p);
            }else{
                $req[] = $p;
            }
        }
        return $this->send($req);
    }

    private function recv_resp($cmd, $params){
        $resp = $this->recv();
        if($resp === false){
            return new SSDB_Response('error', 'Unknown error');
        }else if(!$resp){
            return new SSDB_Response('disconnected', 'Connection closed');
        }
        if($resp[0] == 'noauth'){
            $errmsg = isset($resp[1])? $resp[1] : '';
            return new SSDB_Response($resp[0], $errmsg);
        }
        switch($cmd){
            case 'dbsize':
            case 'ping':
            case 'qset':
            case 'getbit':
            case 'setbit':
            case 'countbit':
            case 'strlen':
            case 'set':
            case 'setx':
            case 'setnx':
            case 'zset':
            case 'hset':
            case 'qpush':
            case 'qpush_front':
            case 'qpush_back':
            case 'qtrim_front':
            case 'qtrim_back':
            case 'del':
            case 'zdel':
            case 'hdel':
            case 'hsize':
            case 'zsize':
            case 'qsize':
            case 'hclear':
            case 'zclear':
            case 'qclear':
            case 'multi_set':
            case 'multi_del':
            case 'multi_hset':
            case 'multi_hdel':
            case 'multi_zset':
            case 'multi_zdel':
            case 'incr':
            case 'decr':
            case 'zincr':
            case 'zdecr':
            case 'hincr':
            case 'hdecr':
            case 'zget':
            case 'zrank':
            case 'zrrank':
            case 'zcount':
            case 'zsum':
            case 'zremrangebyrank':
            case 'zremrangebyscore':
            case 'ttl':
            case 'expire':
                if($resp[0] == 'ok'){
                    $val = isset($resp[1])? intval($resp[1]) : 0;
                    return new SSDB_Response($resp[0], $val);
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
            case 'zavg':
                if($resp[0] == 'ok'){
                    $val = isset($resp[1])? floatval($resp[1]) : (float)0;
                    return new SSDB_Response($resp[0], $val);
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
            case 'get':
            case 'substr':
            case 'getset':
            case 'hget':
            case 'qget':
            case 'qfront':
            case 'qback':
                if($resp[0] == 'ok'){
                    if(count($resp) == 2){
                        return new SSDB_Response('ok', $resp[1]);
                    }else{
                        return new SSDB_Response('server_error', 'Invalid response');
                    }
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
                break;
            case 'qpop':
            case 'qpop_front':
            case 'qpop_back':
                if($resp[0] == 'ok'){
                    $size = 1;
                    if(isset($params[1])){
                        $size = intval($params[1]);
                    }
                    if($size <= 1){
                        if(count($resp) == 2){
                            return new SSDB_Response('ok', $resp[1]);
                        }else{
                            return new SSDB_Response('server_error', 'Invalid response');
                        }
                    }else{
                        $data = array_slice($resp, 1);
                        return new SSDB_Response('ok', $data);
                    }
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
                break;
            case 'keys':
            case 'zkeys':
            case 'hkeys':
            case 'hlist':
            case 'zlist':
            case 'qslice':
                if($resp[0] == 'ok'){
                    $data = array();
                    if($resp[0] == 'ok'){
                        $data = array_slice($resp, 1);
                    }
                    return new SSDB_Response($resp[0], $data);
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
            case 'auth':
            case 'exists':
            case 'hexists':
            case 'zexists':
                if($resp[0] == 'ok'){
                    if(count($resp) == 2){
                        return new SSDB_Response('ok', (bool)$resp[1]);
                    }else{
                        return new SSDB_Response('server_error', 'Invalid response');
                    }
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
                break;
            case 'multi_exists':
            case 'multi_hexists':
            case 'multi_zexists':
                if($resp[0] == 'ok'){
                    if(count($resp) % 2 == 1){
                        $data = array();
                        for($i=1; $i<count($resp); $i+=2){
                            $data[$resp[$i]] = (bool)$resp[$i + 1];
                        }
                        return new SSDB_Response('ok', $data);
                    }else{
                        return new SSDB_Response('server_error', 'Invalid response');
                    }
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
                break;
            case 'scan':
            case 'rscan':
            case 'zscan':
            case 'zrscan':
            case 'zrange':
            case 'zrrange':
            case 'hscan':
            case 'hrscan':
            case 'hgetall':
            case 'multi_hsize':
            case 'multi_zsize':
            case 'multi_get':
            case 'multi_hget':
            case 'multi_zget':
            case 'zpop_front':
            case 'zpop_back':
                if($resp[0] == 'ok'){
                    if(count($resp) % 2 == 1){
                        $data = array();
                        for($i=1; $i<count($resp); $i+=2){
                            if($cmd[0] == 'z'){
                                $data[$resp[$i]] = intval($resp[$i + 1]);
                            }else{
                                $data[$resp[$i]] = $resp[$i + 1];
                            }
                        }
                        return new SSDB_Response('ok', $data);
                    }else{
                        return new SSDB_Response('server_error', 'Invalid response');
                    }
                }else{
                    $errmsg = isset($resp[1])? $resp[1] : '';
                    return new SSDB_Response($resp[0], $errmsg);
                }
                break;
            default:
                return new SSDB_Response($resp[0], array_slice($resp, 1));
        }
        return new SSDB_Response('error', 'Unknown command: $cmd');
    }

    function send($data){
        $ps = array();
        foreach($data as $p){
            $ps[] = strlen($p);
            $ps[] = $p;
        }
        $s = join("\n", $ps) . "\n\n";
        if($this->debug){
            echo '> ' . str_replace(array("\r", "\n"), array('\r', '\n'), $s) . "\n";
        }
        try{
            while(true){
                $ret = @fwrite($this->sock, $s);
                if($ret === false || $ret === 0){
                    $this->close();
                    throw new SSDBException('Connection lost');
                }
                $s = substr($s, $ret);
                if(strlen($s) == 0){
                    break;
                }
                @fflush($this->sock);
            }
        }catch(Exception $e){
            $this->close();
            throw new SSDBException($e->getMessage());
        }
        return $ret;
    }

    function recv(){
        $this->step = self::STEP_SIZE;
        while(true){
            $ret = $this->parse();
            if($ret === null){
                try{
                    $data = @fread($this->sock, 1024 * 1024);
                    if($this->debug){
                        echo '< ' . str_replace(array("\r", "\n"), array('\r', '\n'), $data) . "\n";
                    }
                }catch(Exception $e){
                    $data = '';
                }
                if($data === false || $data === ''){
                    if(feof($this->sock)){
                        $this->close();
                        throw new SSDBException('Connection lost');
                    }else{
                        throw new SSDBTimeoutException('Connection timeout');
                    }
                }
                $this->recv_buf .= $data;
#				echo "read " . strlen($data) . " total: " . strlen($this->recv_buf) . "\n";
            }else{
                return $ret;
            }
        }
    }

    const STEP_SIZE = 0;
    const STEP_DATA = 1;
    public $resp = array();
    public $step;
    public $block_size;

    private function parse(){
        $spos = 0;
        $epos = 0;
        $buf_size = strlen($this->recv_buf);
        // performance issue for large reponse
        //$this->recv_buf = ltrim($this->recv_buf);
        while(true){
            $spos = $epos;
            if($this->step === self::STEP_SIZE){
                $epos = strpos($this->recv_buf, "\n", $spos);
                if($epos === false){
                    break;
                }
                $epos += 1;
                $line = substr($this->recv_buf, $spos, $epos - $spos);
                $spos = $epos;

                $line = trim($line);
                if(strlen($line) == 0){ // head end
                    $this->recv_buf = substr($this->recv_buf, $spos);
                    $ret = $this->resp;
                    $this->resp = array();
                    return $ret;
                }
                $this->block_size = intval($line);
                $this->step = self::STEP_DATA;
            }
            if($this->step === self::STEP_DATA){
                $epos = $spos + $this->block_size;
                if($epos <= $buf_size){
                    $n = strpos($this->recv_buf, "\n", $epos);
                    if($n !== false){
                        $data = substr($this->recv_buf, $spos, $epos - $spos);
                        $this->resp[] = $data;
                        $epos = $n + 1;
                        $this->step = self::STEP_SIZE;
                        continue;
                    }
                }
                break;
            }
        }

        // packet not ready
        if($spos > 0){
            $this->recv_buf = substr($this->recv_buf, $spos);
        }
        return null;
    }
}