station-test-api/task_script/PUMP_MQTT_PUBLISH.php

<?php
require('../vendor/autoload.php');
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
use think\facade\Cache;
// date_default_timezone_set("America/Bahia");
// define('HOST', '127.0.0.1');
// define('PORT', '6379');
// define('PASSWORD', '123456');
// define('DATABASE', 2);
define('HOST', 'r-bp1eebab79320044pd.redis.rds.aliyuncs.com');
define('PORT', '6379');
define('PASSWORD', '7e2b5c91e438be3c!');
define('DATABASE', 4);

function app_redis()
{
    static $redis = null;
    static $conn = false;
    if (!$conn) {
        connect: //定义标签
        $redis = new Redis();
        try {
            //建立的Redis短连接,在请求结束后不会自动关闭,相当于持久连接.
            $conn = $redis->connect(HOST, PORT);
            $conn = $redis->auth(PASSWORD);
            $conn = $redis->select(DATABASE);
            // 连接成功,返回$redis对象,连接失败,返回false.
            return ($conn === true) ? $redis : false;
        } catch (Exception $e) {
            return false;
        }
    } else {
        // 这里假设PHP-FPM在处理一个请求的时间内,Redis连接都是可用的.
        // 所以只在PHP-CLI下检查Redis连接的状态,进行断线重连.
        if (php_sapi_name() === 'cli') {
            try {
                // ping用于检查当前连接的状态,成功时返回+PONG,失败时抛出一个RedisException对象.
                // ping失败时警告:
                // Warning: Redis::ping(): connect() failed: Connection refused
                // var_dump('AAAAAAAAA', $redis);
                // echo 'Redis 连接状态' . $redis->ping() . PHP_EOL;
                @$redis->ping();
                if (!$redis->ping()) {
                    goto connect; //跳转到标签出继续执行连接操作
                }
            } catch (Exception  $e) {
                // 信息如 Connection lost 或 Redis server went away
                echo $e->getMessage();
                echo 'Redis 连接失败 重新连接：' . PHP_EOL;
                // 断线重连
                goto connect;
            }
        }
        return $redis;
    }
}


function sendConfig($topic,$config)
{
    $server   = '116.62.220.88';
    $port     = 1883;
    $clientId = 'local_mqtt_pump_config';
    $username = 'rl517';
    $password = "rlian2022";
    $clean_session = false;

    $connectionSettings  = new ConnectionSettings();
    $connectionSettings = $connectionSettings
        ->setUsername($username)
        ->setPassword($password)
        ->setKeepAliveInterval(60)
        // Last Will 设置
        //        ->setLastWillTopic('emqx/test/last-will')
        //        ->setLastWillMessage('client disconnect')
        //        ->setLastWillQualityOfService(1)
    ;

    $mqtt = new MqttClient($server, $port, $clientId);

    $mqtt->connect($connectionSettings, $clean_session);
    echo 'connect OK'.PHP_EOL;
    echo 'topic:'.$topic.PHP_EOL;
    echo 'config:'.$config.PHP_EOL;
    $res=$mqtt->publish(
        $topic,
        $config,
        1
    );
    echo 'publish end'.PHP_EOL;
    $mqtt->loop(true,true);
    $mqtt->disconnect();
    return $res;
}
function updateSendResult($msgid){
    $conn = new mysqli('rm-bp1h3uqkzy66ckt8y125010.mysql.rds.aliyuncs.com', 'dev', '711e7D69f9d0c3f1', 'catch_pump');
    // $conn = new mysqli('127.0.0.1', 'root', 'root', 'catch_pump');
    if ($conn -> connect_errno) {
        printf("Connect failed: %s\n", $conn->connect_error);
        exit();
    }
    $sql = "UPDATE send_config_log SET result='1',updated_at=".time()." WHERE id=".$msgid;
    if ($conn->query($sql) === TRUE) {
        echo "send_config_log update success";
    } else {
        echo "更新失败: " . $conn->error;
    }

}

function rlog(...$args)
{
    //函数参数是可变参数 使用前查下static开始的大写变量配置是否正确
    //rlog('info', [1,2,3], $a = null, '[]', $this, false);
    $args = func_get_args();
//        if (empty($args[0])) {
//            return;
//        }
    static $LOG_CONSOLE = true;//是否输出到控制台 fpm需要为false cli可以为true

    if (!empty(getenv('PM2_HOME'))) { //pm2
        $LOG_CONSOLE = false;
    } else if (PHP_SAPI == 'fpm-fcgi') { //php-fpm
        $LOG_CONSOLE = false;
    } else if (PHP_SAPI == 'cgi')  { //cli
        $LOG_CONSOLE = true;
    } else if (PHP_SAPI == 'cli-server')  { //php think run
        $LOG_CONSOLE = false;
    } else {
        $LOG_CONSOLE = true;
    }

    static $LOG_NAME = "./pubsub_tcp_iiot.log";//值为空时 不写入文件 第一个参数是imlog时也可以更改
    if (stripos(PHP_OS, 'WIN') !== false) {
        $LOG_NAME = "C:\Windows\Temp\php.log";
    }
    static $LOG_SIZE = 128 * 1024 * 1024;//文件最大尺寸 大于这个尺寸时 会生成个后缀.old的文件

    static $LOG_CACHE = false;//是否缓存日志内容 用于批量写入文件 如需强制刷新第一个参数传sync
    static $CACHE_DURATION = 10;//缓存最大时间 秒
    static $CACHE_SIZE = 1024;//缓存大小
    static $cacheStartTime = 0;
    static $cacheBuf = '';

    static $LOG_TIMES = 10;//调用这个函数最大次数 超过次数或$logCount==1 会判断下文件大小 决定是否新生成文件
    static $logCount = 0;

    static $MAX_LEN = 1024 * 4;//数据不能超过这个  不然就截断了

    $sync = false;//如果是true 使用$LOG_CACHE时会把数据保存到磁盘

    $implicit0 = ['imsync', 'imtrace', 'imlog'];//这个函数的参数0 隐藏用法

    $buf = '';
    if (count($args) == 1 && $args[0] === "\n") {//只有换行时 不写入时间戳了
        $buf = "\n";
    } else {
        $sync = is_string($args[0]) && strtolower($args[0]) == $implicit0[0];

        if (is_string($args[0]) && strpos($args[0], $implicit0[2]) === 0) {//设置日志名
            $LOG_NAME = str_replace($implicit0[2], '', $args[0]);
//                unset($args[0]);//记入日志也可以
        } else {
//            if ($LOG_NAME == 'php.log') {//每个文件会生成不同的文件名
//                $LOG_NAME = str_replace(".php", '', basename(__FILE__)) . '.log';
//            }
        }

        $pid = '';//进程id
        if (function_exists('posix_getpid')) {
            $pid = ' ' . posix_getpid() . ' ';
        }
        $fileLine = '';//文件名:行号
        {
            $debug = debug_backtrace();
            $backtrace = 0;
            if (is_string($args[0]) && strpos($args[0], $implicit0[1]) === 0) {
                $backtrace = intval(str_replace($implicit0[1], '', $args[0]));
                unset($args[0]);
            } else if ($sync) {
                unset($args[0]);
            }
            $fileLine = ($pid == '' ? ' ' : '') . str_replace(".php", '', basename($debug[$backtrace]['file']))
                . ':' . $debug[$backtrace]['line'] . '';

        }

        $allPara = '';
        foreach ($args as $para) {
            if (is_array($para)) {
                $allPara .= json_encode($para) . ' ';
            } else if (is_object($para)) {
                if (method_exists($para, '__toString')) {
                    $allPara .= $para . ' ';
                } else {
                    $allPara .= get_class($para) . json_encode($para) . ' ';
                }
            } else if (is_bool($para)) {
                $allPara .= $para ? 'true ' : 'false ';
            } else if (is_null($para)) {
                $allPara .= 'null ';
            } else {
                $allPara .= $para . ' ';
            }
        }

        $len = strlen($allPara);
        if ($len > $MAX_LEN) {
            $allPara = substr($allPara, 0, $MAX_LEN) . "({$len})......";
        }

        $buf = "[" . date("y-m-d H:i:s.") . substr(microtime(FALSE), 2, 3) . "{$pid}{$fileLine}]" . $allPara . "\n";
    }

    $logCount++;
    if (!empty($LOG_NAME)) {
        if ($LOG_CACHE) {
            if ($cacheBuf == '') {
                $cacheStartTime = time();
            }
            $cacheBuf .= $buf;
            //超过缓存尺寸 或者 超过缓存时长 写缓存到文件
            if (strlen($cacheBuf) > $CACHE_SIZE || time() - $cacheStartTime > $CACHE_DURATION) {
                $cacheStartTime = time();
                goto write;
            } else {
                if ($sync) {
                    goto write;
                } else {
                    goto skipWrite;
                }
            }
        } else {
            $cacheBuf = $buf;
        }
        write: {
            if ($logCount > 100) {//会有缓存
                clearstatcache();
            }
            //超过尺寸后 删除旧文件 把新文件重命名为旧文件
            if (($logCount == 1 || $logCount > $LOG_TIMES)
                && file_exists($LOG_NAME) && filesize($LOG_NAME) > $LOG_SIZE) {

                //获取独占锁
                $fp = fopen($LOG_NAME . '.lock', 'a');
                $k = 0;
                do {
                    $isLock = flocK($fp, LOCK_EX);
                    $k++;
                    if (!$isLock && $k > 1000) {
                        echo "lock 1000\n";
                        goto PUT;
                    }
                } while (!$isLock);

                //另外一个进程进入时 重新判断下
                clearstatcache();
                if (filesize($LOG_NAME) <= $LOG_SIZE) {
                    goto UN;
                }

                //删除旧文件和重命名文件
                $oldLogName = $LOG_NAME . '.old';
                if (file_exists($oldLogName)) {
                    if (!unlink($oldLogName)) {
                        echo "unlink err\n";
                    }
                }
                if (!rename($LOG_NAME, $oldLogName)) {
                    echo "rename err\n";
                }

                //解锁
                UN:
                flock($fp, LOCK_UN);
                fclose($fp);


                $logCount = 0;
            }
            PUT:
            if (!file_put_contents($LOG_NAME, $cacheBuf, FILE_APPEND)) {
                echo "file_put_contents err," . getcwd() . "/{$LOG_NAME} \n";
            }
            $cacheBuf = '';
        }
        skipWrite:{
        }
    }
    if ($LOG_CONSOLE) {
        echo $buf;
    }
}
function printChars($str) {
    $result = '';
    $length = strlen($str);
    for ($i = 0; $i < $length; $i++) {
        $char = $str[$i];
        $ordValue = ord($char);

        // 如果是控制字符（ASCII 值小于32 或者 是换行、回车、制表符）
        if ($ordValue >= 32 && $ordValue <= 126) {

            $result .= $char;
        } else {
            // 对于其他字符直接输出
            $result .= '.';
//            switch ($ordValue) {
//                case 9: // 制表符
//                    $result .= '\t';
//                    break;
//                case 10: // 换行符
//                    $result .= '\n';
//                    break;
//                case 13: // 回车符
//                    $result .= '\r';
//                    break;
//                default:
//                    // 其他控制字符可以使用其 ASCII 值的十六进制形式
//                    $result .= '\\x' . dechex($ordValue);
//                    break;
//            }
        }
    }
    return $result;
}


class MQTT_TYPE {/*对应从1开始*/
    public static $PUB_MEASURE = 'a'; //设备->平台 设备上报 的压力值     /*1*/
    public static $SUB_ALARM = 'b';   //平台->设备 平台下发 的告警值
    public static $SUB_ALARM_REPLAY = 'c'; //设备->平台 设备上报 对平台下发的告警值的回复
    public static $GET_ALARM = 'd'; //平台下发 获取设备告警值

    //设备->平台 设备上报 下面6个 顺序不要变 数据库里面存的是1-6 本页面insertAlarm函数有ord转对应的数字
    public static $PUB_UPPER_ALARM_START_EVENT = 'e';       /*5*/
    public static $PUB_UPPER_ALARM_STOP_EVENT = 'f';
    public static $PUB_LOWER_ALARM_START_EVENT = 'g';
    public static $PUB_LOWER_ALARM_STOP_EVENT = 'h';
    public static $PUB_UPPER_ALARM_SETTING_CHANGE = 'i';
    public static $PUB_LOWER_ALARM_SETTING_CHANGE = 'j';      /*10*/

    public static $UP_ALARM = 'k';//设备->平台

    //平台->设备 校准0
    public static $PUB_CALIBRATION_ZERO = 'l';
    //设备->平台 校准0回复
    public static $PUB_CALIBRATION_ZERO_REPLAY = 'm';
    public static $DISPLAY_UNIT_CHANGE = 'n';
    public static $DISPLAY_RATIO_CHANGE = 'o';                        /*15*/

    public static $LIQUID_ALARM = 'p';//未使用
    public static $LIQUID_ALARM_REPLAY = 'q';//未使用

    public static $SET_DAMPING_TIME = 'r';//双向 设备->平台 平台->设备 设备收到是要设置它的值 服务端收到是对下发值的回复
    public static $GET_DAMPING_TIME = 's';//双向 设备->平台 平台->设备 设备收到是要获取它的值 服务端收到是对下发的回复
}

define("ALTER_SEP", ",");
//成功返回二进制, 失败返回false
function buildMqttData($mqttType, $msgId/*整数 0-4294967295*/, $data) {
    /**test**/
//    $mqttType = MQTT_TYPE::$SUB_ALARM;
//    $data['lower'] = '12.34';
//    $data['upper'] = '56.789';
//
//    $mqttType = MQTT_TYPE::$GET_ALARM;
    /**test**/

    rlog("build", $mqttType, $msgId, $data);

    $msgId = intval($msgId);
    if ($msgId < 0 || $msgId > 0xffffffff) {
        rlog("err msgId");
        return false;
    }
    $MIN = 0;
    $MAX = 39.99;
    $ALERT_LEN = 6;//值最大

    $bin = pack("C1N1", ord($mqttType), $msgId/*time()*/ /*这个时间大小端和设备端相反*/);

    if ($mqttType == MQTT_TYPE::$SUB_ALARM) {
        /*
         * 可以单独设置一个值,不需要设置的值 可像下面这样置空
         *  $data['lower'] = ''
         * */
        if (!isset($data['lower']) || strlen($data['lower']) >= $ALERT_LEN) {
            rlog("lower val err, max len " . $ALERT_LEN);
            return false;
        }
        if (!isset($data['upper']) || strlen($data['upper']) >= $ALERT_LEN) {
            rlog("upper val err, max len " . $ALERT_LEN);
            return false;
        }

        $data['lower'] = trim($data['lower']);
        $data['upper'] = trim($data['upper']);
        if ($data['lower'] == '' && $data['upper'] == '') {
            rlog("val empty");
            return false;
        }

        //0-39.999 值范围
        //char str[] = "23.556,44.25";//设置下限和上限
        //char str[] = " ,44.25";     //设置上限 前面有个空格
        //char str[] = "23.556,";      //设置下限
        if ($data['lower'] == '') {
            $data['lower'] = ' ';//要有个空格
        } else {
            if (!preg_match('/^-?\d+(\.\d+)?$/', $data['lower'])) {
                rlog("err lower " . $data['lower']);
                return false;
            } else {
                if (floatval($data['lower']) < $MIN || floatval($data['lower']) > $MAX) {
                    rlog("err lower range " . $data['lower']);
                    return false;
                }
            }
        }
        if (!empty($data['upper'])) {
            if (!preg_match('/^-?\d+(\.\d+)?$/', $data['upper'])) {
                rlog("err upper " . $data['upper']);
                return false;
            } else {
                if (floatval($data['upper']) < $MIN || floatval($data['upper']) > $MAX) {
                    rlog("err upper range " . $data['upper']);
                    return false;
                }
            }
        }


        $buff = $data['lower'] . ALTER_SEP . $data['upper'];/*1.236,10.256*/
        $bin = $bin . $buff . chr(0);//补零 设备解析是以为\0结尾
    } else if ($mqttType == MQTT_TYPE::$GET_ALARM) {
    } else if ($mqttType == MQTT_TYPE::$SET_DAMPING_TIME) {
        $num = floatval($data);
        if ($num < 0.00 || $num > 60.00) {
            rlog("err damping time " . $data);
            return false;
        }

        $buff = $data;/*0.00-60.00*/
        $bin = $bin . $buff . chr(0);//补零 设备解析是以为\0结尾
    } else if ($mqttType == MQTT_TYPE::$GET_DAMPING_TIME) {
    }else if ($mqttType == MQTT_TYPE::$PUB_CALIBRATION_ZERO) {
    } else {
        rlog("warn unknown type " . ord($mqttType));
        return false;
    }
    rlog('build ' . bin2hex($bin) . '(' . printChars($bin) . ')');
    return $bin;
}




  

while (1) {
    $jsonData= app_redis()->rpop("pump_mqtt_config_list");
    if(!$jsonData){
        sleep(3);
        continue;
    }
    $data=json_decode($jsonData,true);
    var_dump($data);

    $topic="DEV/DATA/".$data['mac'];
    $config_json=$data['config'];
    $sendData=false;

    if($data['type']=='config'){
        $sendData = buildMqttData(MQTT_TYPE::$SUB_ALARM, $config_json['msgId'], ['lower' =>$config_json['lower'], 'upper' =>$config_json['upper']]);
    }elseif($data['type']=='alarm'){
        $sendData = buildMqttData(MQTT_TYPE::$GET_ALARM, $config_json['msgId'], null);
    }elseif($data['type']=='calibration'){
        $sendData = buildMqttData(MQTT_TYPE::$PUB_CALIBRATION_ZERO, $config_json['msgId'], null);
    }elseif($data['type']=='set_damping'){
        $sendData = buildMqttData(MQTT_TYPE::$SET_DAMPING_TIME, $config_json['msgId'], $config_json['damping_time']);
    }elseif($data['type']=='get_damping'){
        $sendData = buildMqttData(MQTT_TYPE::$GET_DAMPING_TIME, $config_json['msgId'], null);
    }else{

    }
    var_dump($sendData);
        
    if($sendData){
        updateSendResult($config_json['msgId']);
        $res=sendConfig($topic,$sendData);
    }
    
    
    
    

}