🐙 tako
Transports

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.

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.

examples/websocket/src/main.rs
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.

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();
}
examples/websocket-compio/src/main.rs
//! 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:

MethodDefaultEffect
.protocols(list)noneSubprotocol allow-list; the first server-preferred match the client offers is echoed back.
.allowed_origins(list)any originReject upgrades whose Origin is not on the list with 403.
.max_frame_size(n)16 MiBLargest frame payload; a larger frame closes the connection with 1009.
.max_message_size(n)64 MiBLargest reassembled message; a larger one closes the connection with 1009.
.read_buffer_size(n)4 KiBInitial capacity, from 1 byte to 64 KiB; the buffer grows with incoming frame data.
.write_buffer_size(n)64 KiBQueued bytes past which send() writes at once.
.upgrade_timeout(d)noneGive up if the server never hands over the upgraded connection.
.max_lifetime(d)noneClose with 1001 once the connection has been open this long.
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.

  • Routing — registering the upgrade handler as a route.
  • HTTP — the listeners WebSocket rides on.
  • Streams — SSE and other streaming responses.
  • Feature flags — the ws feature.

Last updated on

HTTP/3

Serve HTTP/3 over QUIC on Tokio (quinn) or Compio (compio-quic) with mandatory TLS, congestion control, and stream caps.

WebTransport

Browser-compatible W3C WebTransport over HTTP/3 on Tokio or Compio, routed through the Router, plus a raw QUIC session helper on Tokio.

On this page

WebSocketMessagesSending and batchingCompioConfigurationTLS listenersRelated