Arc<Mutex>

Guides

Rust walkthroughs

ā† Back to all guides

What is the purpose of hyper::body::Incoming::into_data_stream for streaming HTTP body data?

into_data_stream converts an Incoming HTTP body into a stream of data frames, enabling asynchronous iteration over body chunks without buffering the entire request in memory. The method transforms the body into a Stream<Item = Result<Bytes, hyper::Error>>, allowing incremental processing of large or streaming request bodies while preserving the ability to handle backpressure and cancellation.

Basic Streaming Conversion

use hyper::body::Incoming;
use hyper::Request;
use futures::StreamExt;
 
async fn basic_streaming() {
    // Incoming body from an HTTP request
    let request: Request<Incoming> = /* ... */;
    let body = request.into_body();
    
    // Convert to stream
    let mut stream = body.into_data_stream();
    
    // Iterate over chunks as they arrive
    while let Some(chunk_result) = stream.next().await {
        match chunk_result {
            Ok(bytes) => {
                println!("Received {} bytes", bytes.len());
                // Process chunk immediately
            }
            Err(e) => {
                eprintln!("Stream error: {}", e);
                break;
            }
        }
    }
}

into_data_stream provides a Stream interface for incremental body consumption.

Streaming vs Buffering

use hyper::body::Incoming;
use http_body_util::BodyExt;
use futures::StreamExt;
 
async fn compare_approaches() {
    let body: Incoming = /* ... */;
    
    // Approach 1: Buffer entire body (memory intensive)
    let buffered = body.collect().await.unwrap().to_bytes();
    println!("Total size: {}", buffered.len());
    // Entire body held in memory at once
    
    // Approach 2: Stream chunks (memory efficient)
    let body: Incoming = /* ... */;
    let mut stream = body.into_data_stream();
    let mut total = 0;
    
    while let Some(chunk) = stream.next().await {
        let bytes = chunk.unwrap();
        total += bytes.len();
        // Process chunk, then release memory
    }
    println!("Total size: {}", total);
}

Streaming avoids buffering the entire body; chunks are processed and released incrementally.

Large File Upload Processing

use hyper::body::Incoming;
use futures::StreamExt;
use tokio::io::AsyncWriteExt;
 
async fn stream_upload_to_file(body: Incoming) -> std::io::Result<()> {
    let mut file = tokio::fs::File::create("upload.bin").await?;
    let mut stream = body.into_data_stream();
    
    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result.map_err(|e| {
            std::io::Error::new(std::io::ErrorKind::Other, e)
        })?;
        
        // Write chunk to file immediately
        file.write_all(&chunk).await?;
    }
    
    file.flush().await?;
    Ok(())
}
 
// This handles gigabyte-sized uploads without running out of memory

Streaming enables processing of arbitrarily large uploads within fixed memory bounds.

Real-time Data Processing

use hyper::body::Incoming;
use futures::StreamExt;
 
async fn process_json_stream(body: Incoming) {
    let mut stream = body.into_data_stream();
    let mut buffer = Vec::new();
    
    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result.unwrap();
        buffer.extend_from_slice(&chunk);
        
        // Process complete JSON objects as they arrive
        while let Some(pos) = buffer.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = buffer.drain(..=pos).collect();
            let json: serde_json::Value = serde_json::from_slice(&line).unwrap();
            process_json_object(json).await;
        }
    }
    
    // Process remaining data
    if !buffer.is_empty() {
        let json: serde_json::Value = serde_json::from_slice(&buffer).unwrap();
        process_json_object(json).await;
    }
}
 
async fn process_json_object(json: serde_json::Value) {
    // Handle each JSON object as it becomes available
}

Stream processing enables real-time handling of newline-delimited JSON or similar protocols.

Integration with Tower Services

use hyper::body::Incoming;
use futures::StreamExt;
use tower::Service;
 
async fn tower_integration(body: Incoming) {
    // The stream can be used with Tower middleware
    let mut stream = body.into_data_stream();
    
    // Apply streaming transformations
    let transformed = async_stream::stream! {
        while let Some(chunk_result) = stream.next().await {
            match chunk_result {
                Ok(bytes) => {
                    // Transform each chunk
                    let transformed = transform_chunk(bytes);
                    yield Ok(transformed);
                }
                Err(e) => yield Err(e),
            }
        }
    };
    
    // Use transformed stream
    futures::pin_mut!(transformed);
    while let Some(chunk) = transformed.next().await {
        // Handle transformed data
    }
}
 
fn transform_chunk(bytes: bytes::Bytes) -> bytes::Bytes {
    // Example: uppercase text data
    let upper: Vec<u8> = bytes.iter().map(|b| b.to_ascii_uppercase()).collect();
    bytes::Bytes::from(upper)
}

Streams integrate with async streaming utilities and Tower middleware.

Backpressure Handling

use hyper::body::Incoming;
use futures::StreamExt;
use tokio::sync::mpsc;
 
async fn backpressure_example(body: Incoming) {
    let (tx, mut rx) = mpsc::channel::<bytes::Bytes>(32);
    
    // Spawn task to consume body
    tokio::spawn(async move {
        let mut stream = body.into_data_stream();
        while let Some(chunk_result) = stream.next().await {
            if let Ok(chunk) = chunk_result {
                // Channel provides backpressure
                // If channel is full, this await pauses
                if tx.send(chunk).await.is_err() {
                    break; // Receiver dropped
                }
            }
        }
    });
    
    // Consumer with bounded processing
    while let Some(chunk) = rx.recv().await {
        process_chunk(chunk).await;
        // Processing rate controls backpressure
    }
}
 
async fn process_chunk(chunk: bytes::Bytes) {
    // Simulate slow processing
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
}

The stream respects backpressure through async/await; slow consumers naturally slow producers.

Error Handling Patterns

use hyper::body::Incoming;
use hyper::Error;
use futures::StreamExt;
 
async fn error_handling_patterns(body: Incoming) -> Result<(), Box<dyn std::error::Error>> {
    let mut stream = body.into_data_stream();
    let mut total_bytes = 0;
    
    while let Some(chunk_result) = stream.next().await {
        match chunk_result {
            Ok(bytes) => {
                total_bytes += bytes.len();
                // Process successfully received data
            }
            Err(e) => {
                // Stream errors from hyper
                // Common errors:
                // - Connection reset
                // - Timeout
                // - Invalid chunk encoding
                
                if e.is_timeout() {
                    return Err(format!("Stream timeout: {}", e).into());
                }
                if e.is_parse() {
                    return Err(format!("Parse error: {}", e).into());
                }
                return Err(format!("Stream error: {}", e).into());
            }
        }
    }
    
    println!("Successfully received {} bytes", total_bytes);
    Ok(())
}

The stream yields Result<Bytes, Error>, handling network errors gracefully.

Content-Length and Trailing Headers

use hyper::body::Incoming;
use futures::StreamExt;
 
async fn content_aware_processing(body: Incoming) {
    // Incoming preserves HTTP semantics
    // Content-Length headers inform the stream
    
    let mut stream = body.into_data_stream();
    let mut received = 0u64;
    
    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result.unwrap();
        received += chunk.len() as u64;
        
        // Track progress for clients that sent Content-Length
        println!("Received {} bytes", received);
    }
    
    // Stream naturally ends when body is complete
    // No need to check Content-Length manually
}

The stream respects HTTP semantics including Content-Length handling.

Comparison with Other Body Types

use hyper::body::Incoming;
use http_body_util::{BodyExt, Full, Empty};
use bytes::Bytes;
use futures::StreamExt;
 
async fn body_types_comparison() {
    // Incoming: Streaming body from network
    let incoming: Incoming = /* ... */;
    let stream1 = incoming.into_data_stream();
    // Data arrives as network packets
    
    // Full<Bytes>: Complete body in memory
    let full = Full::new(Bytes::from("hello"));
    let stream2 = full.into_data_stream();
    // Single chunk with all data
    
    // Empty: No body
    let empty = Empty::<Bytes>::new();
    let stream3 = empty.into_data_stream();
    // Immediately ends
    
    // All convert to same Stream type
    // But data arrival differs
}

Different body types convert to streams with different data arrival patterns.

Memory Efficiency Comparison

use hyper::body::Incoming;
use http_body_util::BodyExt;
use futures::StreamExt;
 
async fn memory_comparison() {
    // Scenario: 100MB upload
    
    // Approach 1: Buffer everything (BAD for large bodies)
    let body: Incoming = /* ... */;
    let all_bytes = body.collect().await.unwrap().to_bytes();
    // Memory: ~100MB held at once
    
    // Approach 2: Stream chunks (GOOD for large bodies)
    let body: Incoming = /* ... */;
    let mut stream = body.into_data_stream();
    let mut process_buf = vec![0u8; 8192];  // Fixed buffer
    
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.unwrap();
        // Memory: chunk.len() bytes at a time
        // Typical chunk: 8KB-64KB
        // Never exceeds typical chunk size
        
        // Process chunk immediately
        process_chunk(&chunk);
    }
}
 
fn process_chunk(chunk: &[u8]) {
    // Process without accumulating
}

Streaming maintains constant memory usage regardless of body size.

Chunked Transfer Encoding

use hyper::body::Incoming;
use futures::StreamExt;
 
async fn chunked_encoding(body: Incoming) {
    // HTTP/1.1 chunked transfer encoding
    // Incoming handles dechunking automatically
    
    let mut stream = body.into_data_stream();
    
    // Stream yields dechunked data
    // Each chunk_result contains raw data bytes
    // Chunk framing is stripped by hyper
    
    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result.unwrap();
        // chunk contains application data
        // No chunk size prefixes or \r\n delimiters
        println!("Data chunk: {} bytes", chunk.len());
    }
    
    // HTTP/2 and HTTP/3 use different framing
    // But into_data_stream abstracts this away
}

into_data_stream handles HTTP framing transparently across HTTP versions.

Integration with Axum

use axum::{
    extract::Body,
    http::Request,
};
use futures::StreamExt;
 
async fn axum_streaming(body: Body) {
    // Axum Body is a wrapper around Incoming
    let mut stream = body.into_data_stream();
    
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.unwrap();
        process(chunk).await;
    }
}
 
// Or use Axum's streaming extractors
use axum::extract::Multipart;
 
async fn axum_multipart(mut multipart: Multipart) {
    while let Some(field) = multipart.next_field().await.unwrap() {
        let name = field.name().unwrap().to_string();
        
        // Stream field data
        while let Some(chunk) = field.chunk().await.unwrap() {
            println!("Field {} received {} bytes", name, chunk.len());
        }
    }
}

Axum integrates with hyper's streaming model for efficient body handling.

Canceling Stream Consumption

use hyper::body::Incoming;
use futures::StreamExt;
 
async fn cancel_early(body: Incoming) -> Result<(), hyper::Error> {
    let mut stream = body.into_data_stream();
    
    // Read with a limit
    let max_bytes = 1024 * 1024;  // 1MB limit
    let mut received = 0;
    
    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result?;
        received += chunk.len();
        
        if received > max_bytes {
            // Stop consuming
            // Dropping the stream signals to the remote
            // that we're not interested in more data
            return Ok(());
        }
        
        process_chunk(chunk);
    }
    
    Ok(())
}

Dropping the stream cancels body consumption; the remote receives backpressure or reset signals.

Timeout on Stream

use hyper::body::Incoming;
use futures::StreamExt;
use tokio::time::{timeout, Duration};
 
async fn stream_with_timeout(body: Incoming) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
    let mut stream = body.into_data_stream();
    let mut all_data = Vec::new();
    
    loop {
        // Apply timeout to each chunk
        let chunk_result = timeout(Duration::from_secs(30), stream.next()).await;
        
        match chunk_result {
            Ok(Some(result)) => {
                all_data.extend_from_slice(&result?);
            }
            Ok(None) => break,  // Stream ended
            Err(_) => {
                // Timeout
                return Err("Stream timeout".into());
            }
        }
    }
    
    Ok(all_data)
}

Wrap stream operations with timeouts to prevent indefinite waiting.

Synthesis

Quick comparison:

Method Returns Use Case
collect() Buf (buffered) Small bodies, need all data
into_data_stream() Stream<Item = Result<Bytes, Error>> Large bodies, streaming
to_bytes() Bytes Complete body in memory

Decision guide:

use hyper::body::Incoming;
use http_body_util::BodyExt;
use futures::StreamExt;
 
async fn choose_approach(body: Incoming) {
    // Use into_data_stream when:
    // - Body size unknown or unbounded
    // - Memory is constrained
    // - Processing can be incremental
    // - Real-time processing needed
    
    let mut stream = body.into_data_stream();
    while let Some(chunk) = stream.next().await {
        process_incremental(chunk.unwrap());
    }
    
    // Use collect/to_bytes when:
    // - Body is small and bounded
    // - Need entire body for parsing
    // - Memory is not a concern
    
    let complete_body = body.collect().await.unwrap().to_bytes();
    let json: serde_json::Value = serde_json::from_slice(&complete_body).unwrap();
}

Key insight: into_data_stream transforms hyper's Incoming body into a standard Stream interface, enabling the Rust ecosystem's rich async streaming tools to process HTTP bodies incrementally. The key benefit is memory efficiency: instead of buffering an entire request body (which could be gigabytes for file uploads), you process chunks as they arrive and release them immediately. This is essential for production servers handling large uploads, streaming APIs, or real-time data processing. The stream naturally integrates with backpressureā€”if your processing slows down, the TCP connection slows down, preventing memory exhaustion. Errors are delivered as Result items in the stream, allowing you to handle partial reads and connection failures gracefully. The method abstracts away HTTP-version-specific details (chunked encoding for HTTP/1.1, DATA frames for HTTP/2, streams for HTTP/3), presenting a unified streaming interface regardless of protocol version.