Streams

Tako's streaming surface covers Server-Sent Events, WebSocket, file streaming with range / conditional GET, and WebTransport over HTTP/3. All four live in the tako-rs-streams crate and are re-exported under tako::*.

Server-Sent Events

Sse::new(stream) accepts any Stream<Item = Into<Bytes>> and frames it as text/event-stream. Default headers include Cache-Control: no-cache plus X-Accel-Buffering: no so reverse proxies do not buffer the response.

use bytes::Bytes;
use futures_util::{StreamExt, stream};
use tako::Method;
use tako::responder::Responder;
use tako::router::Router;
use tako::sse::Sse;

async fn ticker() -> impl Responder {
  let s = stream::unfold(0u64, |i| async move {
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    Some((Bytes::from(format!("tick: {i}")), i + 1))
  });
  Sse::new(s)
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
  let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
  let mut router = Router::new();
  router.route(Method::GET, "/events", ticker);
  tako::Server::builder().build().spawn_http(listener, router)
    .result().await.expect("HTTP server failed");
  Ok(())
}

For structured SSE with event:, id:, and retry: fields, use Sse::events(stream) plus the SseEvent builder, and call .keep_alive(Duration::from_secs(15)) to emit :keep-alive comments during idle periods. tako::sse::last_event_id(headers) reads the Last-Event-ID header sent by EventSource clients on reconnect.

See SSE transport for the full transport reference.

SSE requires the sse feature.

WebSocket

WebSocketUpgrade::new(req, handler) performs the upgrade from inside an HTTP handler and runs handler on the upgraded WebSocket:

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() && ws.send(message).await.is_err() {
        break;
      }
    }
  })
}

Enable ws; together with compio it runs on Compio. Configure subprotocols, frame and message limits, buffer sizes, allowed origins, and the upgrade timeout on the builder. tako answers pings and the closing handshake; permessage-deflate is not provided.

See WebSocket transport for the full transport reference.

File serving

tako::file_stream::FileStream (feature file-stream) streams an open file and provides conditional and single-range response helpers. Metadata validators are weak ETags. tako::r#static::ServeDirBuilder and ServeFile stream assets, handle HEAD and conditional/range requests, and optionally select .br / .gz sidecars. On Tokio, files up to 128 KiB are read whole in the blocking call that opens them; larger files stream in 64 KiB chunks. Dotfiles are denied by default. Use an explicit fallback file path and keep served roots read-only to untrusted processes.

use anyhow::Result;
use tako::Method;
use tako::file_stream::FileStream;
use tako::responder::Responder;
use tako::router::Router;
use tako::types::Request;
use tokio::fs::File;
use tokio::net::TcpListener;
use tokio_util::io::ReaderStream;

async fn serve_file(_: Request) -> impl Responder {
  let file = File::open("test.txt").await.unwrap();
  let stream = ReaderStream::new(file);
  let file_stream = FileStream::new(stream, Some("test.txt".to_string()), None);
  file_stream.into_response()
}

#[tokio::main]
async fn main() -> Result<()> {
  let listener = TcpListener::bind("127.0.0.1:8080").await?;

  let mut router = Router::new();
  router.route(Method::GET, "/file", serve_file);

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

WebTransport

W3C WebTransport sessions run on the HTTP/3 server behind the webtransport feature, on either runtime: a CONNECT route takes the WebTransport extractor and gets streams and datagrams that browsers can reach. See WebTransport.

See also:

• examples/streams and examples/http3-sse for SSE,
• examples/websocket, examples/websocket-tls, and examples/websocket-compio for WebSocket,
• examples/file-stream for ranged file serving,
• examples/webtransport for browser WebTransport, and examples/raw-quic for raw QUIC sessions.