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;
                        }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;

        $temp = $devList;
        if (!empty($temp)) {
            foreach ($temp as $v) {
                $lamp = $websocketDB->query("select manu,address,id from lampinfo where id = ".$v['lamp_id']);
                if (empty($lamp) || empty($lamp[0]['id'])) {
                    # code...
                }
            }
        }
    });

    // 设备升级进度
    Timer::add(2, 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']}'");
        //     }
        // }
    });
};

// 启动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();
    }
};
// 客户端断开连接
// $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();