ZengQingming/JoyD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Claw 项目完整结构提交

Browse Source
This commit is contained in:
zqm
2026-03-16 15:47:55 +08:00
parent ca4970bcbf
commit fb0aeb6ca2
118 changed files with 28648 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
Claw/Server/SmartClaw/Cargo.toml Normal file
View File

@@ -0,0 +1,19 @@
[package]
name = "smartclaw"
version = "0.1.0"
edition = "2024"
[dependencies]
actix-web = "^4.0"
tokio = { version = "^1.0", features = ["full"] }
serde = { version = "^1.0", features = ["derive"] }
serde_json = "^1.0"
reqwest = { version = "^0.11", features = ["json"] }
# mini-redis = "0.4" # 使用自定义嵌入式Redis实现
tokio-tungstenite = "^0.18"
shared = { path = "../shared" }
env_logger = "^0.10"
awc = { version = "^3.0", features = ["rustls"] }
futures-util = "^0.3"
chrono = "^0.4"
# heed = "^0.20" # 暂时移除，后续实现HeedDB功能

501
Claw/Server/SmartClaw/src/main.rs Normal file
View File

@@ -0,0 +1,501 @@
use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer, Responder, middleware::Logger};
use std::env;
use shared::{TaskRequest, TaskResponse, HealthResponse, utils};
mod websocket_client;
use websocket_client::WebSocketClientManager;
/// 智能控制核心服务
struct SmartClawService;
impl SmartClawService {
/// 处理来自网关服务的任务
async fn process_task(task: TaskRequest) -> TaskResponse {
println!("🤖 SmartClaw 收到任务:");
println!(" 用户ID: {}", task.user_id);
println!(" 任务类型: {}", task.task_type);
println!(" 内容长度: {} 字符", task.content.len());
println!(" 优先级: {}", task.priority);
// 记录任务开始时间
let start_time = std::time::Instant::now();
// 根据任务类型进行不同的处理
let result = match task.task_type {
shared::TaskType::TextProcessing => {
Self::process_text_task(&task.content).await
},
shared::TaskType::DataAnalysis => {
Self::process_data_analysis_task(&task.content).await
},
shared::TaskType::AIChat => {
Self::process_ai_chat_task(&task.content).await
},
shared::TaskType::FileProcessing => {
Self::process_file_task(&task.content).await
},
shared::TaskType::Custom(ref custom_type) => {
Self::process_custom_task(&task.content, custom_type).await
},
};
let processing_time = start_time.elapsed().as_millis() as u64;
match result {
Ok(response_data) => {
println!("✅ 任务处理成功，耗时: {}ms", processing_time);
let mut response = utils::create_success_response(
"任务处理成功",
Some(utils::generate_task_id(&task.user_id)),
Some(response_data)
);
response.processing_time = Some(processing_time);
response
},
Err(error_message) => {
println!("❌ 任务处理失败: {}", error_message);
let mut response = utils::create_error_response(
"任务处理失败",
Some(error_message)
);
response.processing_time = Some(processing_time);
response
}
}
}
/// 处理文本处理任务
async fn process_text_task(content: &str) -> Result<serde_json::Value, String> {
println!("📝 处理文本任务");
// TODO: 集成 LMStudio 进行文本处理
// 模拟处理过程
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let word_count = content.split_whitespace().count();
let char_count = content.chars().count();
Ok(serde_json::json!({
"task_type": "text_processing",
"word_count": word_count,
"char_count": char_count,
"processed_content": format!("[已处理] {}", content),
"lmstudio_result": "TODO: 集成 LMStudio 处理结果"
}))
}
/// 处理数据分析任务
async fn process_data_analysis_task(content: &str) -> Result<serde_json::Value, String> {
println!("📊 处理数据分析任务");
// TODO: 集成 LMStudio 进行数据分析
// 模拟数据处理
tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
// 简单的数据分析示例
let lines: Vec<&str> = content.lines().collect();
let line_count = lines.len();
Ok(serde_json::json!({
"task_type": "data_analysis",
"line_count": line_count,
"data_summary": {
"total_lines": line_count,
"sample_data": lines.iter().take(5).collect::<Vec<_>>()
},
"lmstudio_analysis": "TODO: 集成 LMStudio 分析结果"
}))
}
/// 处理 AI 对话任务
async fn process_ai_chat_task(content: &str) -> Result<serde_json::Value, String> {
println!("💬 处理 AI 对话任务");
// TODO: 集成 LMStudio 进行 AI 对话
// 模拟 AI 处理
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
let response = format!("这是一个模拟的 AI 回复，您的消息是: {}", content);
Ok(serde_json::json!({
"task_type": "ai_chat",
"user_message": content,
"ai_response": response,
"lmstudio_response": "TODO: 集成 LMStudio 对话结果"
}))
}
/// 处理文件处理任务
async fn process_file_task(content: &str) -> Result<serde_json::Value, String> {
println!("📁 处理文件任务");
// TODO: 集成 LMStudio 进行文件内容处理
// 模拟文件处理
tokio::time::sleep(tokio::time::Duration::from_millis(300)).await;
Ok(serde_json::json!({
"task_type": "file_processing",
"file_info": {
"content_preview": content.chars().take(100).collect::<String>() + "...",
"content_length": content.len()
},
"processing_result": "文件内容已接收并处理",
"lmstudio_processing": "TODO: 集成 LMStudio 文件处理结果"
}))
}
/// 处理自定义任务
async fn process_custom_task(content: &str, custom_type: &str) -> Result<serde_json::Value, String> {
println!("🔧 处理自定义任务: {}", custom_type);
// 根据自定义类型进行不同的处理
match custom_type {
"translation" => {
Self::process_translation_task(content).await
},
"summarization" => {
Self::process_summarization_task(content).await
},
_ => {
// 默认处理
Ok(serde_json::json!({
"task_type": format!("custom_{}", custom_type),
"content": content,
"result": format!("自定义任务 '{}' 已处理", custom_type),
"note": "这是默认的自定义任务处理结果"
}))
}
}
}
/// 处理翻译任务
async fn process_translation_task(content: &str) -> Result<serde_json::Value, String> {
println!("🌐 处理翻译任务");
// TODO: 集成 LMStudio 进行翻译
let translated = format!("[翻译结果] {}", content);
Ok(serde_json::json!({
"task_type": "custom_translation",
"original_text": content,
"translated_text": translated,
"source_language": "auto",
"target_language": "zh-CN",
"lmstudio_translation": "TODO: 集成 LMStudio 翻译结果"
}))
}
/// 处理摘要任务
async fn process_summarization_task(content: &str) -> Result<serde_json::Value, String> {
println!("📋 处理摘要任务");
// TODO: 集成 LMStudio 进行文本摘要
let summary = format!("这是文本的摘要: {}", content.chars().take(50).collect::<String>());
Ok(serde_json::json!({
"task_type": "custom_summarization",
"original_text": content,
"summary": summary,
"summary_length": summary.len(),
"lmstudio_summary": "TODO: 集成 LMStudio 摘要结果"
}))
}
/// 获取服务状态
fn get_service_status() -> serde_json::Value {
serde_json::json!({
"service": "smartclaw",
"status": "running",
"version": env!("CARGO_PKG_VERSION"),
"capabilities": [
"text_processing",
"data_analysis",
"ai_chat",
"file_processing",
"custom_tasks"
],
"lmstudio_status": "TODO: 检查 LMStudio 连接状态",
"active_tasks": 0, // TODO: 统计当前活跃任务数
"queue_length": 0 // TODO: 统计队列长度
})
}
}
/// 健康检查处理器
async fn health_check() -> impl Responder {
let response = HealthResponse {
status: "healthy".to_string(),
service: "smartclaw".to_string(),
timestamp: utils::current_timestamp(),
version: env!("CARGO_PKG_VERSION").to_string(),
extra: Some(SmartClawService::get_service_status()),
};
HttpResponse::Ok().json(response)
}
/// 任务处理处理器
async fn handle_task(task: web::Json<TaskRequest>) -> impl Responder {
let response = SmartClawService::process_task(task.into_inner()).await;
HttpResponse::Ok().json(response)
}
/// 系统状态处理器
async fn system_status() -> impl Responder {
HttpResponse::Ok().json(SmartClawService::get_service_status())
}
/// LMStudio 状态检查
async fn lmstudio_status() -> impl Responder {
// TODO: 实现 LMStudio 连接状态检查
HttpResponse::Ok().json(serde_json::json!({
"lmstudio": {
"connected": false,
"status": "not_implemented",
"message": "LMStudio 集成尚未实现"
},
"timestamp": utils::current_timestamp()
}))
}
/// WebSocket 断开连接处理器（用于测试）
async fn websocket_disconnect(ws_manager: web::Data<WebSocketClientManager>) -> impl Responder {
println!("🔌 收到WebSocket断开连接请求");
// 获取客户端实例
let client = ws_manager.get_client();
if client.is_connected() {
println!("🔗 WebSocket客户端当前已连接，准备断开...");
// 断开连接
client.disconnect();
println!("✅ WebSocket客户端已断开连接");
HttpResponse::Ok().json(serde_json::json!({
"status": "disconnected",
"message": "WebSocket客户端已断开连接",
"timestamp": utils::current_timestamp()
}))
} else {
println!("⚠️ WebSocket客户端当前未连接");
HttpResponse::Ok().json(serde_json::json!({
"status": "not_connected",
"message": "WebSocket客户端当前未连接",
"timestamp": utils::current_timestamp()
}))
}
}
/// WebSocket 停止管理器处理器（用于测试）
async fn websocket_stop(ws_manager: web::Data<WebSocketClientManager>) -> impl Responder {
println!("🛑 收到WebSocket停止管理器请求");
// 停止管理器
ws_manager.stop();
println!("✅ WebSocket客户端管理器已停止");
HttpResponse::Ok().json(serde_json::json!({
"status": "stopped",
"message": "WebSocket客户端管理器已停止",
"timestamp": utils::current_timestamp()
}))
}
/// WebSocket 连接处理器（用于与网关服务通信）
async fn websocket_handler(req: HttpRequest, _stream: web::Payload) -> impl Responder {
// TODO: 实现 WebSocket 通信
// 这将用于接收来自网关服务的实时任务
println!("📡 WebSocket 连接请求: {:?}", req);
// 获取WebSocket客户端实例（用于测试和演示）
let gateway_url = env::var("GATEWAY_URL").unwrap_or_else(|_| "http://localhost:8000".to_string());
let ws_manager = WebSocketClientManager::new(gateway_url);
// 获取客户端实例（用于测试）
let client = ws_manager.get_client();
// 检查连接状态（用于演示）
if client.is_connected() {
println!("🔗 WebSocket客户端已连接");
// 测试发送消息（用于演示）
let test_message = serde_json::json!({
"type": "test",
"message": "来自SmartClaw的测试消息",
"timestamp": utils::current_timestamp()
}).to_string();
match client.send_message(test_message).await {
Ok(_) => {
println!("✅ 测试消息发送成功");
},
Err(e) => {
println!("⚠️ 测试消息发送失败: {}", e);
}
}
// 测试发送任务响应（用于演示）
let test_response = shared::TaskResponse {
success: true,
message: "测试响应".to_string(),
task_id: Some("test_task_123".to_string()),
result: Some(serde_json::json!({
"test_data": "这是测试数据",
"websocket_status": "connected"
})),
processing_time: Some(100),
error: None,
};
match client.send_task_response(test_response).await {
Ok(_) => {
println!("✅ 测试任务响应发送成功");
},
Err(e) => {
println!("⚠️ 测试任务响应发送失败: {}", e);
}
}
} else {
println!("⚠️ WebSocket客户端未连接");
}
// 暂时返回不支持的消息
HttpResponse::NotImplemented().json(serde_json::json!({
"error": "WebSocket not implemented",
"message": "WebSocket 功能正在开发中",
"websocket_status": if client.is_connected() { "connected" } else { "disconnected" }
}))
}
/// 任务队列状态
async fn queue_status() -> impl Responder {
// TODO: 实现任务队列状态查询
HttpResponse::Ok().json(serde_json::json!({
"queue": {
"length": 0,
"pending": 0,
"processing": 0,
"completed": 0,
"failed": 0
},
"timestamp": utils::current_timestamp()
}))
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
// 初始化日志
env_logger::init_from_env(
env_logger::Env::new().default_filter_or("info,actix_web=info")
);
let port = env::var("PORT").unwrap_or_else(|_| "3001".to_string());
let bind_address = format!("0.0.0.0:{}", port);
// 获取网关服务地址
let gateway_url = env::var("GATEWAY_URL").unwrap_or_else(|_| "http://localhost:8000".to_string());
println!("🚀 SmartClaw 服务启动中...");
println!("📍 绑定地址: {}", bind_address);
println!("🔗 网关服务: {}", gateway_url);
println!("📝 日志级别: info");
println!("🔧 版本: {}", env!("CARGO_PKG_VERSION"));
println!("🎯 环境: {}", env::var("ENVIRONMENT").unwrap_or_else(|_| "development".to_string()));
// 创建 WebSocket 客户端管理器
let ws_manager = WebSocketClientManager::new(gateway_url.clone());
// 启动 WebSocket 连接（在后台任务中）
let ws_manager_for_spawn = ws_manager.clone();
std::thread::spawn(move || {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
println!("🔄 正在启动 WebSocket 客户端连接...");
match ws_manager_for_spawn.start().await {
Ok(_) => {
println!("✅ WebSocket 客户端连接成功");
}
Err(e) => {
println!("❌ WebSocket 客户端连接失败: {}", e);
// 这里可以添加重试逻辑
}
}
});
});
let ws_manager_for_server = ws_manager.clone();
let server = HttpServer::new(move || {
App::new()
.app_data(web::Data::new(ws_manager_for_server.clone()))
.wrap(Logger::default())
.service(
web::scope("/api/v1")
// 健康检查
.route("/health", web::get().to(health_check))
// 系统状态
.route("/system", web::get().to(system_status))
// 任务处理
.route("/task", web::post().to(handle_task))
// LMStudio 状态
.route("/lmstudio/status", web::get().to(lmstudio_status))
// 队列状态
.route("/queue/status", web::get().to(queue_status))
// WebSocket 连接（用于与网关服务通信）
.route("/websocket", web::get().to(websocket_handler))
// WebSocket 断开连接（用于测试）
.route("/websocket/disconnect", web::post().to(websocket_disconnect))
// WebSocket 停止管理器（用于测试）
.route("/websocket/stop", web::post().to(websocket_stop))
)
})
.bind(&bind_address)?
.run();
println!("✅ SmartClaw 服务已启动在 {}", bind_address);
println!("🔍 可用接口:");
println!(" GET /api/v1/health - 健康检查");
println!(" GET /api/v1/system - 系统状态");
println!(" POST /api/v1/task - 处理任务");
println!(" GET /api/v1/lmstudio/status - LMStudio 状态");
println!(" GET /api/v1/queue/status - 队列状态");
println!(" GET /api/v1/websocket - WebSocket 连接");
println!(" POST /api/v1/websocket/disconnect - WebSocket 断开连接（测试）");
println!(" POST /api/v1/websocket/stop - WebSocket 停止管理器（测试）");
println!("💡 WebSocket 客户端已配置，连接到: {}", gateway_url);
// 启动优雅关闭处理
let ws_manager_clone = ws_manager.clone();
tokio::spawn(async move {
// 监听关闭信号（这里简化处理，实际应该监听系统信号）
match tokio::signal::ctrl_c().await {
Ok(_) => {
println!("🛑 收到关闭信号，开始优雅关闭...");
graceful_shutdown(ws_manager_clone).await;
std::process::exit(0);
},
Err(e) => {
println!("⚠️ 监听关闭信号失败: {}", e);
}
}
});
server.await
}
/// 优雅关闭处理
async fn graceful_shutdown(ws_manager: WebSocketClientManager) {
println!("🛑 开始优雅关闭...");
// 停止WebSocket客户端
ws_manager.stop();
println!("🔌 WebSocket客户端已停止");
// 等待一段时间确保所有连接都已关闭
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
println!("✅ 优雅关闭完成");
}

214
Claw/Server/SmartClaw/src/websocket_client.rs Normal file
View File

@@ -0,0 +1,214 @@
use tokio_tungstenite::{connect_async, tungstenite::Message};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::{interval, Duration};
use shared::{TaskRequest, TaskResponse};
/// WebSocket 客户端连接管理器
pub struct WebSocketClient {
gateway_url: String,
sender: Arc<std::sync::Mutex<Option<mpsc::Sender<String>>>>,
is_connected: Arc<std::sync::Mutex<bool>>,
}
impl WebSocketClient {
/// 创建新的 WebSocket 客户端
pub fn new(gateway_url: String) -> Self {
Self {
gateway_url,
sender: Arc::new(std::sync::Mutex::new(None)),
is_connected: Arc::new(std::sync::Mutex::new(false)),
}
}
/// 连接到网关服务
pub async fn connect(&self) -> Result<(), Box<dyn std::error::Error>> {
println!("🔌 正在连接到网关服务: {}", self.gateway_url);
let ws_url = format!("{}/ws", self.gateway_url.replace("http://", "ws://").replace("https://", "wss://"));
println!("🔗 WebSocket URL: {}", ws_url);
// 建立 WebSocket 连接
let (ws_stream, _) = connect_async(&ws_url).await?;
println!("✅ WebSocket 连接建立");
// 设置连接状态
*self.is_connected.lock().unwrap() = true;
// 分割流
let (mut write, mut read) = ws_stream.split();
// 创建消息通道
let (tx, mut rx) = mpsc::channel::<String>(100);
*self.sender.lock().unwrap() = Some(tx);
// 启动消息发送循环
let _write_handle = tokio::spawn(async move {
while let Some(msg) = rx.recv().await {
if let Err(e) = write.send(Message::Text(msg)).await {
println!("❌ 发送消息失败: {}", e);
break;
}
}
});
// 启动消息接收循环
let is_connected_clone = self.is_connected.clone();
let _read_handle = tokio::spawn(async move {
while let Some(msg) = read.next().await {
match msg {
Ok(Message::Text(text)) => {
println!("📨 收到消息: {}", text);
if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&text) {
Self::handle_incoming_message(parsed).await;
}
}
Ok(Message::Close(_)) => {
println!("🔚 收到关闭消息");
*is_connected_clone.lock().unwrap() = false;
break;
}
Ok(_) => {}
Err(e) => {
println!("❌ 接收消息错误: {}", e);
*is_connected_clone.lock().unwrap() = false;
break;
}
}
}
});
// 启动心跳机制
let _heartbeat_handle = {
let is_connected = self.is_connected.clone();
tokio::spawn(async move {
let mut heartbeat_interval = interval(Duration::from_secs(30));
loop {
heartbeat_interval.tick().await;
let connected = *is_connected.lock().unwrap();
if !connected {
println!("💔 心跳检测到连接已断开");
break;
}
let _heartbeat_msg = json!({
"type": "heartbeat",
"service": "smartclaw",
"timestamp": chrono::Utc::now().timestamp()
}).to_string();
// 这里需要重新获取 sender，因为生命周期问题
println!("💓 心跳发送");
}
})
};
// 发送连接确认消息
let connect_msg = json!({
"type": "connect",
"service": "smartclaw",
"version": env!("CARGO_PKG_VERSION"),
"timestamp": chrono::Utc::now().timestamp()
}).to_string();
if let Some(sender) = &*self.sender.lock().unwrap() {
let _ = sender.send(connect_msg).await;
}
println!("🚀 WebSocket 客户端已启动");
Ok(())
}
/// 处理接收到的消息
async fn handle_incoming_message(message: serde_json::Value) {
match message.get("type").and_then(|t| t.as_str()) {
Some("task") => {
// 处理任务消息
if let Ok(task_request) = serde_json::from_value::<TaskRequest>(message) {
println!("📝 收到任务请求: {:?}", task_request);
// 这里可以调用任务处理逻辑
}
}
Some("heartbeat") => {
println!("💓 收到心跳响应");
}
Some("ack") => {
println!("✅ 收到确认消息");
}
Some(msg_type) => {
println!("❓ 收到未知消息类型: {}", msg_type);
}
None => {
println!("❓ 收到无类型消息");
}
}
}
/// 发送消息
pub async fn send_message(&self, message: String) -> Result<(), Box<dyn std::error::Error>> {
if let Some(sender) = &*self.sender.lock().unwrap() {
sender.send(message).await.map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;
Ok(())
} else {
Err("WebSocket 连接未建立".into())
}
}
/// 发送任务响应
pub async fn send_task_response(&self, response: TaskResponse) -> Result<(), Box<dyn std::error::Error>> {
let message = json!({
"type": "task_response",
"task_id": response.task_id,
"data": response,
"timestamp": chrono::Utc::now().timestamp()
}).to_string();
self.send_message(message).await
}
/// 检查连接状态
pub fn is_connected(&self) -> bool {
*self.is_connected.lock().unwrap()
}
/// 断开连接
pub fn disconnect(&self) {
*self.is_connected.lock().unwrap() = false;
*self.sender.lock().unwrap() = None;
println!("🔌 WebSocket 连接已断开");
}
}
/// WebSocket 客户端管理器
#[derive(Clone)]
pub struct WebSocketClientManager {
client: Arc<WebSocketClient>,
}
impl WebSocketClientManager {
/// 创建新的管理器
pub fn new(gateway_url: String) -> Self {
Self {
client: Arc::new(WebSocketClient::new(gateway_url)),
}
}
/// 启动客户端连接
pub async fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
self.client.connect().await
}
/// 获取客户端实例
pub fn get_client(&self) -> Arc<WebSocketClient> {
self.client.clone()
}
/// 停止客户端
pub fn stop(&self) {
self.client.disconnect();
}
}

23
Claw/Server/gateway/Cargo.toml Normal file
View File

@@ -0,0 +1,23 @@
[package]
name = "gateway"
version = "0.1.0"
edition = "2024"
[dependencies]
actix-web = "^4.0"
tokio = { version = "^1.0", features = ["full"] }
serde = { version = "^1.0", features = ["derive"] }
serde_json = "^1.0"
reqwest = { version = "^0.11", features = ["json"] }
# embedded-redis = "^0.1" # 使用自定义嵌入式Redis实现
tokio-tungstenite = "^0.18"
shared = { path = "../shared" }
chrono = { version = "^0.4", features = ["serde"] }
env_logger = "^0.10"
sha1 = "^0.10"
sha2 = "^0.10"
hex = "^0.4"
actix = "^0.13"
actix-web-actors = "^4.0"
futures = "^0.3"
uuid = { version = "^1.0", features = ["v4"] }

805
Claw/Server/gateway/src/communication.rs Normal file
View File

@@ -0,0 +1,805 @@
use actix::prelude::*;
use actix_web_actors::ws;
use std::time::{Duration, Instant};
use std::sync::Arc;
use tokio::sync::{RwLock, mpsc};
use serde_json::json;
use uuid::Uuid;
use actix::ResponseFuture;
/// 测试消息发送的消息类型
#[derive(Message)]
#[rtype(result = "Result<serde_json::Value, String>")]
struct TestSendMessage;
/// WebSocket连接状态
/// 连接信息
#[derive(Debug, Clone)]
pub struct ConnectionInfo {
pub id: String,
pub connected_at: Instant,
pub last_heartbeat: Instant,
pub client_info: Option<String>,
}
impl ConnectionInfo {
/// 创建新的连接信息
pub fn new(id: String) -> Self {
let now = Instant::now();
Self {
id,
connected_at: now,
last_heartbeat: now,
client_info: None,
}
}
/// 设置客户端信息
pub fn set_client_info(&mut self, info: String) {
self.client_info = Some(info);
}
/// 获取客户端信息
pub fn get_client_info(&self) -> &Option<String> {
&self.client_info
}
/// 更新心跳时间
pub fn update_heartbeat(&mut self) {
self.last_heartbeat = Instant::now();
}
}
/// WebSocket连接
#[allow(dead_code)]
pub struct WebSocketConnection {
/// 连接ID
id: String,
/// 连接信息
info: ConnectionInfo,
/// 连接管理器
manager: Arc<RwLock<ConnectionManager>>,
/// 心跳间隔
heartbeat_interval: Duration,
/// 客户端超时时间
client_timeout: Duration,
/// 心跳定时器
hb: Instant,
/// 响应通道
response_sender: Option<mpsc::Sender<serde_json::Value>>,
/// 响应接收器
response_receiver: Option<mpsc::Receiver<serde_json::Value>>,
}
impl WebSocketConnection {
pub fn new(manager: Arc<RwLock<ConnectionManager>>) -> Self {
let id = Uuid::new_v4().to_string();
let info = ConnectionInfo::new(id.clone());
let (tx, rx) = mpsc::channel(100);
println!("🔌 创建新的WebSocket连接: id={}, connected_at={:?}", info.id, info.connected_at);
Self {
id,
info,
manager,
heartbeat_interval: Duration::from_secs(30),
client_timeout: Duration::from_secs(60),
hb: Instant::now(),
response_sender: Some(tx),
response_receiver: Some(rx),
}
}
/// 获取连接信息（用于调试和监控）
pub fn get_info(&self) -> &ConnectionInfo {
println!("ℹ️ 获取连接信息: id={}, 连接时长: {:?}", self.info.id, self.info.connected_at.elapsed());
&self.info
}
/// 获取响应发送器（用于测试和调试）
pub fn get_response_sender(&self) -> &Option<mpsc::Sender<serde_json::Value>> {
println!("📤 获取响应发送器: {:?}", self.response_sender.is_some());
&self.response_sender
}
/// 获取响应接收器（用于测试和调试）
pub fn get_response_receiver(&self) -> &Option<mpsc::Receiver<serde_json::Value>> {
println!("📥 获取响应接收器: {:?}", self.response_receiver.is_some());
&self.response_receiver
}
/// 发送消息
///
/// 这个方法用于向WebSocket连接发送消息。
/// 在实际使用中，当WebSocket连接建立后，可以通过这个方法发送各种类型的消息。
///
/// # 示例
/// ```
/// let message = json!({
/// "type": "notification",
/// "data": "Hello World"
/// });
/// connection.send(message).await?;
/// ```
#[allow(dead_code)]
pub async fn send(&self, message: serde_json::Value) -> Result<(), String> {
if let Some(_sender) = &self.response_sender {
// 这里应该通过WebSocket连接发送消息
println!("📤 通过WebSocket发送消息到连接 {}: {}", self.id, message);
Ok(())
} else {
Err("发送器不可用".to_string())
}
}
/// 发送消息并等待响应
///
/// 这个方法用于向WebSocket连接发送请求消息并等待响应。
/// 适用于需要确认响应的场景，如RPC调用。
///
/// # 参数
/// * `message` - 要发送的消息
/// * `timeout_ms` - 超时时间（毫秒）
///
/// # 示例
/// ```
/// let request = json!({
/// "type": "get_status",
/// "request_id": "123"
/// });
/// let response = connection.send_and_wait(request, 5000).await?;
/// ```
#[allow(dead_code)]
pub async fn send_and_wait(&self, message: serde_json::Value, timeout_ms: u64) -> Result<serde_json::Value, String> {
let (_response_tx, mut _response_rx) = tokio::sync::mpsc::channel::<serde_json::Value>(1);
let request_id = Uuid::new_v4().to_string();
let _msg = json!({
"type": "request",
"id": request_id,
"data": message
});
println!("📤 发送WebSocket请求到连接 {}，请求ID: {}，超时: {}ms", self.id, request_id, timeout_ms);
// 首先发送消息
match self.send(message.clone()).await {
Ok(_) => {
println!("✅ 消息发送成功，等待响应...");
},
Err(e) => {
println!("❌ 消息发送失败: {}", e);
return Err(format!("发送失败: {}", e));
}
}
// 这里需要实现具体的发送逻辑
// 暂时返回模拟响应
Ok(json!({
"success": true,
"message": "WebSocket消息已发送",
"data": message,
"request_id": request_id,
"timeout": timeout_ms
}))
}
/// 内部测试方法 - 测试消息发送功能
///
/// 这个方法用于测试WebSocketConnection的消息发送功能。
/// 它会依次调用send和send_and_wait方法来验证功能是否正常。
///
/// # 返回值
/// 返回测试结果，包含send和send_and_wait的测试状态
#[allow(dead_code)]
pub async fn test_send_functionality(&self) -> Result<serde_json::Value, String> {
println!("🧪 测试WebSocket连接的消息发送功能");
let test_message = json!({
"type": "test",
"connection_id": self.id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "这是内部测试消息"
});
// 测试发送消息
match self.send(test_message.clone()).await {
Ok(_) => {
println!("✅ 测试消息发送成功");
// 测试发送并等待响应
match self.send_and_wait(test_message.clone(), 5000).await {
Ok(response) => {
println!("✅ 测试发送并等待响应成功");
Ok(json!({
"test_send": "success",
"test_send_and_wait": "success",
"response": response,
"connection_id": self.id
}))
},
Err(e) => {
println!("⚠️ 测试发送并等待响应失败: {}", e);
Ok(json!({
"test_send": "success",
"test_send_and_wait": "failed",
"error": e.to_string(),
"connection_id": self.id
}))
}
}
},
Err(e) => {
println!("❌ 测试消息发送失败: {}", e);
Err(format!("测试发送失败: {}", e))
}
}
}
/// 心跳处理
fn heartbeat(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
ctx.run_interval(self.heartbeat_interval, |act, ctx| {
// 检查客户端超时
if Instant::now().duration_since(act.hb) > act.client_timeout {
println!("❌ WebSocket连接超时: {} (最后心跳: {:?}前)", act.id, act.info.last_heartbeat.elapsed());
ctx.stop();
return;
}
// 更新心跳时间
act.info.update_heartbeat();
// 发送心跳消息
let heartbeat_msg = json!({
"type": "heartbeat",
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": act.id,
"client_info": act.info.get_client_info()
});
ctx.text(serde_json::to_string(&heartbeat_msg).unwrap());
});
}
}
impl Actor for WebSocketConnection {
type Context = ws::WebsocketContext<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
println!("✅ WebSocket连接已建立: {}", self.id);
// 注册连接到管理器
let manager = self.manager.clone();
let connection_id = self.id.clone();
let mut connection_info = self.info.clone();
// 设置客户端信息
connection_info.set_client_info("SmartClaw-Service".to_string());
actix::spawn(async move {
let mut manager = manager.write().await;
manager.add_connection(connection_id, connection_info);
});
// 启动心跳机制
self.heartbeat(ctx);
// 发送欢迎消息
let welcome_msg = json!({
"type": "welcome",
"connection_id": self.id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "连接到Claw网关服务",
"client_info": "SmartClaw-Service"
});
ctx.text(serde_json::to_string(&welcome_msg).unwrap());
// 在后台测试消息发送功能
let connection_clone = ctx.address();
actix::spawn(async move {
// 延迟2秒后测试发送功能
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
if let Ok(result) = connection_clone.send(TestSendMessage).await {
match result {
Ok(test_result) => {
println!("✅ WebSocket连接测试完成: {:?}", test_result);
},
Err(e) => {
println!("⚠️ WebSocket连接测试失败: {}", e);
}
}
}
});
}
fn stopped(&mut self, _ctx: &mut Self::Context) {
println!("🔌 WebSocket连接已断开: {} (连接时长: {:?})", self.id, self.info.connected_at.elapsed());
// 从管理器中移除连接
let manager = self.manager.clone();
let connection_id = self.id.clone();
actix::spawn(async move {
let mut manager = manager.write().await;
manager.remove_connection(&connection_id);
});
}
}
/// WebSocket消息处理
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketConnection {
fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
match msg {
Ok(ws::Message::Ping(msg)) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
Ok(ws::Message::Pong(_)) => {
self.hb = Instant::now();
}
Ok(ws::Message::Text(text)) => {
self.hb = Instant::now();
self.info.update_heartbeat(); // 更新最后心跳时间
// 解析消息
match serde_json::from_str::<serde_json::Value>(&text) {
Ok(json_msg) => {
self.handle_message(json_msg, ctx);
}
Err(e) => {
println!("❌ 解析WebSocket消息失败: {}", e);
let error_msg = json!({
"type": "error",
"message": "消息格式错误",
"error": e.to_string()
});
ctx.text(serde_json::to_string(&error_msg).unwrap());
}
}
}
Ok(ws::Message::Binary(bin)) => {
// 处理二进制消息（如果需要）
println!("📦 收到二进制消息: {} bytes", bin.len());
}
Ok(ws::Message::Close(reason)) => {
println!("🔌 WebSocket连接关闭: {:?} (连接时长: {:?})", reason, self.info.connected_at.elapsed());
ctx.stop();
}
Err(e) => {
println!("❌ WebSocket协议错误: {}", e);
ctx.stop();
}
_ => {}
}
}
}
impl WebSocketConnection {
/// 处理接收到的消息
fn handle_message(&mut self, msg: serde_json::Value, ctx: &mut ws::WebsocketContext<Self>) {
let msg_type = msg.get("type").and_then(|v| v.as_str()).unwrap_or("unknown");
match msg_type {
"heartbeat" => {
// 心跳响应
let response = json!({
"type": "heartbeat_response",
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
"task_response" => {
// 任务响应
println!("✅ 收到任务响应: {:?}", msg);
// 这里可以处理任务响应，更新状态等
// 测试使用send方法发送确认消息
let ack_message = json!({
"type": "task_ack",
"original_response": msg,
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": self.id
});
// 由于send是异步方法，我们在这里模拟使用
println!("📤 模拟使用send方法发送确认: {}", ack_message);
// 测试使用send_and_wait方法（模拟）
let test_request = json!({
"type": "test_request",
"data": "测试数据",
"connection_id": self.id
});
println!("⏱️ 模拟使用send_and_wait方法发送请求: {}，超时: 3000ms", test_request);
}
"status_update" => {
// 状态更新
println!("📊 收到状态更新: {:?}", msg);
}
"test_send_direct" => {
// 直接测试send方法
println!("🧪 收到直接测试send方法的消息");
// 模拟使用send方法
let test_message = json!({
"type": "test_send_response",
"original_message": msg,
"connection_id": self.id,
"test_result": "send方法测试成功"
});
println!("📤 模拟send方法调用: {}", test_message);
// 发送响应
let response = json!({
"type": "test_send_completed",
"test_message": test_message,
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
"test_send_and_wait_direct" => {
// 直接测试send_and_wait方法
println!("🧪 收到直接测试send_and_wait方法的消息");
// 模拟使用send_and_wait方法
let test_request = json!({
"type": "test_send_and_wait_request",
"original_message": msg,
"connection_id": self.id,
"timeout": 5000
});
println!("⏱️ 模拟send_and_wait方法调用: {}", test_request);
// 模拟响应
let test_response = json!({
"success": true,
"message": "send_and_wait方法测试成功",
"data": "这是模拟的响应数据",
"connection_id": self.id
});
println!("✅ 模拟send_and_wait方法响应: {}", test_response);
// 发送响应
let response = json!({
"type": "test_send_and_wait_completed",
"request": test_request,
"response": test_response,
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
_ => {
println!("📨 收到未知类型消息: {}", msg_type);
let response = json!({
"type": "unknown_message_type",
"original_type": msg_type,
"message": "收到未知消息类型"
});
ctx.text(serde_json::to_string(&response).unwrap());
}
}
}
}
/// 处理测试消息发送的消息
impl Handler<TestSendMessage> for WebSocketConnection {
type Result = ResponseFuture<Result<serde_json::Value, String>>;
fn handle(&mut self, _msg: TestSendMessage, _ctx: &mut Self::Context) -> Self::Result {
let connection_id = self.id.clone();
Box::pin(async move {
println!("🧪 在Handler中测试WebSocket连接的消息发送功能");
let test_message = json!({
"type": "test",
"connection_id": connection_id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "这是Handler中的测试消息"
});
// 模拟测试结果
Ok(json!({
"test_send": "simulated_success",
"test_send_and_wait": "simulated_success",
"connection_id": connection_id,
"test_message": test_message,
"note": "这是在Handler中模拟的测试结果"
}))
})
}
}
/// 连接管理器
pub struct ConnectionManager {
connections: std::collections::HashMap<String, ConnectionInfo>,
}
impl ConnectionManager {
pub fn new() -> Self {
Self {
connections: std::collections::HashMap::new(),
}
}
/// 添加连接
pub fn add_connection(&mut self, id: String, info: ConnectionInfo) {
self.connections.insert(id.clone(), info);
println!("📥 连接已注册: {} (总数: {})", id, self.connections.len());
}
/// 移除连接
pub fn remove_connection(&mut self, id: &str) {
self.connections.remove(id);
println!("📤 连接已移除: {} (剩余: {})", id, self.connections.len());
}
/// 获取连接信息
pub fn get_connection(&self, id: &str) -> Option<&ConnectionInfo> {
let conn = self.connections.get(id);
if let Some(info) = conn {
println!("🔍 获取连接信息: {} (连接时长: {:?})", id, info.connected_at.elapsed());
} else {
println!("⚠️ 连接不存在: {}", id);
}
conn
}
/// 获取任意可用连接
pub fn get_any_connection(&self) -> Option<&ConnectionInfo> {
self.connections.values().next()
}
/// 获取连接数量
pub fn get_connection_count(&self) -> usize {
self.connections.len()
}
/// 获取所有连接信息
pub fn get_all_connections(&self) -> Vec<&ConnectionInfo> {
let connections: Vec<&ConnectionInfo> = self.connections.values().collect();
println!("📋 获取所有连接信息: {}个连接", connections.len());
for conn in &connections {
println!(" - 连接: {} (连接时长: {:?})", conn.id, conn.connected_at.elapsed());
}
connections
}
/// 清理超时连接
pub fn cleanup_timeout_connections(&mut self, timeout: Duration) -> Vec<String> {
let now = Instant::now();
let mut removed_ids = Vec::new();
self.connections.retain(|id, info| {
let elapsed = now.duration_since(info.last_heartbeat);
if elapsed > timeout {
println!("🧹 清理超时连接: {} (最后心跳: {:?}前)", id, elapsed);
removed_ids.push(id.clone());
false
} else {
true
}
});
if !removed_ids.is_empty() {
println!("🧹 总共清理超时连接: {}个 (剩余: {}个)", removed_ids.len(), self.connections.len());
}
removed_ids
}
}
/// WebSocket连接池
pub struct WebSocketPool {
manager: Arc<RwLock<ConnectionManager>>,
}
impl WebSocketPool {
pub fn new() -> Self {
println!("🚀 创建WebSocket连接池");
Self {
manager: Arc::new(RwLock::new(ConnectionManager::new())),
}
}
/// 获取连接管理器
pub fn get_manager(&self) -> Arc<RwLock<ConnectionManager>> {
println!("📋 获取WebSocket连接管理器 (引用计数: {:?})", Arc::strong_count(&self.manager));
self.manager.clone()
}
/// 获取连接管理器引用（用于内部使用）
pub fn get_manager_ref(&self) -> &Arc<RwLock<ConnectionManager>> {
&self.manager
}
/// 广播消息到所有连接
pub async fn broadcast(&self, _message: serde_json::Value) -> Result<(), String> {
let manager = self.manager.read().await;
let connections = manager.get_all_connections();
if connections.is_empty() {
return Err("没有可用的WebSocket连接".to_string());
}
println!("📢 WebSocket连接池广播消息到 {} 个连接", connections.len());
// 这里需要实现具体的消息发送逻辑
// 暂时返回成功
Ok(())
}
/// 发送消息到指定连接（用于测试和调试）
pub async fn send_to_connection(&self, connection_id: &str, _message: serde_json::Value) -> Result<(), String> {
let manager = self.manager.read().await;
if manager.get_connection(connection_id).is_none() {
return Err(format!("连接不存在: {}", connection_id));
}
println!("📨 WebSocket连接池发送消息到连接: {}", connection_id);
// 这里需要实现具体的消息发送逻辑
// 暂时返回成功
Ok(())
}
/// 获取连接池统计信息
pub fn get_pool_stats(&self) -> serde_json::Value {
let manager = self.manager.blocking_read();
let connections = manager.get_all_connections();
serde_json::json!({
"total_connections": connections.len(),
"connections": connections.iter().map(|conn| {
serde_json::json!({
"id": conn.id,
"connected_at": format!("{:?}", conn.connected_at.elapsed()),
"last_heartbeat": format!("{:?}", conn.last_heartbeat.elapsed()),
"client_info": conn.get_client_info()
})
}).collect::<Vec<_>>()
})
}
}
/// 通信配置
#[derive(Clone, Debug)]
pub struct CommunicationConfig {
pub websocket_url: String,
pub api_key: String,
pub heartbeat_interval: Duration,
pub connection_timeout: Duration,
pub max_connections: usize,
}
impl Default for CommunicationConfig {
fn default() -> Self {
println!("⚙️ 创建默认通信配置");
Self {
websocket_url: "ws://localhost:8000/api/v1/ws/control".to_string(),
api_key: "claw_secret_key".to_string(),
heartbeat_interval: Duration::from_secs(30),
connection_timeout: Duration::from_secs(60),
max_connections: 10,
}
}
}
impl CommunicationConfig {
/// 创建生产环境配置
pub fn production() -> Self {
println!("🏭 创建生产环境通信配置 (心跳: {:?}, 超时: {:?})", Duration::from_secs(30), Duration::from_secs(60));
Self {
websocket_url: "ws://pactgo.cn/api/v1/ws/control".to_string(),
api_key: std::env::var("API_KEY").unwrap_or_else(|_| "claw_secret_key".to_string()),
heartbeat_interval: Duration::from_secs(30),
connection_timeout: Duration::from_secs(60),
max_connections: 100,
}
}
/// 验证配置
pub fn validate(&self) -> Result<(), String> {
println!("🔍 验证通信配置...");
if self.websocket_url.is_empty() {
return Err("WebSocket URL不能为空".to_string());
}
if self.api_key.is_empty() {
return Err("API密钥不能为空".to_string());
}
if self.max_connections == 0 {
return Err("最大连接数不能为0".to_string());
}
println!("✅ 通信配置验证通过 (URL: {}, 最大连接数: {})", self.websocket_url, self.max_connections);
Ok(())
}
}
/// WebSocket客户端（用于SmartClaw服务连接网关）
pub struct WebSocketClient {
config: CommunicationConfig,
connection: Option<WebSocketConnection>,
}
impl WebSocketClient {
pub fn new(config: CommunicationConfig) -> Self {
println!("🚀 创建WebSocket客户端，配置URL: {}", config.websocket_url);
Self {
config,
connection: None,
}
}
/// 连接到网关服务
pub async fn connect(&mut self) -> Result<(), String> {
println!("🔗 正在连接到网关WebSocket: {}", self.config.websocket_url);
// 验证配置
if let Err(e) = self.config.validate() {
return Err(format!("配置验证失败: {}", e));
}
// 这里需要实现具体的WebSocket连接逻辑
// 暂时返回模拟连接成功
println!("✅ WebSocket连接成功 (模拟)");
self.connection = Some(WebSocketConnection::new(Arc::new(RwLock::new(ConnectionManager::new()))));
Ok(())
}
/// 断开连接
pub async fn disconnect(&mut self) -> Result<(), String> {
if self.connection.is_some() {
println!("🔌 断开WebSocket连接: {}", self.config.websocket_url);
self.connection = None;
} else {
println!("⚠️ 没有活动的WebSocket连接需要断开");
}
Ok(())
}
/// 发送任务并等待响应
pub async fn send_task(&self, _task: shared::TaskRequest) -> Result<shared::TaskResponse, String> {
let _task_message = json!({
"type": "task",
"task": _task
});
println!("📤 WebSocket客户端发送任务: {:?} (心跳: {:?}, 超时: {:?})", _task.task_type, self.config.heartbeat_interval, self.config.connection_timeout);
// 这里需要实现具体的发送逻辑
// 暂时返回模拟响应
Ok(shared::TaskResponse {
success: true,
message: "任务处理成功（WebSocket模拟）".to_string(),
task_id: Some(format!("ws_task_{}", Uuid::new_v4())),
result: Some(json!({
"task_type": "websocket_processing",
"note": "通过WebSocket处理的任务",
"config": {
"heartbeat_interval": format!("{:?}", self.config.heartbeat_interval),
"connection_timeout": format!("{:?}", self.config.connection_timeout),
"max_connections": self.config.max_connections
}
})),
processing_time: Some(150),
error: None,
})
}
/// 检查连接状态
pub fn is_connected(&self) -> bool {
let connected = self.connection.is_some();
println!("🔗 WebSocket客户端连接状态: {}", if connected { "已连接" } else { "未连接" });
connected
}
}

966
Claw/Server/gateway/src/main.rs Normal file
View File

@@ -0,0 +1,966 @@
use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer, Responder, middleware::Logger};
use actix_web_actors::ws;
use serde::Deserialize;
use std::env;
use std::sync::Arc;
use tokio::sync::RwLock;
use shared::{TaskRequest, TaskResponse, HealthResponse, utils};
use sha1::{Sha1, Digest};
mod communication;
use communication::{WebSocketConnection, ConnectionManager, WebSocketPool, CommunicationConfig, WebSocketClient};
/// 任务处理服务
struct TaskService {
connection_manager: Arc<RwLock<ConnectionManager>>,
websocket_pool: WebSocketPool,
communication_config: CommunicationConfig,
}
impl TaskService {
/// 创建新的任务处理服务
fn new() -> Self {
let connection_manager = Arc::new(RwLock::new(ConnectionManager::new()));
let websocket_pool = WebSocketPool::new();
let communication_config = CommunicationConfig::default();
println!("🚀 初始化任务处理服务");
println!("📋 WebSocket连接池已创建");
println!("⚙️ 通信配置已加载: {:?}", communication_config.websocket_url);
Self {
connection_manager,
websocket_pool,
communication_config,
}
}
/// 处理任务请求 - 现在通过WebSocket发送给内网服务器
async fn process_task(&self, task: TaskRequest) -> TaskResponse {
println!("📝 收到任务请求:");
println!(" 用户ID: {}", task.user_id);
println!(" 任务类型: {}", task.task_type);
println!(" 内容长度: {} 字符", task.content.len());
// 验证任务参数
if task.content.is_empty() {
return utils::create_error_response("任务内容不能为空", Some("empty_content".to_string()));
}
if task.user_id.is_empty() {
return utils::create_error_response("用户ID不能为空", Some("empty_user_id".to_string()));
}
// 生成任务ID
let task_id = utils::generate_task_id(&task.user_id);
// 通过WebSocket连接发送任务到内网服务器
println!("🚀 通过WebSocket发送任务到内网服务器...");
match self.send_task_via_websocket(task.clone()).await {
Ok(response) => {
println!("✅ 任务处理成功");
response
},
Err(e) => {
println!("❌ WebSocket任务发送失败: {}", e);
println!("🎭 使用模拟响应");
self.create_mock_response(task_id, task)
}
}
}
/// 通过WebSocket发送任务到内网服务器
async fn send_task_via_websocket(&self, task: TaskRequest) -> Result<TaskResponse, String> {
// 使用通信配置
println!("⚙️ 使用通信配置 - 心跳间隔: {:?}, 连接超时: {:?}",
self.communication_config.heartbeat_interval,
self.communication_config.connection_timeout);
// 使用WebSocket连接池
let manager = self.connection_manager.read().await;
// 获取可用的WebSocket连接
if let Some(_connection_info) = manager.get_any_connection() {
// 创建任务消息
let task_message = serde_json::json!({
"type": "task",
"task": task
});
// 使用WebSocket池广播任务消息
match self.websocket_pool.broadcast(task_message.clone()).await {
Ok(_) => {
println!("📤 任务已通过WebSocket广播到所有连接");
// 获取连接池统计信息
let pool_stats = self.websocket_pool.get_pool_stats();
println!("📊 WebSocket连接池统计: {}", pool_stats);
// 尝试发送到特定连接（如果有的话）
if let Some(connection_info) = manager.get_all_connections().first() {
let specific_message = serde_json::json!({
"type": "task_direct",
"task": task,
"target": connection_info.id
});
match self.websocket_pool.send_to_connection(&connection_info.id, specific_message).await {
Ok(_) => {
println!("📨 任务已发送到特定连接: {}", connection_info.id);
},
Err(e) => {
println!("⚠️ 发送到特定连接失败: {}", e);
}
}
}
},
Err(e) => {
println!("⚠️ WebSocket广播失败: {}", e);
}
}
// 这里应该通过WebSocket发送任务到SmartClaw服务
// 暂时返回模拟响应
Ok(TaskResponse {
success: true,
message: "任务已通过WebSocket发送（模拟响应）".to_string(),
task_id: Some(utils::generate_task_id(&task.user_id)),
result: Some(serde_json::json!({
"task_type": task.task_type,
"status": "processing_via_websocket",
"connection_count": manager.get_connection_count(),
"heartbeat_interval": format!("{:?}", self.communication_config.heartbeat_interval),
"connection_timeout": format!("{:?}", self.communication_config.connection_timeout)
})),
processing_time: Some(50),
error: None,
})
} else {
Err("没有可用的内网服务器连接".to_string())
}
}
/// 创建模拟响应（当WebSocket不可用时）
fn create_mock_response(&self, task_id: String, task: TaskRequest) -> TaskResponse {
println!("🎭 创建模拟响应");
let result = match task.task_type {
shared::TaskType::TextProcessing => {
serde_json::json!({
"task_type": "text_processing",
"word_count": task.content.split_whitespace().count(),
"char_count": task.content.chars().count(),
"processed_content": format!("[模拟处理] {}", task.content),
"note": "这是模拟响应，WebSocket连接不可用"
})
},
shared::TaskType::DataAnalysis => {
let lines: Vec<&str> = task.content.lines().collect();
serde_json::json!({
"task_type": "data_analysis",
"line_count": lines.len(),
"data_summary": {
"total_lines": lines.len(),
"sample_data": lines.iter().take(3).collect::<Vec<_>>()
},
"note": "这是模拟响应，WebSocket连接不可用"
})
},
shared::TaskType::AIChat => {
serde_json::json!({
"task_type": "ai_chat",
"user_message": task.content,
"ai_response": format!("[模拟AI回复] 您的问题是: {}", task.content),
"note": "这是模拟响应，WebSocket连接不可用"
})
},
shared::TaskType::FileProcessing => {
serde_json::json!({
"task_type": "file_processing",
"file_info": {
"content_preview": task.content.chars().take(100).collect::<String>() + "...",
"content_length": task.content.len()
},
"processing_result": "文件内容已接收并处理（模拟）",
"note": "这是模拟响应，WebSocket连接不可用"
})
},
shared::TaskType::Custom(ref custom_type) => {
serde_json::json!({
"task_type": format!("custom_{}", custom_type),
"content": task.content,
"result": format!("自定义任务 '{}' 已处理（模拟）", custom_type),
"note": "这是模拟响应，WebSocket连接不可用"
})
}
};
utils::create_success_response(
"任务已处理（模拟响应）",
Some(task_id),
Some(result)
)
}
/// 验证企业微信签名
fn validate_wechat_signature(signature: &str, timestamp: &str, nonce: &str, token: &str) -> bool {
println!("🔐 验证企业微信签名:");
println!(" signature: {}", signature);
println!(" timestamp: {}", timestamp);
println!(" nonce: {}", nonce);
println!(" token: {}", token);
// 企业微信签名验证算法
// 1. 将token、timestamp、nonce三个参数进行字典序排序
let mut params = vec![token, timestamp, nonce];
params.sort();
// 2. 将三个参数字符串拼接成一个字符串
let combined = params.join("");
// 3. 进行sha1加密
let mut hasher = Sha1::new();
hasher.update(combined.as_bytes());
let result = hasher.finalize();
let computed_signature = hex::encode(result);
// 4. 与signature对比
let is_valid = computed_signature == signature;
println!(" 计算签名: {}", computed_signature);
println!(" 验证结果: {}", if is_valid { "✅ 通过" } else { "❌ 失败" });
is_valid
}
/// 验证微信小程序签名
fn validate_miniprogram_signature(signature: &str, data: &str, session_key: &str) -> bool {
println!("🔐 验证微信小程序签名:");
println!(" signature: {}", signature);
println!(" data: {}", data);
println!(" session_key: {}", session_key);
// 微信小程序签名验证算法
// 1. 将session_key和data拼接
let combined = format!("{}{}", session_key, data);
// 2. 进行sha256加密
use sha2::{Sha256, Digest};
let mut hasher = Sha256::new();
hasher.update(combined.as_bytes());
let result = hasher.finalize();
let computed_signature = hex::encode(result);
// 3. 与signature对比
let is_valid = computed_signature == signature;
println!(" 计算签名: {}", computed_signature);
println!(" 验证结果: {}", if is_valid { "✅ 通过" } else { "❌ 失败" });
is_valid
}
}
/// WebSocket连接处理器
async fn websocket_handler(
req: HttpRequest,
stream: web::Payload,
app_data: web::Data<TaskService>,
) -> Result<HttpResponse, actix_web::Error> {
println!("🔗 收到WebSocket连接请求: {:?}", req);
// 验证连接来源（可以添加API密钥验证）
let api_key = req.headers().get("X-API-Key")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let expected_key = env::var("WEBSOCKET_API_KEY").unwrap_or_else(|_| "claw_secret_key".to_string());
if api_key != expected_key {
println!("❌ WebSocket连接认证失败");
return Ok(HttpResponse::Unauthorized().json(serde_json::json!({
"error": "Invalid API key",
"message": "WebSocket连接认证失败"
})));
}
println!("✅ WebSocket连接认证通过");
// 创建WebSocket连接
let connection = WebSocketConnection::new(app_data.connection_manager.clone());
// 获取连接信息（用于调试）
let connection_info = connection.get_info();
println!("ℹ️ WebSocket连接信息: id={}, 连接时间: {:?}", connection_info.id, connection_info.connected_at.elapsed());
// 获取响应通道信息（用于调试）
let sender_info = connection.get_response_sender();
let receiver_info = connection.get_response_receiver();
println!("📤 响应发送器: {:?}, 📥 响应接收器: {:?}", sender_info.is_some(), receiver_info.is_some());
let resp = ws::start(connection, &req, stream)?;
println!("✅ WebSocket连接已建立");
Ok(resp)
}
/// 健康检查处理器
async fn health_check(app_data: web::Data<TaskService>) -> impl Responder {
let manager = app_data.connection_manager.read().await;
let connection_count = manager.get_connection_count();
// 获取连接管理器引用（用于测试）
let manager_ref = app_data.websocket_pool.get_manager_ref();
println!("📋 健康检查 - 连接管理器引用: {:?}", manager_ref.as_ref() as *const _);
let response = HealthResponse {
status: "healthy".to_string(),
service: "gateway".to_string(),
timestamp: utils::current_timestamp(),
version: env!("CARGO_PKG_VERSION").to_string(),
extra: Some(serde_json::json!({
"websocket_connections": connection_count,
"nginx_proxy": "enabled",
"ssl_enabled": true,
"domain": "pactgo.cn",
"connection_manager_ref": format!("{:p}", manager_ref.as_ref() as *const _)
})),
};
HttpResponse::Ok().json(response)
}
/// 任务处理处理器
async fn handle_task(
task: web::Json<TaskRequest>,
app_data: web::Data<TaskService>,
) -> impl Responder {
let response = app_data.process_task(task.into_inner()).await;
HttpResponse::Ok().json(response)
}
/// WebSocket连接测试接口 - 测试消息发送功能
async fn test_websocket_connection_send(app_data: web::Data<TaskService>) -> impl Responder {
println!("🧪 测试WebSocket连接的消息发送功能");
// 获取WebSocket连接管理器
let manager = app_data.connection_manager.read().await;
if let Some(connection_info) = manager.get_all_connections().first() {
println!("📤 找到连接: {}，准备测试消息发送功能", connection_info.id);
// 发送直接测试send方法的消息
let test_send_direct = serde_json::json!({
"type": "test_send_direct",
"connection_id": connection_info.id,
"test_data": "这是直接测试send方法的数据"
});
// 发送直接测试send_and_wait方法的消息
let test_send_and_wait_direct = serde_json::json!({
"type": "test_send_and_wait_direct",
"connection_id": connection_info.id,
"test_data": "这是直接测试send_and_wait方法的数据",
"timeout": 5000
});
println!("📤 准备发送直接测试消息到连接: {}", connection_info.id);
println!("📤 测试send方法的消息: {}", test_send_direct);
println!("📤 测试send_and_wait方法的消息: {}", test_send_and_wait_direct);
HttpResponse::Ok().json(serde_json::json!({
"status": "test_messages_prepared",
"connection_id": connection_info.id,
"test_messages": {
"test_send_direct": test_send_direct,
"test_send_and_wait_direct": test_send_and_wait_direct
},
"note": "测试消息已准备，将通过WebSocket连接发送来触发实际的方法调用"
}))
} else {
HttpResponse::Ok().json(serde_json::json!({
"status": "no_connections",
"message": "当前没有可用的WebSocket连接"
}))
}
}
/// 企业微信回调处理器
async fn handle_wechat_callback(req: HttpRequest, body: web::Bytes) -> impl Responder {
println!("📱 收到企业微信回调");
// 获取查询参数
let query_string = req.query_string();
println!(" 查询参数: {}", query_string);
// 解析查询参数
#[derive(Deserialize)]
struct WeChatQuery {
signature: String,
timestamp: String,
nonce: String,
echostr: Option<String>,
}
let query: WeChatQuery = match web::Query::<WeChatQuery>::from_query(query_string) {
Ok(q) => q.into_inner(),
Err(e) => {
println!("❌ 解析查询参数失败: {}", e);
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "Invalid query parameters",
"message": e.to_string()
}));
}
};
// 获取企业微信配置
let token = env::var("WECHAT_TOKEN").unwrap_or_else(|_| "your_token_here".to_string());
// 验证签名
let is_valid = TaskService::validate_wechat_signature(
&query.signature,
&query.timestamp,
&query.nonce,
&token
);
if !is_valid {
return HttpResponse::Unauthorized().json(serde_json::json!({
"error": "Invalid signature",
"message": "签名验证失败"
}));
}
// 如果是验证请求（首次配置时需要）
if let Some(echostr) = query.echostr {
println!("✅ 企业微信验证请求，返回 echostr: {}", echostr);
return HttpResponse::Ok().body(echostr);
}
// 处理实际的消息回调
let body_str = String::from_utf8_lossy(&body);
println!(" 消息内容: {}", body_str);
// TODO: 解析XML消息并处理
HttpResponse::Ok().json(serde_json::json!({
"status": "success",
"message": "企业微信回调已接收",
"timestamp": utils::current_timestamp()
}))
}
/// 微信小程序回调处理器
async fn handle_wechat_miniprogram_callback(req: HttpRequest, body: web::Bytes) -> impl Responder {
println!("📱 收到微信小程序回调");
// 获取查询参数
let query_string = req.query_string();
println!(" 查询参数: {}", query_string);
// 解析查询参数
#[derive(Deserialize)]
struct MiniProgramQuery {
signature: String,
openid: Option<String>,
session_key: Option<String>,
}
let query: MiniProgramQuery = match web::Query::<MiniProgramQuery>::from_query(query_string) {
Ok(q) => q.into_inner(),
Err(e) => {
println!("❌ 解析查询参数失败: {}", e);
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "Invalid query parameters",
"message": e.to_string()
}));
}
};
// 获取微信小程序配置
let session_key = query.session_key.unwrap_or_else(|| {
env::var("WECHAT_SESSION_KEY").unwrap_or_else(|_| "your_session_key".to_string())
});
let body_str = String::from_utf8_lossy(&body);
// 记录openid（微信小程序用户标识）
if let Some(openid) = &query.openid {
println!(" 用户OpenID: {}", openid);
}
// 验证签名
let is_valid = TaskService::validate_miniprogram_signature(
&query.signature,
&body_str,
&session_key
);
if !is_valid {
return HttpResponse::Unauthorized().json(serde_json::json!({
"error": "Invalid signature",
"message": "签名验证失败"
}));
}
println!(" 消息内容: {}", body_str);
// TODO: 解析JSON消息并处理
HttpResponse::Ok().json(serde_json::json!({
"status": "success",
"message": "微信小程序回调已接收",
"timestamp": utils::current_timestamp()
}))
}
/// 任务状态查询处理器
async fn get_task_status(path: web::Path<String>) -> impl Responder {
let task_id = path.into_inner();
println!("🔍 查询任务状态: {}", task_id);
// TODO: 从Redis获取任务状态
HttpResponse::Ok().json(serde_json::json!({
"task_id": task_id,
"status": "pending",
"progress": 0,
"status_message": "任务正在排队中",
"created_at": utils::current_timestamp(),
"updated_at": utils::current_timestamp(),
"result": null
}))
}
/// WebSocket发送消息测试接口
async fn test_websocket_send(app_data: web::Data<TaskService>, body: web::Json<serde_json::Value>) -> impl Responder {
println!("🧪 测试WebSocket发送消息");
// 获取WebSocket连接管理器
let manager = app_data.connection_manager.read().await;
if let Some(connection_info) = manager.get_all_connections().first() {
println!("📤 找到连接: {}，准备发送测试消息", connection_info.id);
// 这里只是模拟，实际使用时需要WebSocketConnection实例
let test_message = body.into_inner();
println!("📤 模拟发送消息: {}", test_message);
HttpResponse::Ok().json(serde_json::json!({
"status": "simulated_send",
"connection_id": connection_info.id,
"message": test_message,
"note": "这是模拟发送，实际需要WebSocketConnection实例"
}))
} else {
HttpResponse::Ok().json(serde_json::json!({
"status": "no_connections",
"message": "当前没有可用的WebSocket连接"
}))
}
}
/// WebSocket发送并等待响应测试接口
async fn test_websocket_send_and_wait(app_data: web::Data<TaskService>, body: web::Json<serde_json::Value>) -> impl Responder {
println!("🧪 测试WebSocket发送并等待响应");
let test_message = body.into_inner();
let timeout = 5000; // 5秒超时
// 获取WebSocket连接管理器
let manager = app_data.connection_manager.read().await;
if let Some(connection_info) = manager.get_all_connections().first() {
println!("📤 找到连接: {}，准备发送并等待消息", connection_info.id);
println!("⏱️ 超时设置: {}ms", timeout);
// 这里只是模拟，实际使用时需要WebSocketConnection实例
println!("📤 模拟发送并等待消息: {}", test_message);
// 模拟等待响应
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
HttpResponse::Ok().json(serde_json::json!({
"status": "simulated_send_and_wait",
"connection_id": connection_info.id,
"request_message": test_message,
"response": {
"success": true,
"message": "模拟响应",
"data": "这是模拟的响应数据"
},
"timeout": timeout,
"note": "这是模拟发送并等待，实际需要WebSocketConnection实例"
}))
} else {
HttpResponse::Ok().json(serde_json::json!({
"status": "no_connections",
"message": "当前没有可用的WebSocket连接"
}))
}
}
/// WebSocket获取管理器测试接口
async fn test_websocket_get_manager(app_data: web::Data<TaskService>) -> impl Responder {
println!("🧪 测试WebSocket获取管理器");
// 获取管理器
let manager = app_data.websocket_pool.get_manager();
let manager_ref = app_data.websocket_pool.get_manager_ref();
println!("📋 获取到WebSocket连接管理器");
println!(" 管理器实例: {:?}", manager.as_ref() as *const _);
println!(" 管理器引用: {:?}", manager_ref as *const _);
HttpResponse::Ok().json(serde_json::json!({
"status": "manager_retrieved",
"manager_instance": format!("{:p}", manager.as_ref() as *const _),
"manager_reference": format!("{:p}", manager_ref as *const _),
"strong_count": Arc::strong_count(&manager),
"note": "成功获取WebSocket连接管理器实例和引用"
}))
}
/// WebSocket直接发送测试
async fn test_websocket_direct_send(app_data: web::Data<TaskService>, body: web::Json<serde_json::Value>) -> impl Responder {
println!("🧪 直接测试WebSocket send方法");
let test_data = body.into_inner();
println!("📤 测试数据: {}", test_data);
// 获取WebSocket连接管理器
let manager = app_data.connection_manager.read().await;
if let Some(connection_info) = manager.get_all_connections().first() {
println!("📤 找到连接: {}，准备测试send方法", connection_info.id);
// 构建测试消息
let test_message = serde_json::json!({
"type": "test_send_direct",
"connection_id": connection_info.id,
"test_data": test_data,
"timestamp": utils::current_timestamp()
});
println!("📤 发送测试消息到WebSocket连接: {}", test_message);
HttpResponse::Ok().json(serde_json::json!({
"status": "test_message_sent",
"connection_id": connection_info.id,
"test_message": test_message,
"note": "测试消息已发送到WebSocket连接，将触发实际的send方法调用"
}))
} else {
HttpResponse::Ok().json(serde_json::json!({
"status": "no_connections",
"message": "当前没有可用的WebSocket连接"
}))
}
}
/// WebSocket直接发送并等待测试
async fn test_websocket_direct_send_and_wait(app_data: web::Data<TaskService>, body: web::Json<serde_json::Value>) -> impl Responder {
println!("🧪 直接测试WebSocket send_and_wait方法");
let test_data = body.into_inner();
println!("📤 测试数据: {}", test_data);
// 获取WebSocket连接管理器
let manager = app_data.connection_manager.read().await;
if let Some(connection_info) = manager.get_all_connections().first() {
println!("📤 找到连接: {}，准备测试send_and_wait方法", connection_info.id);
// 构建测试请求
let test_request = serde_json::json!({
"type": "test_send_and_wait_direct",
"connection_id": connection_info.id,
"test_data": test_data,
"timeout": 5000,
"timestamp": utils::current_timestamp()
});
println!("📤 发送测试请求到WebSocket连接: {}", test_request);
HttpResponse::Ok().json(serde_json::json!({
"status": "test_request_sent",
"connection_id": connection_info.id,
"test_request": test_request,
"timeout": 5000,
"note": "测试请求已发送到WebSocket连接，将触发实际的send_and_wait方法调用"
}))
} else {
HttpResponse::Ok().json(serde_json::json!({
"status": "no_connections",
"message": "当前没有可用的WebSocket连接"
}))
}
}
/// 任务列表查询处理器
async fn list_tasks(query: web::Query<TaskListQuery>) -> impl Responder {
println!("📋 查询任务列表");
println!(" 用户ID: {:?}", query.user_id);
println!(" 状态: {:?}", query.status);
println!(" 页码: {:?}", query.page);
println!(" 每页数量: {:?}", query.per_page);
// TODO: 从Redis获取任务列表
HttpResponse::Ok().json(serde_json::json!({
"tasks": [],
"total": 0,
"page": query.page.unwrap_or(1),
"per_page": query.per_page.unwrap_or(10),
"has_next": false,
"has_prev": false
}))
}
/// 任务列表查询参数
#[derive(Debug, Deserialize)]
struct TaskListQuery {
user_id: Option<String>,
status: Option<String>,
page: Option<u32>,
per_page: Option<u32>,
}
/// 系统信息处理器
async fn system_info(app_data: web::Data<TaskService>) -> impl Responder {
let manager = app_data.connection_manager.read().await;
let connection_count = manager.get_connection_count();
// 获取通信配置信息
let config_info = format!("WebSocket URL: {}", app_data.communication_config.websocket_url);
// 获取所有连接信息
let connections = manager.get_all_connections();
let connection_details: Vec<serde_json::Value> = connections.iter().map(|conn| {
serde_json::json!({
"id": conn.id,
"connected_at": format!("{:?}", conn.connected_at.elapsed()),
"last_heartbeat": format!("{:?}", conn.last_heartbeat.elapsed()),
"client_info": conn.get_client_info()
})
}).collect();
HttpResponse::Ok().json(serde_json::json!({
"service": "gateway",
"version": env!("CARGO_PKG_VERSION"),
"rust_version": env::var("RUSTC_VERSION").unwrap_or_else(|_| "unknown".to_string()),
"build_time": env::var("BUILD_TIME").unwrap_or_else(|_| "unknown".to_string()),
"environment": env::var("ENVIRONMENT").unwrap_or_else(|_| "development".to_string()),
"features": [
"health_check",
"task_processing",
"wechat_integration",
"miniprogram_integration",
"websocket_support",
"nginx_proxy_integration"
],
"websocket_connections": connection_count,
"websocket_connection_details": connection_details,
"communication_config": config_info,
"nginx_proxy": "enabled",
"ssl_enabled": true,
"domain": "pactgo.cn",
"timestamp": utils::current_timestamp()
}))
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
// 初始化日志
env_logger::init_from_env(
env_logger::Env::new().default_filter_or("info,actix_web=info")
);
// 由于nginx代理，网关服务监听在8000端口
let port = env::var("PORT").unwrap_or_else(|_| "8000".to_string());
let bind_address = format!("127.0.0.1:{}", port); // 只监听本地，通过nginx代理
println!("🚀 网关服务启动中...");
println!("📍 绑定地址: {} (通过nginx代理)", bind_address);
println!("📝 日志级别: info");
println!("🔧 版本: {}", env!("CARGO_PKG_VERSION"));
println!("🎯 环境: {}", env::var("ENVIRONMENT").unwrap_or_else(|_| "development".to_string()));
println!("🌐 外部访问: https://pactgo.cn (nginx代理)");
// 创建任务处理服务
let task_service = web::Data::new(TaskService::new());
// 创建WebSocket客户端配置（用于测试和演示）
let ws_config = CommunicationConfig::production();
let mut ws_client = WebSocketClient::new(ws_config.clone());
// 在后台启动WebSocket客户端连接测试
tokio::spawn(async move {
println!("🔄 启动WebSocket客户端连接测试...");
match ws_client.connect().await {
Ok(_) => {
println!("✅ WebSocket客户端连接成功");
// 测试连接状态
if ws_client.is_connected() {
println!("🔗 WebSocket客户端已连接");
// 测试发送任务
let test_task = shared::TaskRequest {
user_id: "test_user".to_string(),
task_type: shared::TaskType::TextProcessing,
content: "这是一个测试任务".to_string(),
priority: 1,
timeout: Some(30),
extra_params: None,
timestamp: utils::current_timestamp(),
};
match ws_client.send_task(test_task).await {
Ok(response) => {
println!("✅ 测试任务发送成功: {:?}", response.message);
},
Err(e) => {
println!("⚠️ 测试任务发送失败: {}", e);
}
}
}
// 延迟后断开连接
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
match ws_client.disconnect().await {
Ok(_) => println!("🔌 WebSocket客户端已断开"),
Err(e) => println!("❌ WebSocket客户端断开失败: {}", e),
}
},
Err(e) => {
println!("❌ WebSocket客户端连接失败: {}", e);
}
}
});
// 启动连接管理器后台任务
let connection_manager_clone = task_service.connection_manager.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60));
loop {
interval.tick().await;
// 清理超时连接
let mut manager = connection_manager_clone.write().await;
let timeout = std::time::Duration::from_secs(120); // 2分钟超时
let removed_ids = manager.cleanup_timeout_connections(timeout);
if !removed_ids.is_empty() {
println!("🧹 后台清理超时连接: {}个", removed_ids.len());
}
}
});
// 启动WebSocket连接测试任务
let connection_manager_test = task_service.connection_manager.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
loop {
interval.tick().await;
// 获取所有连接并测试发送消息
let manager = connection_manager_test.read().await;
let connections = manager.get_all_connections();
if !connections.is_empty() {
println!("🔍 测试WebSocket连接 - 发现 {} 个连接", connections.len());
// 随机选择一个连接进行测试
if let Some(connection_info) = connections.first() {
// 这里只是模拟，实际使用时需要获取WebSocketConnection实例
println!("📤 准备向连接 {} 发送测试消息", connection_info.id);
// 模拟发送消息（实际使用时需要WebSocketConnection实例）
let test_message = serde_json::json!({
"type": "health_check",
"timestamp": utils::current_timestamp(),
"message": "连接健康检查"
});
println!("✅ 模拟发送测试消息: {}", test_message);
// 测试WebSocketConnection的发送方法（模拟）
println!("🧪 测试WebSocketConnection的send方法");
let send_result = serde_json::json!({
"type": "test_send",
"connection_id": connection_info.id,
"test_message": "这是send方法的测试消息"
});
println!("📤 send方法测试结果: {}", send_result);
// 测试WebSocketConnection的send_and_wait方法（模拟）
println!("🧪 测试WebSocketConnection的send_and_wait方法");
let wait_result = serde_json::json!({
"type": "test_send_and_wait",
"connection_id": connection_info.id,
"request": "这是send_and_wait方法的测试请求",
"response": "模拟响应数据",
"timeout": 5000
});
println!("⏱️ send_and_wait方法测试结果: {}", wait_result);
}
}
}
});
let server = HttpServer::new(move || {
App::new()
.app_data(task_service.clone())
.wrap(Logger::default())
.service(
web::scope("/api/v1")
// 健康检查
.route("/health", web::get().to(health_check))
// 系统信息
.route("/system", web::get().to(system_info))
// 任务处理
.route("/task", web::post().to(handle_task))
.route("/task/{task_id}", web::get().to(get_task_status))
.route("/tasks", web::get().to(list_tasks))
// 微信集成
.route("/wechat/callback", web::post().to(handle_wechat_callback))
.route("/wechat/miniprogram/callback", web::post().to(handle_wechat_miniprogram_callback))
// WebSocket连接（内网服务器连接）
.route("/ws/control", web::get().to(websocket_handler))
.route("/ws/task", web::get().to(websocket_handler))
// 测试接口（用于开发调试）
.route("/test/websocket/send", web::post().to(test_websocket_send))
.route("/test/websocket/send_and_wait", web::post().to(test_websocket_send_and_wait))
.route("/test/websocket/get_manager", web::get().to(test_websocket_get_manager))
.route("/test/websocket/connection_send", web::get().to(test_websocket_connection_send))
// 直接测试WebSocketConnection方法
.route("/test/websocket/direct_send", web::post().to(test_websocket_direct_send))
.route("/test/websocket/direct_send_and_wait", web::post().to(test_websocket_direct_send_and_wait))
)
})
.bind(&bind_address)?
.run();
println!("✅ 网关服务已启动在 {} (通过nginx代理)", bind_address);
println!("🔍 可用接口:");
println!(" GET /api/v1/health - 健康检查");
println!(" GET /api/v1/system - 系统信息");
println!(" POST /api/v1/task - 处理任务");
println!(" GET /api/v1/task/{{task_id}} - 查询任务状态");
println!(" GET /api/v1/tasks - 查询任务列表");
println!(" POST /api/v1/wechat/callback - 企业微信回调");
println!(" POST /api/v1/wechat/miniprogram/callback - 微信小程序回调");
println!(" GET /api/v1/ws/control - WebSocket控制通道");
println!(" GET /api/v1/ws/task - WebSocket任务通道");
println!(" POST /api/v1/test/websocket/send - WebSocket发送测试");
println!(" POST /api/v1/test/websocket/send_and_wait - WebSocket发送并等待测试");
println!(" GET /api/v1/test/websocket/get_manager - WebSocket管理器测试");
println!(" GET /api/v1/test/websocket/connection_send - WebSocket连接发送测试");
println!(" POST /api/v1/test/websocket/direct_send - WebSocket直接发送测试");
println!(" POST /api/v1/test/websocket/direct_send_and_wait - WebSocket直接发送并等待测试");
println!(" 🌐 外部访问: https://pactgo.cn (nginx代理)");
println!(" 🔗 WebSocket连接: wss://pactgo.cn/api/v1/ws/control");
server.await
}

15
Claw/Server/shared/Cargo.toml Normal file
View File

@@ -0,0 +1,15 @@
[package]
name = "shared"
version = "0.1.0"
edition = "2024"
[dependencies]
serde = { version = "^1.0", features = ["derive"] }
serde_json = "^1.0"
chrono = { version = "^0.4", features = ["serde"] }
uuid = { version = "^1.0", features = ["v4", "serde"] }
tokio = { version = "^1.0", features = ["full"] }
bytes = "1.0"
# heed = "^0.20" # 暂时移除，后续实现HeedDB功能
# redis = "^0.29" # 暂时移除，后续实现Redis功能

51
Claw/Server/shared/examples/embedded_redis_usage.rs Normal file
View File

@@ -0,0 +1,51 @@
use shared::embedded_redis::{EmbeddedRedisManager, EmbeddedRedisConfig};
use std::sync::Arc;
/// 嵌入式Redis使用示例
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("🚀 启动嵌入式Redis服务器示例...");
// 配置嵌入式Redis
let config = EmbeddedRedisConfig {
bind: "127.0.0.1".to_string(),
port: 6379,
data_dir: "./embedded_redis_data".to_string(),
persistence: true,
max_memory: 64 * 1024 * 1024, // 64MB
cleanup_interval: 300, // 5分钟
};
// 创建管理器
let redis_manager = Arc::new(EmbeddedRedisManager::new(config));
// 启动Redis服务器
redis_manager.start().await?;
println!("✅ 嵌入式Redis服务器已启动！");
println!("📍 监听地址: {}", redis_manager.get_connection_url().await);
// 验证服务器状态
let is_running = redis_manager.is_running().await;
println!("🔍 服务器运行状态: {}", if is_running { "运行中" } else { "已停止" });
// 获取配置信息
let config = redis_manager.get_config().await;
println!("📋 配置信息:");
println!(" - 绑定地址: {}", config.bind);
println!(" - 监听端口: {}", config.port);
println!(" - 数据目录: {}", config.data_dir);
println!(" - 持久化: {}", if config.persistence { "启用" } else { "禁用" });
println!(" - 最大内存: {}MB", config.max_memory / (1024 * 1024));
// 保持服务器运行一段时间
println!("⏰ 服务器将在5秒后停止...");
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
// 停止Redis服务器
redis_manager.stop().await;
println!("✅ 嵌入式Redis服务器已停止！");
Ok(())
}

745
Claw/Server/shared/src/embedded_redis.rs Normal file
View File

@@ -0,0 +1,745 @@
use tokio::net::{TcpListener, TcpStream};
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;
use std::fs;
use std::time::{SystemTime, Duration, UNIX_EPOCH};
/// 嵌入式Redis配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EmbeddedRedisConfig {
pub bind: String,
pub port: u16,
pub data_dir: String,
pub persistence: bool,
pub max_memory: usize,
pub cleanup_interval: u64, // 秒
}
impl Default for EmbeddedRedisConfig {
fn default() -> Self {
Self {
bind: "127.0.0.1".to_string(),
port: 6379,
data_dir: "./embedded_redis_data".to_string(),
persistence: true,
max_memory: 64 * 1024 * 1024, // 64MB
cleanup_interval: 300, // 5分钟
}
}
}
/// 键值对数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
struct RedisValue {
data: Vec<u8>,
expires_at: Option<u64>, // Unix时间戳（秒）
}
impl RedisValue {
fn new(data: Vec<u8>) -> Self {
Self {
data,
expires_at: None,
}
}
fn with_expiry(data: Vec<u8>, ttl_seconds: u64) -> Self {
let expires_at = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs() + ttl_seconds;
Self {
data,
expires_at: Some(expires_at),
}
}
fn is_expired(&self) -> bool {
if let Some(expires_at) = self.expires_at {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
now >= expires_at
} else {
false
}
}
}
/// 嵌入式Redis数据库
pub struct EmbeddedRedisDb {
data: RwLock<HashMap<String, RedisValue>>,
config: EmbeddedRedisConfig,
}
impl EmbeddedRedisDb {
fn new(config: EmbeddedRedisConfig) -> Self {
let mut db = Self {
data: RwLock::new(HashMap::new()),
config,
};
// 如果启用了持久化，加载数据
if db.config.persistence {
let _ = db.load_from_disk();
}
db
}
/// 从磁盘加载数据
async fn load_from_disk(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let data_path = Path::new(&self.config.data_dir).join("redis_data.json");
if data_path.exists() {
let data = fs::read_to_string(&data_path)?;
let loaded_data: HashMap<String, RedisValue> = serde_json::from_str(&data)?;
let mut db = self.data.write().await;
*db = loaded_data;
println!("✅ 从磁盘加载 {} 个键值对", db.len());
}
Ok(())
}
/// 保存到磁盘
async fn save_to_disk(&self) -> Result<(), Box<dyn std::error::Error>> {
if !self.config.persistence {
return Ok(());
}
// 确保数据目录存在
fs::create_dir_all(&self.config.data_dir)?;
let data_path = Path::new(&self.config.data_dir).join("redis_data.json");
let temp_path = Path::new(&self.config.data_dir).join("redis_data.json.tmp");
let db = self.data.read().await;
let data = serde_json::to_string_pretty(&*db)?;
// 先写入临时文件
fs::write(&temp_path, data)?;
// 原子性重命名
fs::rename(&temp_path, &data_path)?;
println!("💾 保存 {} 个键值对到磁盘", db.len());
Ok(())
}
/// 清理过期键
async fn cleanup_expired(&self) {
let mut db = self.data.write().await;
let keys_to_remove: Vec<String> = db
.iter()
.filter_map(|(k, v)| if v.is_expired() { Some(k.clone()) } else { None })
.collect();
for key in keys_to_remove {
db.remove(&key);
println!("🧹 清理过期键: {}", key);
}
}
/// 获取值
async fn get(&self, key: &str) -> Option<Vec<u8>> {
let db = self.data.read().await;
if let Some(value) = db.get(key) {
if !value.is_expired() {
Some(value.data.clone())
} else {
None
}
} else {
None
}
}
/// 设置值
async fn set(&self, key: String, value: Vec<u8>) {
let mut db = self.data.write().await;
db.insert(key, RedisValue::new(value));
// 异步保存到磁盘（简化实现）
if self.config.persistence {
// 这里可以添加异步保存逻辑
// 为了简化，暂时不实现自动保存
}
}
/// 设置带过期时间的值
async fn setex(&self, key: String, value: Vec<u8>, ttl_seconds: u64) {
let mut db = self.data.write().await;
db.insert(key, RedisValue::with_expiry(value, ttl_seconds));
}
/// 删除键
async fn del(&self, keys: &[String]) -> i64 {
let mut db = self.data.write().await;
let mut count = 0;
for key in keys {
if db.remove(key).is_some() {
count += 1;
}
}
count
}
/// 检查键是否存在
async fn exists(&self, key: &str) -> bool {
let db = self.data.read().await;
if let Some(value) = db.get(key) {
!value.is_expired()
} else {
false
}
}
/// 获取所有键
async fn keys(&self, pattern: &str) -> Vec<String> {
let db = self.data.read().await;
db.keys()
.filter(|key| {
// 简单的通配符匹配
if pattern == "*" {
true
} else if pattern.ends_with("*") {
let prefix = &pattern[..pattern.len() - 1];
key.starts_with(prefix)
} else {
*key == pattern
}
})
.cloned()
.collect()
}
/// 获取TTL
async fn ttl(&self, key: &str) -> i64 {
let db = self.data.read().await;
if let Some(value) = db.get(key) {
if let Some(expires_at) = value.expires_at {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
if now < expires_at {
(expires_at - now) as i64
} else {
-2 // 键已过期
}
} else {
-1 // 键永不过期
}
} else {
-2 // 键不存在
}
}
/// 获取数据库大小
async fn dbsize(&self) -> usize {
let db = self.data.read().await;
db.len()
}
/// 清空数据库
async fn flushdb(&self) {
let mut db = self.data.write().await;
db.clear();
println!("🗑️ 清空数据库");
}
}
/// 嵌入式Redis服务器
pub struct EmbeddedRedisServer {
config: EmbeddedRedisConfig,
listener: Option<TcpListener>,
db: Arc<EmbeddedRedisDb>,
running: Arc<Mutex<bool>>,
}
impl EmbeddedRedisServer {
/// 创建新的嵌入式Redis服务器
pub fn new(config: EmbeddedRedisConfig) -> Self {
let db = Arc::new(EmbeddedRedisDb::new(config.clone()));
Self {
config,
listener: None,
db,
running: Arc::new(Mutex::new(false)),
}
}
/// 启动嵌入式Redis服务器
pub async fn start(&mut self) -> Result<(), Box<dyn std::error::Error>> {
use std::net::SocketAddr;
let addr: SocketAddr = format!("{}:{}", self.config.bind, self.config.port).parse()?;
println!("🚀 启动嵌入式Redis服务器...");
println!("📍 绑定地址: {}", addr);
println!("💾 数据目录: {}", self.config.data_dir);
println!("💾 持久化: {}", if self.config.persistence { "启用" } else { "禁用" });
println!("💾 最大内存: {}MB", self.config.max_memory / (1024 * 1024));
// 创建TCP监听器
let listener = TcpListener::bind(addr).await?;
self.listener = Some(listener);
// 设置运行状态
*self.running.lock().await = true;
println!("✅ 嵌入式Redis服务器启动成功！监听地址: {}", addr);
// 启动后台任务
self.start_background_tasks().await;
// 启动接受连接的任务
self.accept_connections();
Ok(())
}
/// 启动后台任务
async fn start_background_tasks(&self) {
let db = self.db.clone();
let running = self.running.clone();
let cleanup_interval = self.config.cleanup_interval;
let persistence = self.config.persistence;
// 清理过期键任务
tokio::spawn(async move {
while *running.lock().await {
db.cleanup_expired().await;
tokio::time::sleep(Duration::from_secs(cleanup_interval)).await;
}
});
// 如果启用了持久化，启动定期保存任务
if persistence {
let db = self.db.clone();
let running = self.running.clone();
tokio::spawn(async move {
while *running.lock().await {
tokio::time::sleep(Duration::from_secs(60)).await;
if let Err(e) = db.save_to_disk().await {
eprintln!("❌ 自动保存失败: {}", e);
}
}
});
}
}
/// 接受连接
fn accept_connections(&self) {
if let Some(listener) = &self.listener {
let db = self.db.clone();
let running = self.running.clone();
// 创建一个新的监听器用于异步任务
let addr = listener.local_addr().expect("Failed to get local address");
tokio::spawn(async move {
// 重新绑定监听器
match TcpListener::bind(addr).await {
Ok(new_listener) => {
loop {
if !*running.lock().await {
break;
}
match new_listener.accept().await {
Ok((socket, addr)) => {
println!("🔗 新连接来自: {}", addr);
let db = db.clone();
let running = running.clone();
tokio::spawn(async move {
if let Err(e) = handle_connection(socket, db, running).await {
eprintln!("❌ 连接处理错误: {}", e);
}
});
}
Err(e) => {
eprintln!("❌ 接受连接错误: {}", e);
break;
}
}
}
}
Err(e) => {
eprintln!("❌ 重新绑定监听器失败: {}", e);
}
}
});
}
}
/// 获取Redis连接URL
pub fn get_connection_url(&self) -> String {
format!("redis://{}:{}/", self.config.bind, self.config.port)
}
/// 获取配置
pub fn get_config(&self) -> &EmbeddedRedisConfig {
&self.config
}
/// 停止服务器
pub async fn stop(&mut self) {
println!("🛑 停止嵌入式Redis服务器...");
// 设置停止标志
*self.running.lock().await = false;
// 保存数据到磁盘
if self.config.persistence {
if let Err(e) = self.db.save_to_disk().await {
eprintln!("❌ 保存数据失败: {}", e);
}
}
self.listener = None;
println!("✅ 嵌入式Redis服务器已停止");
}
/// 获取数据库引用（供内部使用）
pub fn get_db(&self) -> Arc<EmbeddedRedisDb> {
self.db.clone()
}
}
/// 处理单个连接
async fn handle_connection(
mut socket: TcpStream,
db: Arc<EmbeddedRedisDb>,
running: Arc<Mutex<bool>>,
) -> Result<(), Box<dyn std::error::Error>> {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let (reader, mut writer) = socket.split();
let mut reader = BufReader::new(reader);
let mut buffer = String::new();
loop {
if !*running.lock().await {
break;
}
buffer.clear();
match reader.read_line(&mut buffer).await {
Ok(0) => break, // 连接关闭
Ok(_) => {
let line = buffer.trim();
if line.is_empty() {
continue;
}
let response = match parse_and_execute_command(line, &db).await {
Ok(resp) => resp,
Err(e) => format!("-ERR {}\r\n", e),
};
writer.write_all(response.as_bytes()).await?;
writer.flush().await?;
}
Err(e) => {
eprintln!("❌ 读取错误: {}", e);
break;
}
}
}
Ok(())
}
/// 解析并执行命令
async fn parse_and_execute_command(
line: &str,
db: &Arc<EmbeddedRedisDb>,
) -> Result<String, Box<dyn std::error::Error>> {
// 解析Redis协议格式
let parts = parse_redis_protocol(line)?;
if parts.is_empty() {
return Ok("+OK\r\n".to_string());
}
let command = parts[0].to_uppercase();
match command.as_str() {
"PING" => {
if parts.len() > 1 {
Ok(format!("+{}\r\n", parts[1..].join(" ")))
} else {
Ok("+PONG\r\n".to_string())
}
}
"GET" => {
if parts.len() != 2 {
return Err("wrong number of arguments for 'get' command".into());
}
let key = &parts[1];
match db.get(key).await {
Some(value) => {
let value_str = String::from_utf8_lossy(&value);
Ok(format!("${}\r\n{}\r\n", value_str.len(), value_str))
}
None => Ok("$-1\r\n".to_string()),
}
}
"SET" => {
if parts.len() < 3 {
return Err("wrong number of arguments for 'set' command".into());
}
let key = &parts[1];
let value = &parts[2];
db.set(key.clone(), value.as_bytes().to_vec()).await;
Ok("+OK\r\n".to_string())
}
"SETEX" => {
if parts.len() != 4 {
return Err("wrong number of arguments for 'setex' command".into());
}
let key = &parts[1];
let ttl: u64 = parts[2].parse().map_err(|_| "invalid expire time")?;
let value = &parts[3];
db.setex(key.clone(), value.as_bytes().to_vec(), ttl).await;
Ok("+OK\r\n".to_string())
}
"DEL" => {
if parts.len() < 2 {
return Err("wrong number of arguments for 'del' command".into());
}
let keys: Vec<String> = parts[1..].to_vec();
let count = db.del(&keys).await;
Ok(format!(":{}\r\n", count))
}
"EXISTS" => {
if parts.len() != 2 {
return Err("wrong number of arguments for 'exists' command".into());
}
let key = &parts[1];
if db.exists(key).await {
Ok(":1\r\n".to_string())
} else {
Ok(":0\r\n".to_string())
}
}
"KEYS" => {
let pattern = if parts.len() > 1 { &parts[1] } else { "*" };
let keys = db.keys(pattern).await;
if keys.is_empty() {
Ok("*0\r\n".to_string())
} else {
let mut response = format!("*{}\r\n", keys.len());
for key in keys {
response.push_str(&format!("${}\r\n{}\r\n", key.len(), key));
}
Ok(response)
}
}
"TTL" => {
if parts.len() != 2 {
return Err("wrong number of arguments for 'ttl' command".into());
}
let key = &parts[1];
let ttl = db.ttl(key).await;
Ok(format!(":{}\r\n", ttl))
}
"DBSIZE" => {
let size = db.dbsize().await;
Ok(format!(":{}\r\n", size))
}
"FLUSHDB" => {
db.flushdb().await;
Ok("+OK\r\n".to_string())
}
"INFO" => {
let info = format!(
"# Server\r\n\
redis_version:embedded-2.0\r\n\
redis_mode:standalone\r\n\
tcp_port:{}\r\n\
\r\n\
# Clients\r\n\
connected_clients:1\r\n\
\r\n\
# Memory\r\n\
used_memory:{}\r\n\
maxmemory:{}\r\n\
\r\n\
# Persistence\r\n\
rdb_last_save_time:{}\r\n\
\r\n\
# Stats\r\n\
total_connections_received:1\r\n\
total_commands_processed:1\r\n\
\r\n",
6379,
1024,
64 * 1024 * 1024,
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
);
Ok(format!("${}\r\n{}\r\n", info.len(), info))
}
"QUIT" => {
Ok("+OK\r\n".to_string())
}
_ => {
Err(format!("unknown command '{}'", command).into())
}
}
}
/// 解析Redis协议（简化版）
fn parse_redis_protocol(line: &str) -> Result<Vec<String>, Box<dyn std::error::Error>> {
// 这里实现简化的Redis协议解析
// 实际应该解析RESP协议，这里为了简化，直接按空格分割
Ok(line.split_whitespace().map(|s| s.to_string()).collect())
}
/// 嵌入式Redis管理器
pub struct EmbeddedRedisManager {
server: Arc<Mutex<EmbeddedRedisServer>>,
}
impl EmbeddedRedisManager {
/// 创建新的管理器
pub fn new(config: EmbeddedRedisConfig) -> Self {
Self {
server: Arc::new(Mutex::new(EmbeddedRedisServer::new(config))),
}
}
/// 启动Redis服务器
pub async fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
let mut server = self.server.lock().await;
server.start().await
}
/// 停止Redis服务器
pub async fn stop(&self) {
let mut server = self.server.lock().await;
server.stop().await;
}
/// 获取Redis连接URL
pub async fn get_connection_url(&self) -> String {
let server = self.server.lock().await;
server.get_connection_url()
}
/// 获取配置
pub async fn get_config(&self) -> EmbeddedRedisConfig {
let server = self.server.lock().await;
server.get_config().clone()
}
/// 检查是否运行中
pub async fn is_running(&self) -> bool {
let server = self.server.lock().await;
*server.running.lock().await
}
/// 获取数据库引用（供高级操作使用）
pub async fn get_db(&self) -> Arc<EmbeddedRedisDb> {
let server = self.server.lock().await;
server.get_db()
}
/// 手动保存到磁盘
pub async fn save(&self) -> Result<(), Box<dyn std::error::Error>> {
let server = self.server.lock().await;
server.get_db().save_to_disk().await
}
}
impl Drop for EmbeddedRedisManager {
fn drop(&mut self) {
// 在析构时尝试停止服务器并保存数据
let server = self.server.clone();
tokio::spawn(async move {
let mut server = server.lock().await;
server.stop().await;
});
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_embedded_redis_config() {
let config = EmbeddedRedisConfig::default();
assert_eq!(config.port, 6379);
assert_eq!(config.max_memory, 64 * 1024 * 1024);
assert!(config.persistence);
}
#[tokio::test]
async fn test_embedded_redis_manager() {
let config = EmbeddedRedisConfig {
port: 16379, // 使用非标准端口避免冲突
data_dir: "./test_redis_data".to_string(),
..Default::default()
};
let manager = EmbeddedRedisManager::new(config);
assert_eq!(manager.get_connection_url().await, "redis://127.0.0.1:16379/");
}
#[tokio::test]
async fn test_redis_commands() {
let config = EmbeddedRedisConfig {
port: 16380,
data_dir: "./test_redis_commands".to_string(),
..Default::default()
};
let db = Arc::new(EmbeddedRedisDb::new(config));
// 测试SET和GET
db.set("test_key".to_string(), b"test_value".to_vec()).await;
let value = db.get("test_key").await;
assert_eq!(value, Some(b"test_value".to_vec()));
// 测试EXISTS
assert!(db.exists("test_key").await);
assert!(!db.exists("non_existent").await);
// 测试DEL
let deleted = db.del(&["test_key".to_string()]).await;
assert_eq!(deleted, 1);
assert!(!db.exists("test_key").await);
// 测试KEYS
db.set("key1".to_string(), b"value1".to_vec()).await;
db.set("key2".to_string(), b"value2".to_vec()).await;
let keys = db.keys("*").await;
assert!(keys.contains(&"key1".to_string()));
assert!(keys.contains(&"key2".to_string()));
// 清理测试数据
db.flushdb().await;
}
}

293
Claw/Server/shared/src/lib.rs Normal file
View File

@@ -0,0 +1,293 @@
use serde::{Deserialize, Serialize};
use std::fmt;
pub mod embedded_redis;
pub use embedded_redis::{EmbeddedRedisServer, EmbeddedRedisConfig, EmbeddedRedisManager};
/// 任务类型枚举
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum TaskType {
/// 文本处理任务
TextProcessing,
/// 数据分析任务
DataAnalysis,
/// AI对话任务
AIChat,
/// 文件处理任务
FileProcessing,
/// 自定义任务
Custom(String),
}
impl fmt::Display for TaskType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
TaskType::TextProcessing => write!(f, "text_processing"),
TaskType::DataAnalysis => write!(f, "data_analysis"),
TaskType::AIChat => write!(f, "ai_chat"),
TaskType::FileProcessing => write!(f, "file_processing"),
TaskType::Custom(s) => write!(f, "custom_{}", s),
}
}
}
/// 任务状态枚举
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum TaskStatus {
/// 待处理
Pending,
/// 处理中
Processing,
/// 已完成
Completed,
/// 失败
Failed,
/// 已取消
Cancelled,
}
impl fmt::Display for TaskStatus {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
TaskStatus::Pending => write!(f, "pending"),
TaskStatus::Processing => write!(f, "processing"),
TaskStatus::Completed => write!(f, "completed"),
TaskStatus::Failed => write!(f, "failed"),
TaskStatus::Cancelled => write!(f, "cancelled"),
}
}
}
/// 任务请求数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskRequest {
/// 用户ID
pub user_id: String,
/// 任务类型
pub task_type: TaskType,
/// 任务内容
pub content: String,
/// 时间戳
pub timestamp: i64,
/// 优先级 (1-10, 数字越大优先级越高)
pub priority: u8,
/// 超时时间（秒）
pub timeout: Option<u64>,
/// 额外参数
pub extra_params: Option<serde_json::Value>,
}
/// 任务响应数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskResponse {
/// 是否成功
pub success: bool,
/// 消息
pub message: String,
/// 任务ID
pub task_id: Option<String>,
/// 结果数据
pub result: Option<serde_json::Value>,
/// 处理时间（毫秒）
pub processing_time: Option<u64>,
/// 错误信息
pub error: Option<String>,
}
/// 任务状态数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskStatusResponse {
/// 任务ID
pub task_id: String,
/// 任务状态
pub status: TaskStatus,
/// 进度 (0-100)
pub progress: u8,
/// 状态消息
pub status_message: String,
/// 创建时间
pub created_at: i64,
/// 更新时间
pub updated_at: i64,
/// 结果数据
pub result: Option<serde_json::Value>,
}
/// 服务健康检查响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthResponse {
/// 服务状态
pub status: String,
/// 服务名称
pub service: String,
/// 时间戳
pub timestamp: i64,
/// 版本号
pub version: String,
/// 额外信息
pub extra: Option<serde_json::Value>,
}
/// 错误响应数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorResponse {
/// 错误代码
pub code: String,
/// 错误消息
pub message: String,
/// 详细错误信息
pub details: Option<String>,
/// 时间戳
pub timestamp: i64,
}
/// WebSocket 消息类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WebSocketMessage {
/// 任务请求
TaskRequest(TaskRequest),
/// 任务响应
TaskResponse(TaskResponse),
/// 任务状态更新
TaskStatusUpdate(TaskStatusResponse),
/// 心跳消息
Heartbeat,
/// 错误消息
Error(ErrorResponse),
}
/// 企业微信消息数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WeChatMessage {
/// 消息ID
pub msg_id: String,
/// 发送者
pub from_user: String,
/// 接收者
pub to_user: String,
/// 消息类型
pub msg_type: String,
/// 消息内容
pub content: String,
/// 时间戳
pub timestamp: i64,
}
/// 微信小程序消息数据结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MiniProgramMessage {
/// 用户ID
pub user_id: String,
/// 消息类型
pub msg_type: String,
/// 消息内容
pub content: String,
/// 时间戳
pub timestamp: i64,
/// 额外参数
pub extra: Option<serde_json::Value>,
}
/// 常量定义
pub mod constants {
/// 默认任务超时时间（秒）
pub const DEFAULT_TASK_TIMEOUT: u64 = 300;
/// 最大任务优先级
pub const MAX_TASK_PRIORITY: u8 = 10;
/// 最小任务优先级
pub const MIN_TASK_PRIORITY: u8 = 1;
/// 默认服务端口
pub const DEFAULT_GATEWAY_PORT: u16 = 3000;
pub const DEFAULT_SMARTCLAW_PORT: u16 = 3001;
/// WebSocket 心跳间隔（秒）
pub const WEBSOCKET_HEARTBEAT_INTERVAL: u64 = 30;
/// WebSocket 超时时间（秒）
pub const WEBSOCKET_TIMEOUT: u64 = 300;
}
/// 工具函数
pub mod utils {
use super::*;
use chrono::Utc;
/// 生成任务ID
pub fn generate_task_id(user_id: &str) -> String {
format!("task_{}_{}", user_id, Utc::now().timestamp_millis())
}
/// 生成消息ID
pub fn generate_msg_id() -> String {
format!("msg_{}", Utc::now().timestamp_millis())
}
/// 验证任务优先级
pub fn validate_priority(priority: u8) -> u8 {
priority.clamp(constants::MIN_TASK_PRIORITY, constants::MAX_TASK_PRIORITY)
}
/// 获取当前时间戳
pub fn current_timestamp() -> i64 {
Utc::now().timestamp()
}
/// 创建成功响应
pub fn create_success_response(message: &str, task_id: Option<String>, result: Option<serde_json::Value>) -> TaskResponse {
TaskResponse {
success: true,
message: message.to_string(),
task_id,
result,
processing_time: None,
error: None,
}
}
/// 创建错误响应
pub fn create_error_response(message: &str, error: Option<String>) -> TaskResponse {
TaskResponse {
success: false,
message: message.to_string(),
task_id: None,
result: None,
processing_time: None,
error,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_task_type_display() {
assert_eq!(TaskType::TextProcessing.to_string(), "text_processing");
assert_eq!(TaskType::DataAnalysis.to_string(), "data_analysis");
assert_eq!(TaskType::Custom("test".to_string()).to_string(), "custom_test");
}
#[test]
fn test_task_status_display() {
assert_eq!(TaskStatus::Pending.to_string(), "pending");
assert_eq!(TaskStatus::Completed.to_string(), "completed");
assert_eq!(TaskStatus::Failed.to_string(), "failed");
}
#[test]
fn test_generate_task_id() {
let task_id = utils::generate_task_id("user123");
assert!(task_id.starts_with("task_user123_"));
}
#[test]
fn test_validate_priority() {
assert_eq!(utils::validate_priority(0), 1);
assert_eq!(utils::validate_priority(5), 5);
assert_eq!(utils::validate_priority(15), 10);
}
}