上面说到，客户端和服务端通讯使用底层协议主要有三种选择：
TCP
Websocket
QUIC
长期来看，QUIC 是一个很有前途的技术方案，因为协议本身还在不断的演化中，对于新的 5G 网络有很好的支持，特别重要的是，对于元宇宙所必须支持的 VR/AR 解决方案，QUIC 对视频/音频的良好支持，以及实时切换接入点的同时能保持链接等特点，我们商业化的元宇宙，应该需要使用 QUIC 协议作为底层传输基础。
Rust 有非常多的 QUIC 实现，一个比较完善而且很活跃的实现是 quinn 项目。
当然，我们的第一个教学版本还是选择 Websocket
服务端，我们使用 tokio 的官方 web 框架 Axum，其支持 Websocket 的代码包括下面几个部分。
接受 http 连接请求并升级到 websocket 的函数
async fn ws_handler(ws: WebSocketUpgrade, auth: Option<TypedHeader<headers::Authorization<headers::authorization::Bearer>>>) -> impl IntoResponse { if let Some(TypedHeader(a)) = auth { println!("`{:?}` connected", a.0.token()); } ws.on_upgrade(handle_socket)}
复制代码
注意我们使用了 http 的 Authorization 头作为认证方式，http 有 Basic AgentInfo 等认证方式，我们使用 基于 token 的 Beared 认证方式，a.0.token() 这个函数从 Beared 里面取出 access token。
我们暂时不考虑 access token 从哪里获取，对一个成熟的 元宇宙服务器来说，这应该是一个专门的 login 服务，支持 微信/QQ 或者手机验证或者用户名密码的方式来获取 access token。
接受客户端消息，并定时发送 一条文本消息
async fn handle_socket(mut socket: WebSocket) { let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(10)); loop { tokio::select! { msg = socket.recv() => { println!("{:?}", msg); } _ = interval.tick()=> { socket.send(Message::Text(String::from("Hi!"))).await; } } }}
复制代码
我们暂时没有处理连接被关闭的情况。
我们先考虑 rust 客户端，由于 Websocket 本身就是为 Web 而生的，所以 Javascript 对 Websocket 的支持比 Rust 要好多了。
我们使用 websocket crate，提供客户端的支持。
一个简单的 Websocket Rust 客户端的代码如下
这段代码 从命令行取得 access token 加入到 http 头 跟 服务器建立 ws 连接
const CONNECTION: &'static str = "ws://127.0.0.1:3000/ws";
fn main() { let args: Vec<String> = std::env::args().collect(); if args.len() < 2 { return; } let mut my_headers = Headers::new(); my_headers.set(Authorization(Bearer {token: args[1].clone()})); let client = ClientBuilder::new(CONNECTION).unwrap().add_protocol("rust-websocket").custom_headers(&my_headers).connect_insecure().unwrap(); println!("Successfully connected");
复制代码
然后我们使用两个独立的线程，分别循环接受和发送消息。
发送线程
let (mut receiver, mut sender) = client.split().unwrap(); let (tx, rx) = channel(); let tx_1 = tx.clone();
let send_loop = thread::spawn(move || { loop { match rx.recv() { Ok(msg) => { match msg { OwnedMessage::Close(_) => { sender.send_message(&msg).unwrap(); }, _=> { match sender.send_message(&msg) { Err(e)=> { println!("Send Loop: {:?}", e); sender.send_message(&Message::close()).unwrap(); }, _=> {} } } } }, Err(e) => { println!("Send Loop: {:?}", e); return; } }; } });
复制代码
接收线程
let receive_loop = thread::spawn(move || { for message in receiver.incoming_messages() { let message = match message { Ok(m) => m, Err(e) => { println!("Receive Loop: {:?}", e); let _ = tx_1.send(OwnedMessage::Close(None)); return; } }; match message { OwnedMessage::Close(_) => { let _ = tx_1.send(OwnedMessage::Close(None)); return; } OwnedMessage::Ping(data) => { match tx_1.send(OwnedMessage::Pong(data)) { Ok(()) => (), Err(e) => { println!("Receive Loop: {:?}", e); return; } } } _ => println!("Receive Loop: {:?}", message), } } });
复制代码
命令行 接受输入，发送到服务器
loop { let mut input = String::new(); stdin().read_line(&mut input).unwrap(); let trimmed = input.trim(); let message = match trimmed { "/close" => { let _ = tx.send(OwnedMessage::Close(None)); break; } "/ping" => OwnedMessage::Ping(b"PING".to_vec()), _ => OwnedMessage::Text(trimmed.to_string()), };
match tx.send(message) { Ok(()) => (), Err(e) => { println!("Main Loop: {:?}", e); break; } } }
复制代码
评论