Rust WebSocket库websocket-base的使用，高效实现WebSocket通信与实时数据传输

安装

在项目目录中运行以下Cargo命令：

cargo add websocket-base

或者在Cargo.toml中添加以下行：

websocket-base = "0.26.5"

基本使用示例

以下是一个使用websocket-base库实现WebSocket通信的完整示例：

use websocket_base::{ClientBuilder, Message, OwnedMessage};
use std::thread;

// WebSocket客户端示例
fn main() {
    // 创建WebSocket客户端连接
    let client = ClientBuilder::new("ws://echo.websocket.org")
        .unwrap()
        .connect_secure(None)
        .unwrap();

    let (mut receiver, mut sender) = client.split().unwrap();

    // 启动接收消息线程
    let receive_thread = thread::spawn(move || {
        for message in receiver.incoming_messages() {
            match message {
                Ok(OwnedMessage::Text(text)) => {
                    println!("Received text message: {}", text);
                }
                Ok(OwnedMessage::Binary(bin)) => {
                    println!("Received binary message: {:?}", bin);
                }
                Ok(OwnedMessage::Close(_)) => {
                    println!("Connection closed");
                    break;
                }
                Ok(OwnedMessage::Ping(data)) => {
                    println!("Received ping, sending pong");
                    sender.send_message(&Message::pong(data)).unwrap();
                }
                Err(e) => {
                    println!("Error receiving message: {}", e);
                    break;
                }
                _ => {}
            }
        }
    });

    // 发送消息
    sender.send_message(&Message::text("Hello, WebSocket!")).unwrap();
    sender.send_message(&Message::binary(vec![1, 2, 3])).unwrap();
    sender.send_message(&Message::close()).unwrap();

    receive_thread.join().unwrap();
}

服务器端示例

use websocket_base::{Server, Message, OwnedMessage};
use std::thread;

// WebSocket服务器示例
fn main() {
    let server = Server::bind("127.0.0.1:8080").unwrap();

    for connection in server.filter_map(Result::ok) {
        thread::spawn(move || {
            let client = connection.accept().unwrap();
            let (mut sender, mut receiver) = client.split().unwrap();

            for message in receiver.incoming_messages() {
                match message {
                    Ok(OwnedMessage::Text(text)) => {
                        println!("Received text: {}", text);
                        sender.send_message(&Message::text(format!("Echo: {}", text))).unwrap();
                    }
                    Ok(OwnedMessage::Binary(bin)) => {
                        println!("Received binary: {:?}", bin);
                        sender.send_message(&Message::binary(bin)).unwrap();
                    }
                    Ok(OwnedMessage::Close(_)) => {
                        println!("Client disconnected");
                        break;
                    }
                    Ok(OwnedMessage::Ping(data)) => {
                        println!("Received ping, sending pong");
                        sender.send_message(&Message::pong(data)).unwrap();
                    }
                    Err(e) => {
                        println!("Error: {}", e);
                        break;
                    }
                    _ => {}
                }
            }
        });
    }
}

完整示例代码

基于上述内容，下面是一个更完整的WebSocket客户端和服务器交互示例：

客户端完整示例

use websocket_base::{ClientBuilder, Message, OwnedMessage};
use std::thread;
use std::time::Duration;

fn main() {
    // 1. 创建WebSocket客户端连接
    let client = ClientBuilder::new("ws://127.0.0.1:8080")
        .expect("创建客户端失败")
        .connect_secure(None)
        .expect("连接服务器失败");

    // 2. 分离读写通道
    let (mut receiver, mut sender) = client.split().expect("分离通道失败");

    // 3. 接收消息线程
    let receive_thread = thread::spawn(move || {
        for message in receiver.incoming_messages() {
            match message {
                Ok(OwnedMessage::Text(text)) => {
                    println!("[客户端] 收到文本消息: {}", text);
                }
                Ok(OwnedMessage::Binary(bin)) => {
                    println!("[客户端] 收到二进制消息: {:?}", bin);
                }
                Ok(OwnedMessage::Close(_)) => {
                    println!("[客户端] 连接关闭");
                    break;
                }
                Ok(OwnedMessage::Ping(data)) => {
                    println!("[客户端] 收到Ping，发送Pong");
                    sender.send_message(&Message::pong(data)).unwrap();
                }
                Err(e) => {
                    println!("[客户端] 接收消息错误: {}", e);
                    break;
                }
                _ => {}
            }
        }
    });

    // 4. 发送消息
    for i in 0..5 {
        let text = format!("消息 {}", i);
        println!("[客户端] 发送: {}", text);
        sender.send_message(&Message::text(text)).unwrap();
        thread::sleep(Duration::from_secs(1));
    }

    // 5. 关闭连接
    sender.send_message(&Message::close()).unwrap();
    receive_thread.join().unwrap();
}

服务器端完整示例

use websocket_base::{Server, Message, OwnedMessage};
use std::thread;
use std::time::Duration;

fn main() {
    // 1. 绑定服务器地址
    let server = Server::bind("127.0.0.1:8080").expect("绑定地址失败");

    println!("WebSocket服务器启动在 ws://127.0.0.1:8080");

    // 2. 处理每个连接
    for connection in server.filter_map(Result::ok) {
        thread::spawn(move || {
            // 3. 接受连接
            let client = connection.accept().expect("接受连接失败");
            let (mut sender, mut receiver) = client.split().expect("分离通道失败");

            println!("[服务器] 新客户端连接");

            // 4. 处理消息
            for message in receiver.incoming_messages() {
                match message {
                    Ok(OwnedMessage::Text(text)) => {
                        println!("[服务器] 收到文本: {}", text);
                        // 返回带前缀的响应
                        let response = format!("服务器响应: {}", text);
                        sender.send_message(&Message::text(response)).unwrap();
                    }
                    Ok(OwnedMessage::Binary(bin)) => {
                        println!("[服务器] 收到二进制: {:?}", bin);
                        // 原样返回二进制数据
                        sender.send_message(&Message::binary(bin)).unwrap();
                    }
                    Ok(OwnedMessage::Close(_)) => {
                        println!("[服务器] 客户端断开连接");
                        break;
                    }
                    Ok(OwnedMessage::Ping(data)) => {
                        println!("[服务器] 收到Ping，发送Pong");
                        sender.send_message(&Message::pong(data)).unwrap();
                    }
                    Err(e) => {
                        println!("[服务器] 错误: {}", e);
                        break;
                    }
                    _ => {}
                }
            }
        });
    }
}

特性

• 支持WebSocket协议(RFC 6455)
• 同时支持客户端和服务器端实现
• 异步和同步API
• 支持TLS加密连接
• 支持文本和二进制消息传输
• 处理Ping/Pong帧以保持连接活跃