Rust使用websocket创建聊天室

type Tx = mpsc::UnboundedSender<Message>;type Rx = mpsc::UnboundedReceiver<Message>;
struct Shared {    peers: HashMap<SocketAddr, Tx>,}
struct Peer {    stream: WebSocketStream<TcpStream>,    rx: Rx,}
impl Shared {    /// Create a new, empty, instance of `Shared`.    fn new() -> Self {        Shared {            peers: HashMap::new(),        }    }    // 广播    async fn broadcast(&mut self, sender: SocketAddr, msg: &str) {        for peer in self.peers.iter_mut() {            if *peer.0 != sender {                let _ = peer.1.send(Message::Text(msg.to_string()));            }        }    }}
impl Peer {    /// Create a new instance of `Peer`.    async fn new(state: Arc<Mutex<Shared>>, stream: WebSocketStream<TcpStream>) -> Result<Peer> {        // Get the client socket address        let addr = stream.get_ref().peer_addr()?;
        // Create a channel for this peer        let (tx, rx) = mpsc::unbounded_channel();
        // Add an entry for this `Peer` in the shared state map.        state.lock().await.peers.insert(addr, tx);        Ok(Peer { stream, rx })    }}

async fn handle_connection(    state: Arc<Mutex<Shared>>,    raw_stream: TcpStream,    addr: SocketAddr,) -> Result<()> {    println!("Incoming TCP connection from: {}", addr);
    let mut ws_stream = accept_async(raw_stream)        .await        .expect("Error during the websocket handshake occurred");    println!("WebSocket connection established: {}", addr);    ws_stream        .send(Message::Text("请输入您的昵称：".to_string()))        .await?;    let mut username = String::new();
    loop {        let line = ws_stream.next().await;        match line {            Some(Ok(line)) => {                if line.is_binary() || line.is_text() {                    let msg = line.to_text().unwrap();                    if msg.len() > 0 {                        username.push_str(line.to_text().unwrap());                        break;                    }                }                ws_stream                    .send(Message::Text("请输入您的昵称：".to_string()))                    .await?;            }            // We didn't get a line so we return early here.            _ => {                println!("Failed to get username from {}. Client disconnected.", addr);                return Ok(());            }        }    }    // 获取到昵称，发送消息给ta    let hello_msg = format!("您设置的昵称是{}，欢迎进入聊天室", username);    ws_stream.send(Message::Text(hello_msg)).await?;
    let mut peer = Peer::new(state.clone(), ws_stream).await?;
    {        let mut state = state.lock().await;        let n = state.peers.len();        let msg = format!("欢迎[{}]进入聊天室，当前人数：{}", username, n);        println!("{}", msg);        state.broadcast(addr, &msg).await;    }
    let mut interval = tokio::time::interval(Duration::from_secs(30));    loop {        tokio::select! {            // 通道中有值时            Some(msg) = peer.rx.recv() => {                peer.stream.send(msg).await?;            },            // websocket接收到消息时            msg = timeout(Duration::from_secs(33),peer.stream.next()) => {                match msg {                    Ok(Some(Ok(msg)))=>{                        if msg.is_text() || msg.is_binary(){                            let content = msg.to_text().unwrap();                            let content = content.trim();                            if content.len() > 0 {                                let mut state = state.lock().await;                                let msg = format!("{}: {}", username, content);                                state.broadcast(addr, &msg).await;                            }
                        }else if msg.is_close() {                            println!("客户端断开连接");                            break;                        }                    },                    Ok(Some(Err(e)))=>{                        println!("客户端发生错误,{}",e);                    },
                    Ok(None)=>{                        println!("客户端已经掉线");                        break;                    },
                    Err(e)=>{                        println!("获取消息超时,{}",e);                        break;                    }                }            }
            // 时钟到达时            _ = interval.tick()=> {                let payload = Vec::new();                let ping = Message::Ping(payload);                peer.stream.send(ping).await?;            }
        }    }    {        let mut state = state.lock().await;        state.peers.remove(&addr);        let n = state.peers.len();        let msg = format!("[{}]离开了聊天室，当前人数：{}", username, n);        println!("{}", msg);        state.broadcast(addr, &msg).await;    }    Ok(())}

use std::{    collections::HashMap, env, io::Error as IoError, net::SocketAddr, sync::Arc, time::Duration,};use tokio::sync::{mpsc, Mutex};
use futures_util::{SinkExt, StreamExt};use tokio::net::{TcpListener, TcpStream};use tokio::time::timeout;use tokio_tungstenite::WebSocketStream;use tokio_tungstenite::{    accept_async,    tungstenite::{Message, Result},};
#[tokio::main]async fn main() -> Result<(), IoError> {    let addr = env::args()        .nth(1)        .unwrap_or_else(|| "0.0.0.0:9528".to_string());
    let state = Arc::new(Mutex::new(Shared::new()));
    // Create the event loop and TCP listener we'll accept connections on.    let listener = TcpListener::bind(&addr).await.expect("Failed to bind");    println!("Listening on: {}", addr);
    // Let's spawn the handling of each connection in a separate task.    while let Ok((stream, addr)) = listener.accept().await {        let state = Arc::clone(&state);        tokio::spawn(handle_connection(state, stream, addr));    }
    Ok(())}