Streaming responses

use axum::{
    body::{Body, Bytes},
    extract::Request,
    http::StatusCode,
    routing::post,
    Router,
};
use futures_util::Stream;

async fn handler(request: Request) -> Body {
    // Stream the request body back as response
    request.into_body()
}

let app = Router::new().route("/echo", post(handler));

use axum::{
    body::{Body, Bytes},
    extract::Request,
};
use futures_util::{Stream, TryStreamExt};
use std::io;

async fn stream_handler(request: Request) -> Result<(), (StatusCode, String)> {
    // Convert body to a stream of Bytes
    let stream = request.into_body().into_data_stream();
    
    // Process the stream
    stream_to_file("output.txt", stream).await
}

async fn stream_to_file<S, E>(path: &str, stream: S) -> Result<(), (StatusCode, String)>
where
    S: Stream<Item = Result<Bytes, E>>,
    E: Into<BoxError>,
{
    // Convert stream to AsyncRead and write to file
    let body_with_io_error = stream.map_err(io::Error::other);
    let body_reader = StreamReader::new(body_with_io_error);
    
    let mut file = File::create(path).await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    
    tokio::io::copy(&mut body_reader, &mut file).await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    
    Ok(())
}

use axum::body::{to_bytes, Body};

#[tokio::main]
async fn main() -> Result<(), axum_core::Error> {
    let body = Body::from(vec![1, 2, 3]);
    
    // Limit to 1MB (use usize::MAX for no limit)
    let bytes = to_bytes(body, 1024 * 1024).await?;
    
    assert_eq!(&bytes[..], &[1, 2, 3]);
    Ok(())
}

use axum::body::{to_bytes, Body};
use http_body_util::LengthLimitError;

#[tokio::main]
async fn main() {
    let body = Body::from(vec![1, 2, 3]);
    
    match to_bytes(body, 1).await {
        Ok(_) => panic!("should have hit the limit"),
        Err(err) => {
            let source = std::error::Error::source(&err).unwrap();
            assert!(source.is::<LengthLimitError>());
        }
    }
}

use axum::body::Body;

async fn handler() -> Body {
    Body::from("Hello, World!")
}

// Or from bytes
async fn bytes_handler() -> Body {
    Body::from(vec![1, 2, 3, 4])
}

use axum::body::{Body, Bytes};
use futures_util::stream;

async fn stream_response() -> Body {
    let data = vec![
        Bytes::from("chunk 1\n"),
        Bytes::from("chunk 2\n"),
        Bytes::from("chunk 3\n"),
    ];
    
    let stream = stream::iter(data.into_iter().map(Ok::<_, std::io::Error>));
    Body::from_stream(stream)
}

use axum::{
    body::Body,
    extract::Path,
    http::{header, StatusCode},
    response::IntoResponse,
};
use tokio::fs::File;
use tokio_util::io::ReaderStream;

async fn download_file(Path(filename): Path<String>) -> impl IntoResponse {
    let file = match File::open(&filename).await {
        Ok(file) => file,
        Err(_) => return Err((StatusCode::NOT_FOUND, "File not found")),
    };
    
    let stream = ReaderStream::new(file);
    let body = Body::from_stream(stream);
    
    Ok((
        [
            (header::CONTENT_TYPE, "application/octet-stream"),
            (header::CONTENT_DISPOSITION, &format!("attachment; filename=\"{}\"", filename)),
        ],
        body,
    ))
}

use axum::{
    body::Body,
    extract::Request,
    response::Response,
};
use reqwest::Client;

async fn proxy(request: Request) -> Result<Response, StatusCode> {
    let client = Client::new();
    
    // Forward the request body as a stream
    let response = client
        .post("http://backend-service/api")
        .body(request.into_body())
        .send()
        .await
        .map_err(|_| StatusCode::BAD_GATEWAY)?;
    
    // Stream the response back
    let mut builder = Response::builder().status(response.status());
    
    for (key, value) in response.headers() {
        builder = builder.header(key, value);
    }
    
    Ok(builder.body(Body::from_stream(response.bytes_stream())).unwrap())
}

use axum::{
    body::Bytes,
    extract::Multipart,
};
use futures_util::Stream;

async fn upload_handler(mut multipart: Multipart) -> Result<(), StatusCode> {
    while let Ok(Some(field)) = multipart.next_field().await {
        let file_name = field.file_name()
            .ok_or(StatusCode::BAD_REQUEST)?
            .to_owned();
        
        // Process field as a stream
        process_field_stream(file_name, field).await?;
    }
    
    Ok(())
}

async fn process_field_stream<S, E>(
    name: String,
    stream: S,
) -> Result<(), StatusCode>
where
    S: Stream<Item = Result<Bytes, E>>,
{
    // Process the stream...
    Ok(())
}

use axum::body::HttpBody;

// Re-exported as:
use http_body::Body as HttpBody;

use axum::body::Bytes;

// Re-exported as:
use bytes::Bytes;

use axum::body::{Body, BodyDataStream};

let body = Body::from("data");
let stream: BodyDataStream = body.into_data_stream();

use axum::body::to_bytes;

// Always set appropriate limits when consuming bodies
let bytes = to_bytes(body, 10 * 1024 * 1024).await?; // 10MB limit