IDCOL/websocket/idcolWss.php

<?php
use Workerman\Worker;
require_once __DIR__ . '/Autoloader.php';
require_once './Lib/mysql/src/Connection.php';
use \Workerman\Lib\Timer;
require_once './Lib/MQTT/src/Client.php';
require_once './Lib/MQTT/src/Protocols/Mqtt.php';

$ws_worker = new Worker("websocket://0.0.0.0:8844");

// global $clientArr;
// $clientArr =[];

$ws_worker->onWorkerStart = function($ws_worker)
{
    // 将db实例存储在全局变量中(也可以存储在某类的静态成员中)
    global $websocketDB;
    global $mqtt;

    $websocketDB = new \Workerman\MySQL\Connection('rm-wz98r5cn33zq4ou980o.mysql.rds.aliyuncs.com', '3306', 'idcol20', 'idcol@1234', 'idcol');
    global $devList;
    $devList = array();
    global $devUpdateList;
    $devUpdateList = array();

    $options = array(
        'username'=>'weclouds',
        'password'=>'weclouds@1234'
    );
    $mqtt = new \Workerman\Mqtt\Client('mqtt://streetserver.weclouds.xyz:1883',$options);
    $mqtt->onConnect = function($mqtt) {

        $room = array(
            // '/IDCOL/CmdInput/#' => 2,
            '/IDCOL/CmdOutput/#' => 2,
        );

        $mqtt->subscribe($room, null, function($exception, $granted){});


        $mqtt->onMessage = function($room, $message){
            global $devUpdateList;
            global $StrategyReport;
            $idArr = explode('/', $room);
            $id = $idArr[count($idArr) - 1];
            $id = strtolower($id);
            if (strpos($room,'CmdOutput')) {
                $msg = unpack('H*',$message);
                if (isset($devUpdateList[$id]) && !empty($devUpdateList[$id])) {
                    $temp = $devUpdateList[$id];
                    $type = substr($msg[1], 0,2);
                    if ($type == '99') {
                        $status = substr($msg[1],18,2);
                        $devUpdateList[$id]['status'] = $status;
                        if ($status == '00') {
                            $devUpdateList[$id]['type'] = 1;
                            $devUpdateList[$id]['time'] = time();
                        }elseif ($status == '01') {
                            $devUpdateList[$id]['type'] = 2;
                        }
                    }
                }
            }
        };
        
    };
    $mqtt->connect();

    Timer::add(0.5, function()use($ws_worker){
    	// global $clientArr;
    	global $websocketDB;
        $connections = $ws_worker->connections;
        // var_dump($connections);
        $data = $websocketDB->select('id,client,msg')->from('message')->query();
        if (!empty($data)) {
        	foreach ($data as $v) {
        		if (!empty($connections[$v['client']])) $connections[$v['client']]->send($v['msg']); 
        		$websocketDB->query("DELETE FROM `message` WHERE id='{$v['id']}'");
        	}
        }
    });

    // 下发设备升级指令
    Timer::add(1, function()use($ws_worker){
        global $mqtt;
        global $devUpdateList;
        global $devList;
        global $websocketDB;

        $url = 'http://120.77.179.55';
        $topick = '/IDCOL/CmdInput/';
        $temp = $devList;
        if (!empty($temp)) {
            foreach ($temp as $k => $v) {
                $client = $v['client'];
                $lamp = $websocketDB->query("select manu,address,id from lampinfo where id = ".$v['lamp_id']);
                if (!empty($lamp) && !empty($lamp[0]['id'])) {
                    $manu = $lamp[0]['id']['manu'];
                    $address = $lamp[0]['id']['address'];
                    if (!empty($manu)) {
                        $firmware_info = $websocketDB->query("select version,id,path from firmware_info where manuId = ".$v['manu']);
                        if (!empty($firmware_info) && !empty($firmware_info[0]['id'])) {
                            $file_path = $url.$firmware_info[0]['path'];
                            $cmd = '190100010000000000';
                            while (strlen($file_path) < 200) {
                                $file_path .= '0';
                            }
                            $cmd .= $file_path;
                            $cmd = crc16(unpack('H*', $cmd));
                            $mqtt->publish($topick.$address, unpack('H*', $cmd));

                            $index = strtolower($address);
                            $devUpdateList[$index] = array('client'=>$client,'lamp_id'=>$v['lamp_id'],'time'=>time(),'type'=>0,'step'=>0);
                            $websocketDB->update('lampinfo')->cols(array('is_update'=>1))->where('id='.$v['lamp_id']);
                            unset($devList[$k]);
                            continue;
                        }else{
                            $data = ['code'=>'9999','status'=>2,'msg'=>'No firmware file found, please contact your administrator','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'status'=>2]];
                        }
                    }else{
                        $data = ['code'=>'9999','status'=>2,'msg'=>'Device not bound to manufacturer','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'status'=>2]];
                    }
                }else{
                    $data = ['code'=>'9999','status'=>2,'msg'=>'Device does not exist','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'status'=>2]];
                }
                $client->send(json_encode($data));
                unset($devList[$k]);
            }
        }
    });

    // 设备升级进度
    Timer::add(2, function()use($ws_worker){
        global $devUpdateList;
        global $websocketDB;
        $temp = $devUpdateList;

        if (!empty($temp)) {
            foreach ($temp as $k => $v) {
                if ($v['type'] == 0) {  // 升级指令还没返回
                    if (time() - $v['time'] > 10) {  // 指令超时
                        $data = ['code'=>'9999','status'=>2,'msg'=>'The command is timed out, please try again later','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'status'=>2]];
                        unset($devUpdateList[$k]);
                    }
                }elseif ($v['type'] == 1) {  // 升级中
                    if (time() - $v['time'] > 60 * 5) {  // 指令超时
                        $data = ['code'=>'9999','status'=>2,'msg'=>'Please try again later. Please upgrade the device','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'status'=>2]];
                        unset($devUpdateList[$k]);
                    }else{
                        if ($devUpdateList[$k]['step'] < 95) $devUpdateList[$k]['step'] += 1;
                        
                        $data = ['code'=>'0000','status'=>2,'msg'=>'Successful operation','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'step'=>$v['step'],'status'=>2]];
                    }
                }elseif ($v['type'] == 2) {  // 升级成功
                    $data = ['code'=>'0000','status'=>2,'msg'=>'Successful operation','data'=>['type'=>'devUpdate','id'=>$v['lamp_id'],'step'=>100,'status'=>2]];
                    unset($devUpdateList[$k]);
                }
                $v['client']->send(json_encode($data));
            }
        }
    });
};

// 启动1个进程对外提供服务
$ws_worker->count = 1;

// 客户端连接回调
$ws_worker->onConnect = function($connection)
{
	// global $clientArr;
	// 标记连接的客户端
    // $key=uniqid();
    $key = $connection->id;
    $data = array('code'=>'0000','msg'=>'操作成功','data'=>['clientid'=>$key,'status'=>1]);
    $connection->send(json_encode($data));
    // $clientArr[$key] = $connection;
};
$ws_worker->onMessage = function($connection, $data)
{
    
    global $devList;
    $res = json_decode($data,true);

    if ($res && isset($res['type'])) {
        if ($res['type'] == 'updateDev' && isset($res['id']) && !empty($res['id'])){
            $arr = explode(','$res['id']);
            if (!empty($arr)) {
                foreach ($arr as $key) {
                    if (is_int($key) && !empty($key)) {
                        $devList[] = array('lamp_id'=>intval($key),'client'=>$connection);
                    }
                }
            }else{
                $connection->close();
            }
        }else{
            $connection->close();
        }
    }else{
        $connection->close();
    }
};

// modbus crc 16 校验码计算
function crc16($string) {
  $crc = 0xFFFF;
  for ($x = 0; $x < strlen ($string); $x++) {
    $crc = $crc ^ ord($string[$x]);
    for ($y = 0; $y < 8; $y++) {
      if (($crc & 0x0001) == 0x0001) {
        $crc = (($crc >> 1) ^ 0xA001);
      } else { $crc = $crc >> 1; }
    }
  }
  $crc = strval(base_convert($crc, 10, 16));
  $crc = substr('0000'.$crc, -4);
  return substr($crc, -2).substr($crc, 0,2);
}
// 客户端断开连接
// $ws_worker->onClose = function($connection)
// {
// 	global $clientArr;
//     if (!empty($clientArr)) {
//     	foreach ($clientArr as $k => $v) {
//     		if ($v == $connection) {
//     			unset($clientArr[$k]);
//     			break;
//     		}
//     	}
//     }
// };
// 运行worker
Worker::runAll();