From c2c89014d1d0c0ca4faf65344ff66a9df6c85fab Mon Sep 17 00:00:00 2001
From: JoeAllen <2494136784@qq.com>
Date: Mon, 12 Oct 2026 15:12:18 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E6=81=A2=E5=A4=8D=E8=AF=AF=E5=88=A0?=
 =?UTF-8?q?=E7=9A=84=20examples/flood.rs=EF=BC=88=E6=B8=85=E7=90=86?=
 =?UTF-8?q?=E6=8E=A2=E9=92=88=E6=97=B6=E8=AF=AF=E8=BF=9E=E7=9B=AE=E5=BD=95?=
 =?UTF-8?q?=E5=88=A0=E9=99=A4=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 gpui-app/examples/flood.rs | 66 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 66 insertions(+)
 create mode 100644 gpui-app/examples/flood.rs

diff --git a/gpui-app/examples/flood.rs b/gpui-app/examples/flood.rs
new file mode 100644
index 0000000..168f219
--- /dev/null
+++ b/gpui-app/examples/flood.rs
@@ -0,0 +1,66 @@
+//! 消息流量压测工具：连接 broker 后按指定速率向主题发布消息。
+//!
+//! 用法：cargo run --example flood -- <host> <port> <topic> <count> <per_second> [payload_bytes]
+//! 例：向私有 broker 发 500 条 1KB 消息、每秒 50 条：
+//!   cargo run --example flood -- mqtt.nex-i.cn 1883 mqttx-flood 500 50 1024
+
+use std::time::{Duration, Instant};
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() < 6 {
+        eprintln!("usage: flood <host> <port> <topic> <count> <per_second> [payload_bytes]");
+        std::process::exit(1);
+    }
+    let host = args[1].clone();
+    let port: u16 = args[2].parse().unwrap();
+    let topic = args[3].clone();
+    let count: u64 = args[4].parse().unwrap();
+    let per_second: u64 = args[5].parse().unwrap();
+    let payload_size: usize = args.get(6).map(|s| s.parse().unwrap()).unwrap_or(512);
+
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .unwrap();
+    rt.block_on(async move {
+        let opts = rumqttc::MqttOptions::new(
+            format!("mqttx_flood_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]),
+            rumqttc::Broker::tcp(&host, port),
+        );
+        let (client, mut eventloop) = rumqttc::AsyncClient::builder(opts).build();
+
+        // eventloop 独立持续 poll：请求出队全靠它
+        tokio::spawn(async move {
+            loop {
+                if eventloop.poll().await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        let payload = vec![b'x'; payload_size];
+        let interval = Duration::from_secs_f64(1.0 / per_second.max(1) as f64);
+        let start = Instant::now();
+        for i in 0..count {
+            let topic_i = format!("{topic}/{i}");
+            let options = rumqttc::PublishOptions::new(rumqttc::QoS::AtMostOnce);
+            if let Err(e) = client.publish(&topic_i, payload.clone(), options).await {
+                eprintln!("publish error at {i}: {e}");
+                break;
+            }
+            let next = start + interval * (i as u32 + 1);
+            let now = Instant::now();
+            if next > now {
+                tokio::time::sleep(next - now).await;
+            }
+        }
+        eprintln!(
+            "flooded {count} msgs ({payload_size}B each) in {:.1}s",
+            start.elapsed().as_secs_f64()
+        );
+        // 等网络冲刷
+        tokio::time::sleep(Duration::from_secs(3)).await;
+    });
+}