Snippet stash

Websockets_Rust_example

WebSocket — это протокол, позволяющий устанавливать постоянное двустороннее соединение между клиентом и сервером. Ключевые моменты:

• Двусторонняя связь: клиент и сервер могут отправлять сообщения друг другу в любое время, без необходимости ждать запроса.
• Одно соединение: нет постоянных открытых HTTP-запросов, как при long polling.
• Низкая задержка: подходит для чатов, игр, финансовых данных в реальном времени.
• Протокол поверх TCP: использует «handshake» через HTTP для установления соединения, затем переключается на собственный протокол ws:// или wss:// (шифрованный).
• Сообщения: могут быть текстовыми или бинарными (например, JSON, Protobuf, изображения и т.д.).
• Событийная модель: обычно обрабатываются события открытия соединения, получения сообщений и закрытия соединения.

WebSocket (RFC 6455)

• c410-f3r/wtx — клиент и сервер с поддержкой шифрования.
• housleyjk/ws-rs — лёгкая, событийно-ориентированная реализация WebSocket.
• iddm/urlshortener-rs — очень простая библиотека для сокращения URL.
• ratchet, ratchet_rs — Ratchet — это быстрая, лёгкая и полностью асинхронная реализация протокола WebSocket с поддержкой расширений и Deflate-сжатия.
• rust-websocket — фреймворк для работы с WebSocket-соединениями (как клиентами, так и серверами).
• snapview/tungstenite-rs — лёгкая потоковая реализация WebSocket.
• vi/websocat — CLI-инструмент для работы с WebSocket’ами, сочетающий функциональность Netcat, Curl и Socat.

awesome-rust#web-programming

Сравнение библиотек WebSocket для Rust

Итог:

• Если нужен CLI → websocat.
• Если нужен чистый sync → tungstenite или ws-rs.
• Если нужен асинхронный runtime → ratchet или async-tungstenite.
• Если нужна гибридная поддержка (sync + async) → rust-websocket.
• Если нужен сильный упор на безопасность/шифрование → wtx.

Статья Павла Колодки Efficient Client-Server Communication: An Overview of Protocols and Techniques

Обзор различных методов и протоколов, используемых для эффективной коммуникации между клиентом и сервером в веб-разработке и реальном времени. Автор рассматривает такие технологии, как HTTP, polling, webhooks, SSE и WebSockets, объясняя их принципы работы, преимущества и недостатки.

Основные моменты статьи

1. HTTP

HTTP (Hypertext Transfer Protocol) — это полудуплексный протокол, используемый для обмена данными между веб-браузерами (клиентами) и веб-серверами. Он служит основой для обмена данными в Всемирной паутине. HTTP работает поверх транспортного протокола TCP и использует порты 80 и 443 для незащищенных и защищенных соединений соответственно. Это классический пример синхронного взаимодействия клиент-сервер, когда клиент инициирует запрос, ожидая результат, а сервер отвечает соответствующим образом.

2. Polling

• Short polling: Клиент периодически отправляет запросы на сервер с определенным интервалом, чтобы проверить наличие новых данных.
• Long polling: Клиент отправляет запрос на сервер, и сервер удерживает соединение открытым до тех пор, пока не появятся новые данные или не истечет тайм-аут. После получения данных сервер отправляет ответ, и клиент немедленно отправляет новый запрос.

3. Webhook

Webhook — это способ, при котором сервер отправляет данные клиенту (или другому серверу) в ответ на определенное событие, без необходимости клиента запрашивать эти данные. Это позволяет реализовать асинхронную коммуникацию, где сервер уведомляет клиента о произошедших изменениях.

4. Server-Sent Events (SSE)

SSE — это технология, позволяющая серверу отправлять обновления клиенту через однонаправленное соединение HTTP. Это удобно для приложений, требующих получения обновлений в реальном времени, таких как ленты новостей или уведомления.

5. WebSocket

WebSocket — это протокол, обеспечивающий двустороннюю связь между клиентом и сервером через одно постоянное соединение. Это особенно полезно для приложений, требующих низкой задержки и высокой частоты обновлений, таких как онлайн-игры или чаты.

crate tokio-tungstenite

• Асинхронная библиотека на базе Tokio.
• Позволяет создавать WebSocket-клиенты и серверы.
• Подходит для high-performance приложений с большим количеством соединений.

use tokio::net::TcpListener;
use tokio_tungstenite::accept_async;
use futures_util::{StreamExt, SinkExt};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:9001").await.unwrap();
    while let Ok((stream, _)) = listener.accept().await {
        tokio::spawn(async move {
            let ws_stream = accept_async(stream).await.unwrap();
            let (mut write, mut read) = ws_stream.split();

            while let Some(msg) = read.next().await {
                let msg = msg.unwrap();
                write.send(msg).await.unwrap(); // echo back
            }
        });
    }
}

warp + warp::ws

• Warp — это современный веб-фреймворк на Rust.
• Имеет встроенную поддержку WebSocket через warp::ws.
• Удобно интегрировать в REST API и микросервисы.

Подход с warp особенно удобен, если нужно сочетать WebSocket с обычными HTTP эндпоинтами.

use warp::Filter;

#[tokio::main]
async fn main() {
    let ws_route = warp::path("ws")
        .and(warp::ws())
        .map(|ws: warp::ws::Ws| {
            ws.on_upgrade(|socket| async move {
                let (mut tx, mut rx) = socket.split();
                while let Some(msg) = rx.next().await {
                    let msg = msg.unwrap();
                    tx.send(msg).await.unwrap(); // echo
                }
            })
        });

    warp::serve(ws_route).run(([127,0,0,1], 3030)).await;
}

Как использовать асинхронную базу данных из обработчика WebSocket.

type PgPool = deadpool_r2d2::Pool;

struct WsMessage(String)

impl Handler for MyWebSocket {
    type Result = ();

    fn handle(&mut self, msg: WsMessage, ctx: &mut Context) -> Self::Result {
        ctx.text(msg.0);
    }
}

pub struct MyWebSocket {
    db: deadpool::managed::Object    
}

impl MyWebSocket {
    pub fn new(client_db: deadpool::managed::Object) -> Self {
        Self { db:client_db }
    }
}

impl Actor for MyWebSocket {
    type Context = ws::WebsocketContext;
}

impl StreamHandler> for MyWebSocket {
    fn handle(&mut self, msg: Result, ctx: &mut Self::Context) {
        match msg {
         Ok(ws::Message::Text(text)) => {
            // Отправить ответ в обработчике WsMessage
              let db = self.db.clone();
              let addr = ctx.addess();
              let fut = async move {
                let stmt = db.prepare_cached("SELECT 1 + $1").await.unwrap();
                let rows = db.query(&stmt, &[&3]).await.unwrap();
                let value: i32 = rows[0].get(0);
                addr.send(WsMessage(String::from(value))).await.unwrap();
             };
             let fut = actix::fut::wrap_future::<_, Self>(fut);
             ctx.spawn(fut);

            ИЛИ
            // Отправить ответ сразу используя  ActorFuture
             let db = self.db.clone();

             let fut = async move {
                let stmt = db.prepare_cached("SELECT 1 + $1").await.unwrap();
                let rows = db.query(&stmt, &[&3]).await.unwrap();
                let value: i32 = rows[0].get(0);
                value
             };
             let fut = actix::fut::wrap_future::<_, Self>(fut);
             let fut = fut.map(|result, actor, ctx| {
                 ctx.text(result.to_string());
             });
            ctx.spawn(fut)

         },
         ....
        }
    }
}
fn create_pool(max_size: usize) -> PgPool {
    let config:tokio_postgres::Config = config().expect("Error configure");
    let mgr_config = ManagerConfig {
        recycling_method: RecyclingMethod::Fast
    };
    let mgr:deadpool_postgres::Manager = Manager::from_config(config, NoTls, mgr_config);
    let pool:deadpool_r2d2::Pool = 
        Pool::builder(mgr).runtime(deadpool_postgres::Runtime::Tokio1).max_size(max_size).build().unwrap();
    pool
} 

#[get("ws/")]
async fn ws_index(req: HttpRequest, stream: web::Payload,db_pool: web::Data) -> Result {
    let client:deadpool::managed::Object = db_pool.get().await.unwrap();
    let resp = ws::start(MyWebSocket::new(client), &req, stream);
    resp
}

#[actix_web::main]
async fn main() -> std::io::Result<()> { 
    let pool:PgPool = create_pool(2);
    HttpServer::new(move|| {
        App::new()
            .app_data(web::Data::new(pool.clone()))
            .wrap( middleware::DefaultHeaders::new().header("Access-Control-Allow-Origin", "*"))
            .service(ws_index)
            .wrap(middleware::Logger::default())
    })
    .workers(2)
    .bind(("0.0.0.0", 4011))?
    .run()
    .await
}