Streaming Responses Guide
Armature provides comprehensive support for streaming HTTP responses, enabling efficient delivery of large data sets, real-time data, and chunked transfers.
Table of Contents
- Overview
- Features
- Basic Usage
- Stream Types
- StreamingResponse
- Helper Functions
- Progress Tracking
- Integration Examples
- Best Practices
- API Reference
- Summary
Overview
Traditional HTTP responses buffer the entire response body in memory before sending. This works well for small responses but becomes problematic for:
- Large files: Buffering a 1GB file requires 1GB of memory
- Real-time data: Data must wait until generation completes
- Long-running queries: Users see nothing until everything is ready
Streaming responses solve these problems by sending data as it becomes available using HTTP chunked transfer encoding.
Features
- โ Chunked transfer encoding
- โ Async stream-based response bodies
- โ Multiple stream types (bytes, JSON, text)
- โ NDJSON (Newline-Delimited JSON) support
- โ Progress tracking and callbacks
- โ File streaming utilities
- โ Iterator-to-stream helpers
- โ Backpressure handling
Basic Usage
Simple Byte Stream
use armature_core::streaming::{ByteStream, StreamingResponse};
async fn stream_data() -> StreamingResponse {
let (stream, sender) = ByteStream::new();
// Spawn a task to produce data
tokio::spawn(async move {
for i in 0..100 {
sender.send(format!("Line {}\n", i).into_bytes()).await.ok();
}
sender.close().await;
});
StreamingResponse::new(stream)
.content_type("text/plain")
.no_cache()
}
JSON Streaming (NDJSON)
use armature_core::streaming::{JsonStream, StreamingResponse};
use serde::Serialize;
#[derive(Serialize)]
struct User {
id: u64,
name: String,
}
async fn stream_users() -> StreamingResponse {
let (stream, sender) = JsonStream::new();
tokio::spawn(async move {
let users = load_users_from_database().await;
for user in users {
sender.send_json(&user).await.ok();
}
sender.close().await;
});
StreamingResponse::ndjson(stream)
}
Stream Types
ByteStream
The foundational stream type for raw binary data.
use armature_core::streaming::{ByteStream, ByteStreamSender};
// Create with default buffer size (64)
let (stream, sender) = ByteStream::new();
// Create with custom buffer size
let (stream, sender) = ByteStream::with_buffer_size(256);
ByteStreamSender Methods
| Method | Description |
|---|---|
send(data) |
Send bytes (Vec, &[u8], etc.) |
send_bytes(bytes) |
Send a Bytes object |
send_str(s) |
Send a string slice |
send_error(msg) |
Signal an error |
close() |
Close the stream |
bytes_sent() |
Get total bytes sent |
is_closed() |
Check if receiver dropped |
JsonStream
For streaming JSON objects as NDJSON (newline-delimited JSON).
use armature_core::streaming::{JsonStream, JsonStreamSender};
let (stream, sender) = JsonStream::new();
// Send a serializable value
sender.send_json(&my_struct).await?;
// Send raw JSON string
sender.send_raw(r#"{"key": "value"}"#).await?;
// Send an error as JSON
sender.send_error("Something went wrong").await?;
NDJSON format:
{"id":1,"name":"Alice"}
{"id":2,"name":"Bob"}
{"id":3,"name":"Charlie"}
TextStream
For streaming text lines.
use armature_core::streaming::{TextStream, TextStreamSender};
let (stream, sender) = TextStream::new();
// Send a line (newline added automatically)
sender.send_line("First line").await?;
// Send raw text (no newline)
sender.send("Some text").await?;
StreamingResponse
The main response type for streaming.
Creating Responses
use armature_core::streaming::{StreamingResponse, ByteStream, JsonStream, TextStream};
// From ByteStream
let (byte_stream, _) = ByteStream::new();
let response = StreamingResponse::new(byte_stream);
// From JsonStream (NDJSON)
let (json_stream, _) = JsonStream::new();
let response = StreamingResponse::ndjson(json_stream);
// From TextStream
let (text_stream, _) = TextStream::new();
let response = StreamingResponse::text(text_stream);
// Empty response
let response = StreamingResponse::empty();
Response Configuration
let response = StreamingResponse::new(stream)
.status(200) // Set HTTP status
.content_type("application/octet-stream") // Set Content-Type
.header("X-Custom", "value") // Add custom header
.no_cache() // Disable caching
.cors("*") // Enable CORS
.nosniff(); // X-Content-Type-Options: nosniff
Converting to Buffered Response
If you need to convert a streaming response to a regular HttpResponse (not recommended for large streams):
let buffered = streaming_response.into_buffered().await?;
// buffered is now an HttpResponse with full body in memory
Helper Functions
stream_iter
Stream items from a synchronous iterator:
use armature_core::streaming::stream_iter;
let items = vec![1, 2, 3, 4, 5];
let (stream, handle) = stream_iter(
items.into_iter(),
|i| format!("Item: {}\n", i).into_bytes()
);
stream_iter_with_delay
Stream items with a delay between each:
use armature_core::streaming::stream_iter_with_delay;
use std::time::Duration;
let items = vec!["a", "b", "c"];
let (stream, handle) = stream_iter_with_delay(
items.into_iter(),
|s| s.as_bytes().to_vec(),
Duration::from_millis(100)
);
stream_json_iter
Stream JSON items from an iterator:
use armature_core::streaming::stream_json_iter;
let users = vec![
User { id: 1, name: "Alice".into() },
User { id: 2, name: "Bob".into() },
];
let (stream, handle) = stream_json_iter(users.into_iter());
stream_reader
Stream data from an async reader (files, network sockets):
use armature_core::streaming::stream_reader;
use tokio::fs::File;
let file = File::open("large_file.bin").await?;
let (stream, handle) = stream_reader(file, 8192); // 8KB chunks
let response = StreamingResponse::new(stream)
.content_type("application/octet-stream");
Progress Tracking
Track streaming progress with ProgressStream:
use armature_core::streaming::{ByteStream, ProgressStream};
let (stream, sender) = ByteStream::new();
let progress_stream = ProgressStream::new(stream)
.on_progress(|bytes_received| {
println!("Received {} bytes", bytes_received);
});
// Later, check total bytes
let total = progress_stream.bytes_received();
Integration Examples
File Download with Progress
use armature_core::streaming::{stream_reader, StreamingResponse};
use tokio::fs::File;
async fn download_file(filename: &str) -> Result<StreamingResponse, Error> {
let file = File::open(filename).await?;
let metadata = file.metadata().await?;
let file_size = metadata.len();
let (stream, _) = stream_reader(file, 64 * 1024); // 64KB chunks
Ok(StreamingResponse::new(stream)
.content_type("application/octet-stream")
.header("Content-Length", file_size.to_string())
.header("Content-Disposition", format!("attachment; filename=\"{}\"", filename)))
}
Database Cursor Streaming
use armature_core::streaming::{JsonStream, StreamingResponse};
async fn stream_query_results() -> StreamingResponse {
let (stream, sender) = JsonStream::new();
tokio::spawn(async move {
let mut cursor = database.query("SELECT * FROM large_table").await.unwrap();
while let Some(row) = cursor.next().await {
match row {
Ok(record) => {
if sender.send_json(&record).await.is_err() {
break; // Client disconnected
}
}
Err(e) => {
sender.send_error(e.to_string()).await.ok();
break;
}
}
}
sender.close().await;
});
StreamingResponse::ndjson(stream)
}
Real-Time Log Streaming
use armature_core::streaming::{TextStream, StreamingResponse};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
async fn stream_logs() -> StreamingResponse {
let (stream, sender) = TextStream::new();
tokio::spawn(async move {
let mut child = Command::new("tail")
.args(["-f", "/var/log/app.log"])
.stdout(std::process::Stdio::piped())
.spawn()
.unwrap();
let stdout = child.stdout.take().unwrap();
let mut reader = BufReader::new(stdout).lines();
while let Ok(Some(line)) = reader.next_line().await {
if sender.send_line(&line).await.is_err() {
break;
}
}
child.kill().await.ok();
});
StreamingResponse::text(stream)
.no_cache()
}
Chunked JSON Array
use armature_core::streaming::{ByteStream, StreamingResponse};
async fn stream_json_array() -> StreamingResponse {
let (stream, sender) = ByteStream::new();
tokio::spawn(async move {
// Send opening bracket
sender.send_str("[").await.ok();
let items = fetch_items().await;
let mut first = true;
for item in items {
if !first {
sender.send_str(",").await.ok();
}
first = false;
let json = serde_json::to_string(&item).unwrap();
sender.send_str(&json).await.ok();
}
// Send closing bracket
sender.send_str("]").await.ok();
sender.close().await;
});
StreamingResponse::new(stream)
.content_type("application/json")
}
Best Practices
1. Handle Client Disconnection
Always check if the client is still connected:
if sender.send(data).await.is_err() {
// Client disconnected, stop producing data
break;
}
Or check explicitly:
if sender.is_closed() {
break;
}
2. Use Appropriate Buffer Sizes
// Small, frequent updates (real-time)
let (stream, sender) = ByteStream::with_buffer_size(16);
// Large file transfers
let (stream, sender) = ByteStream::with_buffer_size(256);
3. Always Close Streams
Ensure streams are properly closed to signal completion:
tokio::spawn(async move {
// ... produce data ...
sender.close().await; // Important!
});
4. Set Cache Headers
Streaming responses should typically not be cached:
StreamingResponse::new(stream)
.no_cache()
5. Consider Content-Type
Set appropriate content types:
| Format | Content-Type |
|---|---|
| NDJSON | application/x-ndjson |
| JSON Lines | application/jsonl |
| Plain text | text/plain; charset=utf-8 |
| Binary | application/octet-stream |
| CSV | text/csv |
6. Error Handling
Propagate errors gracefully:
match process_item(&item) {
Ok(data) => sender.send(data).await.ok(),
Err(e) => {
sender.send_error(e.to_string()).await.ok();
break;
}
}
Common Pitfalls
โ Forgetting to Close
// Bad: Stream never ends
tokio::spawn(async move {
for item in items {
sender.send(item).await.ok();
}
// Missing: sender.close().await;
});
โ Ignoring Backpressure
// Bad: Ignores send errors
for item in items {
let _ = sender.send(item).await; // Client may have disconnected
}
// Good: Respect backpressure
for item in items {
if sender.send(item).await.is_err() {
break;
}
}
โ Blocking the Stream Producer
// Bad: Blocking I/O in async context
tokio::spawn(async move {
let data = std::fs::read("file.txt").unwrap(); // Blocks!
sender.send(data).await.ok();
});
// Good: Use async I/O
tokio::spawn(async move {
let data = tokio::fs::read("file.txt").await.unwrap();
sender.send(data).await.ok();
});
API Reference
Types
| Type | Description |
|---|---|
StreamingResponse |
Main streaming response type |
StreamBody |
Enum of stream body variants |
StreamChunk |
A chunk of streaming data |
ByteStream |
Stream of raw bytes |
ByteStreamSender |
Sender for byte streams |
JsonStream |
Stream of JSON objects (NDJSON) |
JsonStreamSender |
Sender for JSON streams |
TextStream |
Stream of text lines |
TextStreamSender |
Sender for text streams |
ProgressStream |
Progress-tracking wrapper |
Functions
| Function | Description |
|---|---|
stream_iter(iter, transform) |
Stream from iterator |
stream_iter_with_delay(iter, transform, delay) |
Stream with delays |
stream_json_iter(iter) |
Stream JSON from iterator |
stream_reader(reader, chunk_size) |
Stream from async reader |
Summary
Key Points:
- Use streaming for large/real-time data - Don't buffer when you can stream
- Choose the right stream type -
ByteStream,JsonStream, orTextStream - Handle disconnection - Check send results and
is_closed() - Close streams properly - Call
close()when done - Set appropriate headers - Content-Type and Cache-Control
- Consider backpressure - Don't overwhelm the client
When to Use Streaming:
- File downloads/uploads
- Database query results
- Real-time logs/events
- Large JSON responses
- Any response > 1MB
When NOT to Use Streaming:
- Small, cacheable responses
- Simple API responses
- Static content (use
static_assetsmodule instead)