<?php

use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\RedisQueue\Client;

require_once __DIR__ . '/vendor/autoload.php';


date_default_timezone_set('Asia/Shanghai');

// 记录日志消息  
function loginfo($msg = '')
{
    // 日志文件路径  
    $logFile = './logfile.txt';
    $logMessage = date('Y-m-d H:i:s') . ' ' . $msg . "\n";
    // 尝试将日志消息追加到文件  
    // 如果文件不存在，将创建它  
    file_put_contents($logFile, $logMessage, FILE_APPEND);
}

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

// 启动1个进程对外提供服务，这里进程数必须设置为1
$ws_worker->count = 1;

// 当收到客户端发来的数据后返回hello $data给客户端
$ws_worker->onMessage = function (TcpConnection $connection, $data) {
    // 向客户端发送hello $data
    $connection->send('hello ' . $data);
};

$ws_worker->onConnect = function ($connection) {
    global $ws_worker;
    $count = count($ws_worker->connections);
    loginfo("ws_worker->onConnect 当前在线设备数: $count");
};

$ws_worker->onClose = function ($connection) {
    global $ws_worker;
    $count = count($ws_worker->connections);
    loginfo("ws_worker->onClose 当前在线设备数: $count");
};

function sendMsg($data)
{
    global $ws_worker;
    $count = count($ws_worker->connections);
    loginfo("sendMsg 当前在线设备数: $count");
    foreach ($ws_worker->connections as $conn) {
        $conn->send($data);
    }
}

// 这个逻辑必须在 $ws_worker所在的进程中实现
// 否则拿到的 $ws_worker 变量不是监听 40002 端口的 $ws_worker 对象
$ws_worker->onWorkerStart = function () {
    $client = new Client('redis://127.0.0.1:6379');
    // 订阅
    $client->subscribe('package-queue', function ($data) {
        loginfo("package-queue " . $data);
        sendMsg($data);
    });

    // 其他订阅...
    // 消费失败触发的回调(可选)
    $client->onConsumeFailure(function (\Throwable $exception, $package) {
        echo "队列 " . $package['queue'] . " 消费失败\n";
        echo $exception->getMessage(), "\n";
        var_export($package);
    });
};

// 运行worker
Worker::runAll();