# WebSocket

> Upgrade an HTTP request to a WebSocket inside a handler and run the message loop, on either runtime.

WebSocket is not a separate listener: it lives inside an HTTP handler on an
HTTP/1.1 listener, optionally over TLS. `WebSocketUpgrade::new(req, handler)`
answers the RFC 6455 handshake and runs `handler` on the upgraded connection.
Enable the `ws` feature; together with `compio` the same API runs on Compio.

```rust
use tako::Method;
use tako::responder::Responder;
use tako::types::Request;
use tako::ws::WebSocketUpgrade;

async fn ws_echo(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    while let Some(Ok(message)) = ws.recv().await {
      if (message.is_text() || message.is_binary()) && ws.send(message).await.is_err() {
        break;
      }
    }
  })
}

#[tokio::main]
async fn main() {
  let listener = tokio::net::TcpListener::bind("127.0.0.1:8080")
    .await
    .unwrap();
  let mut router = tako::router::Router::new();
  router.route(Method::GET, "/ws/echo", ws_echo);
  tako::Server::builder().build().spawn_http(listener, router).result().await.unwrap();
}
```

The handler is an ordinary route. The `101 Switching Protocols` response is
produced for you, and the connection runs on its own task once the handshake
completes.

Source: [examples/websocket/src/main.rs](https://github.com/rust-dd/tako/blob/main/examples/websocket/src/main.rs)

```rust
use std::time::Duration;

use tako::Method;
use tako::responder::Responder;
use tako::types::Request;
use tako::ws::Message;
use tako::ws::WebSocketUpgrade;

pub async fn ws_echo(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    if ws.send(Message::text("Welcome to Tako WS!")).await.is_err() {
      return;
    }
    while let Some(Ok(message)) = ws.recv().await {
      let reply = match message {
        Message::Text(text) => Message::text(format!("Echo: {text}")),
        Message::Binary(data) => Message::Binary(data),
        _ => continue,
      };
      if ws.send(reply).await.is_err() {
        break;
      }
    }
  })
}

pub async fn ws_tick(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    let mut ticker = tokio::time::interval(Duration::from_secs(1));
    let mut tick = 0u64;
    loop {
      tokio::select! {
        message = ws.recv() => match message {
          Some(Ok(_)) => {}
          _ => break,
        },
        _ = ticker.tick() => {
          if ws.send(Message::text(format!("tick #{tick}"))).await.is_err() {
            break;
          }
          tick += 1;
        }
      }
    }
  })
}

#[tokio::main]
async fn main() {
  let listener = tokio::net::TcpListener::bind("127.0.0.1:8080")
    .await
    .unwrap();
  let mut router = tako::router::Router::new();

  router.route(Method::GET, "/ws/echo", ws_echo);
  router.route(Method::GET, "/ws/tick", ws_tick);

  tako::Server::builder()
    .build()
    .spawn_http(listener, router)
    .result()
    .await
    .expect("HTTP server failed");
}
```

## Messages

`recv()` returns `Option<Result<Message, Error>>`, and `None` once the
connection is closed. It is cancel-safe, so it can race other futures in
`tokio::select!`. A `Message` is `Text`, `Binary`, `Ping`, `Pong` or `Close`.
tako answers pings and the closing handshake itself, so a handler only handles
the messages it cares about. Extensions such as permessage-deflate are not
negotiated.

## Sending and batching

`send()` queues a message. Queued messages go out together, in one write, when
the handler waits for anything (the next `recv()`, a timer, a channel), when
more than `write_buffer_size` bytes are queued, or on `flush()`. An echo loop
that receives 16 pipelined messages answers them with a single write. Called
from another task, for example after `split()`, `send()` writes at once.

`WebSocket` also implements `Stream` and `Sink`, so `split`, `forward` and the
other combinators work. Through `Sink`, `poll_flush` always writes.

When the handler returns and no `WebSocket` handle is left, tako closes the
connection with code 1000. A handler that panics closes it with 1011.

## Compio

With the `compio` and `ws` features the same code runs on Compio. `WebSocket` is
then `!Send`, and the handler runs on the connection's thread.

```rust
use tako::Method;
use tako::responder::Responder;
use tako::types::Request;
use tako::ws::WebSocketUpgrade;

async fn ws_echo(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    while let Some(Ok(message)) = ws.recv().await {
      if (message.is_text() || message.is_binary()) && ws.send(message).await.is_err() {
        break;
      }
    }
  })
}

#[compio::main]
async fn main() {
  let listener = compio::net::TcpListener::bind("127.0.0.1:8080")
    .await
    .unwrap();
  let mut router = tako::router::Router::new();
  router.route(Method::GET, "/ws/echo", ws_echo);
  tako::CompioServer::builder().build().spawn_http(listener, router).result().await.unwrap();
}
```

Source: [examples/websocket-compio/src/main.rs](https://github.com/rust-dd/tako/blob/main/examples/websocket-compio/src/main.rs)

```rust
//! WebSocket echo and counter on the compio runtime.
//!
//! Run with: cargo run --manifest-path examples/websocket-compio/Cargo.toml

use std::time::Duration;

use tako::Method;
use tako::responder::Responder;
use tako::types::Request;
use tako::ws::Message;
use tako::ws::WebSocketUpgrade;

pub async fn ws_echo(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    if ws
      .send(Message::text("Welcome to Tako WS (compio)!"))
      .await
      .is_err()
    {
      return;
    }
    while let Some(Ok(message)) = ws.recv().await {
      let reply = match message {
        Message::Text(text) => Message::text(format!("Echo: {text}")),
        Message::Binary(data) => Message::Binary(data),
        _ => continue,
      };
      if let Err(error) = ws.send(reply).await {
        tracing::error!("Failed to send echo: {error}");
        break;
      }
    }
  })
}

pub async fn ws_count(req: Request) -> impl Responder {
  WebSocketUpgrade::new(req, |mut ws| async move {
    for count in 0u64.. {
      if ws.send(Message::text(format!("count: {count}"))).await.is_err() {
        break;
      }
      compio::time::sleep(Duration::from_secs(1)).await;
    }
  })
}

#[compio::main]
async fn main() {
  let listener = compio::net::TcpListener::bind("127.0.0.1:8080")
    .await
    .unwrap();
  let mut router = tako::router::Router::new();

  router.route(Method::GET, "/ws/echo", ws_echo);
  router.route(Method::GET, "/ws/count", ws_count);

  println!("WebSocket server running at:");
  println!("  - ws://127.0.0.1:8080/ws/echo");
  println!("  - ws://127.0.0.1:8080/ws/count");
  tako::CompioServer::builder()
    .build()
    .spawn_http(listener, router)
    .result()
    .await
    .expect("HTTP server failed");
}
```

## Configuration

`WebSocketUpgrade` is a builder. Each option is a chained method before the
handler runs:

| Method                   | Default    | Effect                                                                                     |
| ------------------------ | ---------- | ------------------------------------------------------------------------------------------ |
| `.protocols(list)`       | none       | Subprotocol allow-list; the first server-preferred match the client offers is echoed back. |
| `.allowed_origins(list)` | any origin | Reject upgrades whose `Origin` is not on the list with `403`.                              |
| `.max_frame_size(n)`     | 16 MiB     | Largest frame payload; a larger frame closes the connection with 1009.                     |
| `.max_message_size(n)`   | 64 MiB     | Largest reassembled message; a larger one closes the connection with 1009.                 |
| `.read_buffer_size(n)`   | 4 KiB      | Initial capacity, from 1 byte to 64 KiB; the buffer grows with incoming frame data.        |
| `.write_buffer_size(n)`  | 64 KiB     | Queued bytes past which `send()` writes at once.                                           |
| `.upgrade_timeout(d)`    | none       | Give up if the server never hands over the upgraded connection.                            |
| `.max_lifetime(d)`       | none       | Close with 1001 once the connection has been open this long.                               |

```rust
use std::time::Duration;
use tako::ws::WebSocketUpgrade;

# fn _doc(req: tako::types::Request) -> impl tako::responder::Responder {
WebSocketUpgrade::new(req, |ws| async move { /* ... */ })
  .protocols(["chat", "superchat"])
  .max_message_size(1 << 20)
  .allowed_origins(["https://app.example.com"])
  .upgrade_timeout(Duration::from_secs(10))
  .max_lifetime(Duration::from_secs(3600))
# }
```

Pings are answered while the handler reads. To detect idle peers, send your own
pings from the handler; `max_lifetime` caps the total lifetime and does not
detect idle peers.

## TLS listeners

`examples/websocket-tls` serves wss on a TLS listener. Upgrades negotiate
HTTP/1.1; HTTP/2 Extended CONNECT (RFC 8441) is not supported. Invalid upgrade
methods return 405, unsupported WebSocket versions return 426 with
`Sec-WebSocket-Version: 13`, and malformed handshake headers or keys return 400.

## Related

* [Routing](/docs/routing) — registering the upgrade handler as a route.
* [HTTP](/docs/transports/http) — the listeners WebSocket rides on.
* [Streams](/docs/streams) — SSE and other streaming responses.
* [Feature flags](/docs/reference/features) — the `ws` feature.

Source: https://tako.rust-dd.com/docs/transports/websocket