<?php
/**
 * mqttbroker 压力测试脚本（零依赖，纯 PHP stream_socket 实现最小 MQTT 客户端）
 *
 * 用途：验证 Broker 在大量并发连接 + 高频发布下的稳定性（对应路线 B2）。
 * 测试项：
 *   1. 并发连接建立（CONNECT/CONNACK）
 *   2. 订阅 + 发布 QoS0 吞吐
 *   3. 客户端掉线重连恢复（可选）
 *
 * 用法：
 *   php addon/mqttbroker/benchmark.php --host=127.0.0.1 --port=1883 --clients=500 --pub=20 --topic=bench/%d
 *
 * 参数：
 *   --host      Broker 地址（默认 127.0.0.1）
 *   --port      监听端口（默认 1883；WebSocket 用 --ws 且 port=8083）
 *   --clients   并发客户端数（默认 200）
 *   --pub       每个客户端发布消息数（默认 10）
 *   --topic     发布主题模板，%d 替换为客户端序号
 *   --qos       发布 QoS（默认 0）
 *   --timeout   单步超时秒（默认 3）
 */
declare(strict_types=1);

$opts = getopt('', ['host:', 'port:', 'clients:', 'pub:', 'topic:', 'qos:', 'timeout:', 'ws']);
$host   = $opts['host']   ?? '127.0.0.1';
$port   = (int) ($opts['port'] ?? 1883);
$clients = (int) ($opts['clients'] ?? 200);
$pubN   = (int) ($opts['pub'] ?? 10);
$topicTpl = $opts['topic'] ?? 'bench/%d';
$qos    = (int) ($opts['qos'] ?? 0);
$timeout = (float) ($opts['timeout'] ?? 3);
$ws     = isset($opts['ws']);

echo "=== mqttbroker benchmark ===\n";
echo "target={$host}:{$port} clients={$clients} pub/client={$pubN} qos={$qos} transport=" . ($ws ? 'ws' : 'tcp') . "\n";

/* ---------- 最小 MQTT 报文 ---------- */
function encStr(string $s): string { $b = strlen($s); return chr($b >> 8) . chr($b & 0xff) . $s; }
function buildConnect(string $cid): string {
    $var = chr(0x00) . chr(0x04) . 'MQTT' . chr(0x04) . chr(0x02) . chr(0x00) . chr(0x3c);
    $body = encStr($cid);
    $len = strlen($body);
    return chr(0x10) . chr($len) . $var . $body;
}
function buildPublish(string $topic, string $payload, int $qos): string {
    $body = encStr($topic) . $payload;
    $len = strlen($body);
    return chr(0x30 | ($qos << 1)) . chr($len) . $body;
}
function buildSubscribe(string $topic, int $qos): string {
    $body = chr(0x00) . chr(0x01) . encStr($topic) . chr($qos);
    $len = strlen($body);
    return chr(0x82) . chr($len) . $body;
}

/* ---------- 连接并建立会话 ---------- */
function dial(string $host, int $port, bool $ws, float $timeout): mixed {
    $ctx = stream_context_create();
    $uri = $ws ? "tcp://{$host}:{$port}" : "tcp://{$host}:{$port}";
    $fp = @stream_socket_client($uri, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, $ctx);
    if (!$fp) { return null; }
    stream_set_timeout($fp, (int) $timeout);
    if ($ws) {
        // 极简 WS 握手（仅 TCP 升级，不处理掩码帧的完整分帧，供基准参考）
        $req = "GET /mqtt HTTP/1.1\r\nHost: {$host}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: " . base64_encode(random_bytes(16)) . "\r\nSec-WebSocket-Protocol: mqtt\r\nSec-WebSocket-Version: 13\r\n\r\n";
        fwrite($fp, $req);
        $hdr = fread($fp, 1024);
        if (strpos($hdr, '101') === false) { fclose($fp); return null; }
    }
    return $fp;
}

$start = microtime(true);
$ok = 0; $fail = 0; $published = 0;
$pool = [];
for ($i = 1; $i <= $clients; $i++) {
    $cid = "bench-{$i}-" . rand(1000, 9999);
    $fp = dial($host, $port, $ws, $timeout);
    if (!$fp) { $fail++; continue; }
    fwrite($fp, buildConnect($cid));
    $ack = @fread($fp, 4);
    if ($ack === false || strlen($ack) < 4) { fclose($fp); $fail++; continue; }
    // 订阅自身主题
    $t = sprintf($topicTpl, $i);
    fwrite($fp, buildSubscribe($t, $qos));
    @fread($fp, 3);
    $pool[$i] = $fp;
    $ok++;
}
$connTime = microtime(true) - $start;
echo sprintf("连接结果: 成功=%d 失败=%d 耗时=%.2fs (%.0f conn/s)\n", $ok, $fail, $connTime, $ok / max(0.001, $connTime));

/* ---------- 发布阶段 ---------- */
$pubStart = microtime(true);
foreach ($pool as $i => $fp) {
    $t = sprintf($topicTpl, $i);
    for ($k = 0; $k < $pubN; $k++) {
        fwrite($fp, buildPublish($t, "msg {$k} from {$i}", $qos));
        $published++;
    }
}
$pubTime = microtime(true) - $pubStart;
echo sprintf("发布结果: 总消息=%d 耗时=%.2fs (%.0f msg/s)\n", $published, $pubTime, $published / max(0.001, $pubTime));

/* ---------- 清理 ---------- */
foreach ($pool as $fp) { @fclose($fp); }
$total = microtime(true) - $start;
echo sprintf("完成: 客户端=%d 消息=%d 总耗时=%.2fs\n", $ok, $published, $total);
echo ($fail === 0 && $ok > 0) ? "RESULT: PASS\n" : "RESULT: PARTIAL/FAIL (fail={$fail})\n";