# Streams

> Server-Sent Events, WebSocket, ranged file streaming, and WebTransport over HTTP/3 in tako.

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.

```rust
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](/docs/transports/sse) 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`:

```rust
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](/docs/transports/websocket) 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.

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

```rust
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](/docs/transports/webtransport).

See also:

* [`examples/streams`](https://github.com/rust-dd/tako/tree/main/examples/streams)
  and [`examples/http3-sse`](https://github.com/rust-dd/tako/tree/main/examples/http3-sse)
  for SSE,
* [`examples/websocket`](https://github.com/rust-dd/tako/tree/main/examples/websocket),
  [`examples/websocket-tls`](https://github.com/rust-dd/tako/tree/main/examples/websocket-tls),
  and [`examples/websocket-compio`](https://github.com/rust-dd/tako/tree/main/examples/websocket-compio)
  for WebSocket,
* [`examples/file-stream`](https://github.com/rust-dd/tako/tree/main/examples/file-stream)
  for ranged file serving,
* [`examples/webtransport`](https://github.com/rust-dd/tako/tree/main/examples/webtransport)
  for browser WebTransport, and
  [`examples/raw-quic`](https://github.com/rust-dd/tako/tree/main/examples/raw-quic)
  for raw QUIC sessions.

<Callout type="info">
  Multipart byteranges and Linux `sendfile(2)` are deferred follow-up
  items.
</Callout>

Source: https://tako.rust-dd.com/docs/streams