Приложение чата с широковещательной рассылкой

В этом упражнении мы хотим использовать новые знания, чтобы реализовать приложение чата с широковещательной рассылкой. Есть сервер чата, к которому подключаются клиенты и публикуют свои сообщения. Клиент читает сообщения пользователя из стандартного ввода и отправляет их на сервер. Сервер чата рассылает каждое полученное сообщение всем клиентам.

Для этого мы используем на сервере широковещательный канал, а для связи клиента с сервером — tokio_websockets.

Создайте новый проект Cargo и добавьте следующие зависимости:

Cargo.toml:

[package]
name = "chat-async"
version = "0.1.0"
edition = "2024"

[dependencies]
futures-util = { version = "0.3.32", features = ["sink"] }
http = "1.4.1"
tokio = { version = "1.52.3", features = ["full"] }
tokio-websockets = { version = "0.13.2", features = ["client", "fastrand", "server", "sha1_smol"] }

Требуемые API

Вам понадобятся следующие функции из tokio и tokio_websockets. Не поленитесь освоить API.

• StreamExt::next(), реализованный для WebSocketStream: для асинхронного чтения сообщений из WebSocket-потока.
• SinkExt::send(), реализованный для WebSocketStream: для асинхронной отправки сообщений в WebSocket-поток.
• Lines::next_line(): для асинхронного чтения сообщений пользователя из стандартного ввода.
• Sender::subscribe(): для подписки на широковещательный канал.

Два бинарных файла

Обычно в проекте Cargo может быть только один бинарный файл и один файл src/main.rs. В этом проекте нам нужны два бинарных файла: один для клиента и один для сервера. Можно было бы сделать их двумя отдельными проектами Cargo, но мы поместим их в один проект Cargo с двумя бинарными файлами. Чтобы это работало, код клиента и сервера должен находиться в src/bin (см. документацию).

Скопируйте следующий код сервера и клиента в src/bin/server.rs и src/bin/client.rs соответственно. Ваша задача — дополнить эти файлы, как описано ниже.

src/bin/server.rs:

// Copyright 2024 Google LLC
// SPDX-License-Identifier: Apache-2.0

use futures_util::sink::SinkExt;
use futures_util::stream::StreamExt;
use std::error::Error;
use std::net::SocketAddr;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast::{Sender, channel};
use tokio_websockets::{Message, ServerBuilder, WebSocketStream};

async fn handle_connection(
    addr: SocketAddr,
    mut ws_stream: WebSocketStream<TcpStream>,
    bcast_tx: Sender<String>,
) -> Result<(), Box<dyn Error + Send + Sync>> {

    // TODO: Подсказку см. в описании задачи ниже.

}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
    let (bcast_tx, _) = channel(16);

    let listener = TcpListener::bind("127.0.0.1:2000").await?;
    println!("listening on port 2000");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("New connection from {addr:?}");
        let bcast_tx = bcast_tx.clone();
        tokio::spawn(async move {
            // Оборачиваем сырой TCP-поток в websocket.
            let (_req, ws_stream) = ServerBuilder::new().accept(socket).await?;

            handle_connection(addr, ws_stream, bcast_tx).await
        });
    }
}

src/bin/client.rs:

// Copyright 2024 Google LLC
// SPDX-License-Identifier: Apache-2.0

use futures_util::SinkExt;
use futures_util::stream::StreamExt;
use http::Uri;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio_websockets::{ClientBuilder, Message};

#[tokio::main]
async fn main() -> Result<(), tokio_websockets::Error> {
    let (mut ws_stream, _) =
        ClientBuilder::from_uri(Uri::from_static("ws://127.0.0.1:2000"))
            .connect()
            .await?;

    let stdin = tokio::io::stdin();
    let mut stdin = BufReader::new(stdin).lines();


    // TODO: Подсказку см. в описании задачи ниже.

}

Запуск бинарных файлов

Запустите сервер командой:

cargo run --bin server

а клиент командой:

cargo run --bin client

Задачи

• Реализуйте функцию handle_connection в src/bin/server.rs.
  • Подсказка: используйте tokio::select!, чтобы одновременно выполнять две задачи в непрерывном цикле. Одна задача получает сообщения от клиента и рассылает их. Другая отправляет клиенту сообщения, полученные сервером.
• Дополните функцию main в src/bin/client.rs.
  • Подсказка: как и раньше, используйте tokio::select! в непрерывном цикле для одновременного выполнения двух задач: (1) чтение сообщений пользователя из стандартного ввода и отправка их на сервер, и (2) получение сообщений от сервера и их отображение пользователю.
• Необязательно: когда закончите, измените код так, чтобы сообщения рассылались всем клиентам, кроме отправителя.