Rust WebSocket通信库holochain_websocket使用指南

概述

holochain_websocket是一个用于构建高效分布式应用的Rust WebSocket通信库，特别适合需要实现网络通信和数据同步的场景。它为Holochain框架提供了WebSocket通信能力，但也可以独立使用。

主要特性

• 轻量级WebSocket通信实现
• 支持分布式应用网络
• 高效的数据同步机制
• 与Holochain框架无缝集成
• 提供消息路由和广播功能

安装方法

在Cargo.toml中添加依赖：

[dependencies]
holochain_websocket = "0.0.10"
tokio = { version = "1", features = ["full"] }  # 添加tokio依赖

完整示例：带心跳检测的WebSocket服务器和客户端

WebSocket服务器示例

use holochain_websocket::{WebsocketServer, WebsocketSender};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // 绑定到本地地址
    let addr = "127.0.0.1:9000".parse().unwrap();
    let server = WebsocketServer::bind(addr).await.unwrap();
    println!("WebSocket服务器启动在: {}", addr);

    // 使用Arc和Mutex来安全地共享客户端连接
    let clients: Arc<Mutex<HashMap<usize, WebsocketSender>>> = Arc::new(Mutex::new(HashMap::new()));

    // 处理新连接
    let clients_for_accept = clients.clone();
    tokio::spawn(async move {
        let mut id_counter = 0;
        while let Some((id, sender, mut receiver)) = server.accept().await {
            let clients_for_handler = clients_for_accept.clone();
            let sender_clone = sender.clone();

            // 将新客户端添加到连接池
            clients_for_accept.lock().await.insert(id, sender_clone);

            // 为每个客户端生成处理任务
            tokio::spawn(async move {
                println!("客户端 {} 已连接", id);

                // 心跳检测任务
                let heartbeat_sender = sender.clone();
                tokio::spawn(async move {
                    loop {
                        sleep(Duration::from_secs(30)).await;
                        if let Err(e) = heartbeat_sender.send("ping".to_string()).await {
                            println!("心跳发送失败: {}", e);
                            break;
                        }
                    }
                });

                // 消息处理循环
                while let Some(msg) = receiver.recv().await {
                    match msg.as_str() {
                        "pong" => {
                            println!("客户端 {} 心跳响应", id);
                        }
                        _ => {
                            println!("客户端 {} 发送: {}", id, msg);

                            // 广播给其他客户端
                            let mut clients = clients_for_handler.lock().await;
                            for (client_id, client_sender) in clients.iter_mut() {
                                if *client_id != id {
                                    let _ = client_sender.send(format!("客户端{}说: {}", id, msg)).await;
                                }
                            }
                        }
                    }
                }

                // 客户端断开连接时清理资源
                println!("客户端 {} 断开连接", id);
                clients_for_handler.lock().await.remove(&id);
            });

            id_counter += 1;
        }
    });

    // 等待Ctrl+C信号
    tokio::signal::ctrl_c().await.unwrap();
    println!("服务器正在关闭...");
}

WebSocket客户端示例

use holochain_websocket::WebsocketSender;
use tokio::net::TcpStream;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // 连接服务器
    let stream = match TcpStream::connect("127.0.0.1:9000").await {
        Ok(s) => s,
        Err(e) => {
            eprintln!("连接服务器失败: {}", e);
            return;
        }
    };

    // 建立WebSocket连接
    let (mut sender, mut receiver) = match WebsocketSender::connect(stream).await {
        Ok((s, r)) => (s, r),
        Err(e) => {
            eprintln!("WebSocket握手失败: {}", e);
            return;
        }
    };

    println!("已连接到服务器");

    // 心跳检测任务
    let mut heartbeat_sender = sender.clone();
    tokio::spawn(async move {
        loop {
            sleep(Duration::from_secs(30)).await;
            if let Err(e) = heartbeat_sender.send("pong".to_string()).await {
                eprintln!("心跳响应失败: {}", e);
                break;
            }
        }
    });

    // 消息发送任务
    let mut message_sender = sender.clone();
    tokio::spawn(async move {
        for i in 1..=5 {
            sleep(Duration::from_secs(1)).await;
            if let Err(e) = message_sender.send(format!("消息 {}", i)).await {
                eprintln!("发送消息失败: {}", e);
                break;
            }
        }
    });

    // 消息接收循环
    while let Some(msg) = receiver.recv().await {
        match msg.as_str() {
            "ping" => {
                if let Err(e) = sender.send("pong".to_string()).await {
                    eprintln!("响应心跳失败: {}", e);
                    break;
                }
            }
            _ => {
                println!("收到消息: {}", msg);
            }
        }
    }

    println!("连接已关闭");
}

完整示例说明

1. 服务器功能:

   • 监听指定端口接受WebSocket连接
   • 维护所有连接的客户端列表
   • 实现心跳检测机制(每30秒发送ping)
   • 支持消息广播功能
   • 自动清理断开连接的客户端资源

2. 客户端功能:

   • 连接到指定的WebSocket服务器
   • 实现心跳响应机制
   • 支持发送和接收消息
   • 自动处理服务器ping消息

3. 扩展功能:

   • 使用Arc和Mutex实现线程安全的连接管理
   • 通过tokio::spawn实现并发处理
   • 完善的错误处理和资源清理

运行步骤

1. 首先启动服务器程序
2. 然后启动一个或多个客户端程序
3. 客户端会自动发送5条测试消息并接收服务器广播
4. 按下Ctrl+C停止服务器

这个完整示例展示了holochain_websocket库的核心功能，包括连接管理、消息传递、心跳检测和广播功能，可以作为开发分布式WebSocket应用的基础模板。