lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增: WebChat 端

Browse Source
This commit is contained in:
lxy
2026-10-05 14:59:11 +08:00
parent 7c76ee6a60
commit 63c5c3d4f0
3 changed files with 473 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-im-gateway/src/adapters.rs
+71 -38
View File
@@ -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<Mutex<HashMap<String, mpsc::Sender<Message>>>>;
/// 内嵌聊天页(单文件,无外部依赖,构建期打进二进制)
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<Mutex<HashMap<String, mpsc::Sender<WsMessage>>>>;
/// axum 路由共享状态(注册表 + 核心入站通道)
#[derive(Clone)]
struct WsState {
registry: WsRegistry,
events: mpsc::Sender<InboundEvent>,
}
/// 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<InboundEvent>) -> 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<WsState>, 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<InboundEvent>,
) -> 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::<Message>(64);
registry.lock().await.insert(chat_id.clone(), ws_tx);
let (ws_tx, mut ws_rx) = mpsc::channel::<WsMessage>(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