Rust SocketCluster 协议 V1 服务器库

作为库实现的可扩展 SocketCluster WebSocket 服务器实现，使用 Rust、Axum 和 Tokio 构建。

功能

• 🚀 基于 WebSocket 的实时通信
• 📡 高效消息分发模式的发布/订阅
• 🔌 可定制的中间件用于处理数据包
• 🔐 JWT 认证支持
• 💓 Ping/Pong 机制用于连接健康监控
• 🌐 与 WebSocket 功能集成的 HTTP 端点
• ⚙️ 使用 TOML 文件进行灵活配置

目录

• 安装
• 快速入门
• 配置
• 使用示例
  • 设置服务器
  • 实现自定义中间件
  • 直接发布消息的 HTTP 端点
• API 文档
• 贡献
• 许可证

安装

将以下内容添加到您的 Cargo.toml

[dependencies]
socketcluster_server = "0.1.0"

快速入门

以下是一个使服务器启动和运行的简单示例

use socketcluster_server::{create_socketcluster_state, ServerConfig, WebSocketSender, ws_handler};
use axum::{Router, routing::get};
use tokio::net::TcpListener;
use std::net::SocketAddr;

#[tokio::main]
async fn main() {
    // Load configuration
    let config = ServerConfig {
        ping_interval: std::time::Duration::from_secs(30),
        ping_timeout: std::time::Duration::from_secs(5),
        port: 8080,
        host: "127.0.0.1".to_string(),
        jwt_secret: "your-secret-key".to_string(),
    };

    // Create application state
    let state = create_socketcluster_state::<WebSocketSender>(config.clone());

    // Set up router
    let app = Router::new()
        .route("/ws", get(ws_handler))
        .with_state(state);

    // Start the server
    let addr = format!("{}:{}", config.host, config.port);
    let listener = TcpListener::bind(&addr).await.unwrap();
    println!("Server listening on: {}", addr);
    axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap();
}

配置

服务器可以通过 TOML 文件进行配置。创建一个 config/settings.toml 文件，其结构如下

ping_interval = 30  # in seconds
ping_timeout = 5    # in seconds
port = 8080
host = "127.0.0.1"
jwt_secret = "your-secret-key"

使用示例

直接发布消息的 HTTP 端点

此服务器是一个库，因此您可以直接使用 Axum。在这里，您可以找到一个简单的示例，说明您如何直接操作 socketcluster 通道以及如何使用 Axum 路由端点进行发布。

// ... use ...

// Axum HTTP handler to send messages directly to the channels
async fn send_handler<S: Sender>(
    State(state): State<AppState<S>>,
    TypedHeader(auth_header): TypedHeader<headers::Authorization<headers::authorization::Bearer>>,
    Json(payload): Json<serde_json::value::Map<String, Value>>,
) -> impl IntoResponse {

    // JWT validation logic here...

    let channel = payload.get("channel").unwrap().as_str().unwrap();
    let message = payload.get("message").unwrap();

    if let Some(subscribers) = state.subscriptions.read().await.get(channel) {
        let publish_event = Packet {
            event: Some(Event::Publish),
            data: Some(message.clone()),
            ..Default::default()
        };

        for sub_socket_id in subscribers {
            if let Some(socket_data) = state.sockets.read().await.get(sub_socket_id) {
                let mut sender = socket_data.sender.lock().await;
                let _ = sender.send(AxumMessage::Text(serde_json::to_string(&publish_event).unwrap())).await;
            }
        }
        (StatusCode::OK, json!({ "ok": true }).to_string()).into_response()
    } else {
        (StatusCode::NOT_FOUND, json!({ "ok": false, "message": "channel not found" }).to_string()).into_response()
    }
}

#[tokio::main]
async fn main() {
  let settings = config::Config::builder()
      .add_source(config::File::with_name("config/settings.toml"))
      .build()
      .expect("Configuration loading failed");

  let config: ServerConfig = settings.clone().try_deserialize().expect("Failed to deserialize configuration");

  let mut sc_state = create_socketcluster_state::<WebSocketSender>(config.clone());

  // Add routes and start the server
  let app = Router::new()
      // Main socketcluster client handler (websocket connection)
      .route("/ws/", get(ws_handler))
      // Axum POST handler to allow sending messages to channels directly
      .route("/publish_to_channel", post(send_handler))
      .with_state(sc_state);

  let addr = format!("{}:{}", config.host, config.port);
  let listener = TcpListener::bind(&addr).await.unwrap();
  axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap();
}

实现自定义中间件

struct LoggingMiddleware;

#[async_trait]
impl Middleware for LoggingMiddleware {
    async fn handle(&self, packet: &mut Packet, _auth_data: &AuthData) -> bool {
        println!("Processing packet: {:?}", packet);
        true // Allow the packet to proceed
    }
}

// In your main function:
let logging_middleware = Arc::new(LoggingMiddleware);
sc_state.add_middleware(logging_middleware);

API 文档

要获取详细的 API 文档，请运行

cargo doc --open

或访问 https://docs.rs/socketcluster_server/latest/socketcluster_server/

贡献

欢迎贡献！请随时提交拉取请求。

许可证

本项目采用 MIT 许可证 - 有关详细信息，请参阅 LICENSE 文件。