From 63c5c3d4f04ffe904374fd0afeeb3304f67c8f96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Mon, 5 Oct 2026 14:59:11 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20WebChat=20=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/df-im-gateway/Cargo.toml | 5 +- crates/df-im-gateway/src/adapters.rs | 109 +++++--- crates/df-im-gateway/web/index.html | 398 +++++++++++++++++++++++++++ 3 files changed, 473 insertions(+), 39 deletions(-) create mode 100644 crates/df-im-gateway/web/index.html diff --git a/crates/df-im-gateway/Cargo.toml b/crates/df-im-gateway/Cargo.toml index c2ad0d7..a14e189 100644 --- a/crates/df-im-gateway/Cargo.toml +++ b/crates/df-im-gateway/Cargo.toml @@ -14,7 +14,10 @@ async-trait = { workspace = true } tracing = { workspace = true } df-tunnel = { path = "../df-tunnel" } -# WS(与 df-tunnel 同款本地版本声明,workspace 无这几项) +# HTTP 静态托管 + WS 服务端(P1:Web Chat 页面由网关直接托管,/ 与 /ws 同端口) +axum = { version = "0.7", features = ["ws"] } + +# WS 客户端(commander 连 u-relay;与 df-tunnel 同款本地版本声明,workspace 无这几项) tokio-tungstenite = { version = "0.23", features = ["native-tls"] } futures-util = "0.3" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/crates/df-im-gateway/src/adapters.rs b/crates/df-im-gateway/src/adapters.rs index ca2d9d9..2b26319 100644 --- a/crates/df-im-gateway/src/adapters.rs +++ b/crates/df-im-gateway/src/adapters.rs @@ -56,22 +56,39 @@ pub trait PlatformAdapter: Send + Sync { async fn reset_stream(&self, _chat_id: &str) {} } -/// Web Chat 适配器(网关直接托管浏览器 WS,无三方依赖) +/// Web Chat 适配器(P1:axum 同端口托管静态页 + WS,浏览器直连) +/// +/// - `GET /` → 内嵌单文件聊天页(include_str!,零运行时文件依赖) +/// - `GET /ws` → WS 升级,协议见 handle_web_ws(与 P0 纯 WS 版完全一致) pub mod web { use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; + use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; + use axum::extract::State; + use axum::response::{Html, Response}; + use axum::routing::get; + use axum::Router; use futures_util::{SinkExt, StreamExt}; use tokio::sync::{mpsc, Mutex}; - use tokio_tungstenite::tungstenite::Message; use super::{InboundEvent, OutboundAction, PlatformAdapter}; - /// 浏览器连接注册表(chat_id → ws 发送通道) - type WsRegistry = Arc>>>; + /// 内嵌聊天页(单文件,无外部依赖,构建期打进二进制) + const INDEX_HTML: &str = include_str!("../web/index.html"); - /// Web 适配器:浏览器 WS 直连网关 0.0.0.0:9191(accept 协议见 handle_web_ws) + /// 浏览器连接注册表(chat_id → ws 发送通道) + type WsRegistry = Arc>>>; + + /// axum 路由共享状态(注册表 + 核心入站通道) + #[derive(Clone)] + struct WsState { + registry: WsRegistry, + events: mpsc::Sender, + } + + /// Web 适配器:HTTP+WS 服务监听 0.0.0.0:9191 #[derive(Clone)] pub struct WebAdapter { registry: WsRegistry, @@ -96,18 +113,15 @@ pub mod web { } async fn run(&self, events: mpsc::Sender) -> anyhow::Result<()> { + let state = WsState { registry: self.registry.clone(), events }; + let app = Router::new() + .route("/", get(|| async { Html(INDEX_HTML) })) + .route("/ws", get(ws_upgrade)) + .with_state(state); let listener = tokio::net::TcpListener::bind("0.0.0.0:9191").await?; - tracing::info!("[web] WS 服务监听 0.0.0.0:9191"); - loop { - let (stream, _peer) = listener.accept().await?; - let registry = self.registry.clone(); - let events = events.clone(); - tokio::spawn(async move { - if let Err(e) = handle_web_ws(stream, registry, events).await { - tracing::warn!("[web] 浏览器连接异常退出: {e}"); - } - }); - } + tracing::info!("[web] HTTP+WS 服务监听 http://0.0.0.0:9191 (页面 / , WS /ws)"); + axum::serve(listener, app).await?; + Ok(()) } async fn dispatch(&self, action: OutboundAction) -> anyhow::Result<()> { @@ -140,12 +154,21 @@ pub mod web { async fn send_ws(&self, chat_id: &str, text: &str) -> anyhow::Result<()> { let registry = self.registry.lock().await; if let Some(tx) = registry.get(chat_id) { - tx.send(Message::Text(text.to_string())).await?; + tx.send(WsMessage::Text(text.to_string())).await?; } Ok(()) } } + /// WS 升级入口 + async fn ws_upgrade(State(state): State, ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(move |socket| async move { + if let Err(e) = handle_web_ws(socket, state).await { + tracing::warn!("[web] WS 会话异常结束: {e:#}"); + } + }) + } + /// 单个浏览器 WS 会话。 /// /// 协议(浏览器 → 网关): @@ -156,19 +179,23 @@ pub mod web { /// 审批: {"type":"approval","approval_id":"...","approved":true|false} /// ``` /// 网关 → 浏览器:stream_delta / stream_done / approval / notice(见 dispatch)。 - async fn handle_web_ws( - stream: tokio::net::TcpStream, - registry: WsRegistry, - events: mpsc::Sender, - ) -> anyhow::Result<()> { - let ws = tokio_tungstenite::accept_async(stream).await?; - let (mut sink, mut upstream) = ws.split(); + async fn handle_web_ws(socket: WebSocket, state: WsState) -> anyhow::Result<()> { + let (mut sink, mut upstream) = socket.split(); // 首帧 hello 绑定 chat_id - let first = upstream.next().await.ok_or_else(|| anyhow::anyhow!("浏览器连接即断"))??; + let first = match upstream.next().await { + Some(Ok(m)) => m, + _ => { + tracing::warn!("[web] 浏览器连接即断(无首帧)"); + return Ok(()); + } + }; let hello: serde_json::Value = match &first { - Message::Text(t) => serde_json::from_str(t)?, - _ => anyhow::bail!("首帧非文本"), + WsMessage::Text(t) => serde_json::from_str(t).unwrap_or_default(), + _ => { + tracing::warn!("[web] 首帧非文本,拒绝绑定"); + return Ok(()); + } }; let chat_id = hello .get("chat_id") @@ -178,8 +205,8 @@ pub mod web { let user_id = hello.get("user_id").and_then(|v| v.as_str()).unwrap_or("anon").to_string(); // 注册发送通道:出站帧经 ws_tx → forward 任务写 socket - let (ws_tx, mut ws_rx) = mpsc::channel::(64); - registry.lock().await.insert(chat_id.clone(), ws_tx); + let (ws_tx, mut ws_rx) = mpsc::channel::(64); + state.registry.lock().await.insert(chat_id.clone(), ws_tx); tracing::info!("[web] 浏览器接入 chat_id={chat_id} user={user_id}"); @@ -193,13 +220,16 @@ pub mod web { let result: anyhow::Result<()> = async { while let Some(msg) = upstream.next().await { - match msg? { - Message::Text(t) => { + let msg = msg?; + match msg { + WsMessage::Text(t) => { let v: serde_json::Value = serde_json::from_str(&t).unwrap_or_default(); match v.get("type").and_then(|x| x.as_str()).unwrap_or("") { "message" => { - let text = v.get("text").and_then(|x| x.as_str()).unwrap_or("").to_string(); - events + let text = + v.get("text").and_then(|x| x.as_str()).unwrap_or("").to_string(); + state + .events .send(InboundEvent::Message { platform: "web".into(), chat_id: chat_id.clone(), @@ -210,7 +240,8 @@ pub mod web { .map_err(|_| anyhow::anyhow!("核心入站通道关闭"))?; } "new_session" => { - events + state + .events .send(InboundEvent::NewSession { platform: "web".into(), chat_id: chat_id.clone(), @@ -221,8 +252,10 @@ pub mod web { "approval" => { let approval_id = v.get("approval_id").and_then(|x| x.as_str()).unwrap_or("").to_string(); - let approved = v.get("approved").and_then(|x| x.as_bool()).unwrap_or(false); - events + let approved = + v.get("approved").and_then(|x| x.as_bool()).unwrap_or(false); + state + .events .send(InboundEvent::Approval { platform: "web".into(), chat_id: chat_id.clone(), @@ -235,7 +268,7 @@ pub mod web { _ => {} } } - Message::Close(_) => break, + WsMessage::Close(_) => break, _ => {} } } @@ -243,7 +276,7 @@ pub mod web { } .await; - registry.lock().await.remove(&chat_id); + state.registry.lock().await.remove(&chat_id); forward.abort(); tracing::info!("[web] 浏览器断开 chat_id={chat_id}"); result diff --git a/crates/df-im-gateway/web/index.html b/crates/df-im-gateway/web/index.html new file mode 100644 index 0000000..56a319a --- /dev/null +++ b/crates/df-im-gateway/web/index.html @@ -0,0 +1,398 @@ + + + + + +DevFlow AI Chat + + + + +
+
+ + 连接中… + + +
+
+
+ + + +
+
+ + +