diff --git a/examples/stream-to-file/Cargo.toml b/examples/stream-to-file/Cargo.toml new file mode 100644 index 00000000..b203f836 --- /dev/null +++ b/examples/stream-to-file/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "example-stream-to-file" +version = "0.1.0" +edition = "2018" +publish = false + +[dependencies] +axum = { path = "../../axum", features = ["multipart"] } +futures = "0.3" +tokio = { version = "1.0", features = ["full"] } +tokio-util = { version = "0.7", features = ["io"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/examples/stream-to-file/src/main.rs b/examples/stream-to-file/src/main.rs new file mode 100644 index 00000000..6a8bad95 --- /dev/null +++ b/examples/stream-to-file/src/main.rs @@ -0,0 +1,117 @@ +//! Run with +//! +//! ```not_rust +//! cargo run -p example-stream-to-file +//! ``` + +use axum::{ + body::Bytes, + extract::{BodyStream, Multipart, Path}, + http::StatusCode, + response::{Html, Redirect}, + routing::{get, post}, + BoxError, Router, +}; +use futures::{Stream, TryStreamExt}; +use std::{io, net::SocketAddr}; +use tokio::{fs::File, io::BufWriter}; +use tokio_util::io::StreamReader; +use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + +#[tokio::main] +async fn main() { + tracing_subscriber::registry() + .with(tracing_subscriber::EnvFilter::new( + std::env::var("RUST_LOG").unwrap_or_else(|_| "example_stream_to_file=debug".into()), + )) + .with(tracing_subscriber::fmt::layer()) + .init(); + + let app = Router::new() + .route("/", get(show_form).post(accept_form)) + .route("/file/:file_name", post(save_request_body)); + + let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); + tracing::debug!("listening on {}", addr); + axum::Server::bind(&addr) + .serve(app.into_make_service()) + .await + .unwrap(); +} + +// Handler that streams the request body to a file. +// +// POST'ing to `/file/foo.txt` will create a file called `foo.txt`. +async fn save_request_body( + Path(file_name): Path, + body: BodyStream, +) -> Result<(), (StatusCode, String)> { + stream_to_file(&file_name, body) + .await + .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string())) +} + +// Handler that returns HTML for a multipart form. +async fn show_form() -> Html<&'static str> { + Html( + r#" + + + + Upload something! + + +
+
+ +
+ +
+ +
+
+ + + "#, + ) +} + +// Handler that accepts a multipart form upload and streams each field to a file. +async fn accept_form(mut multipart: Multipart) -> Result { + while let Some(field) = multipart.next_field().await.unwrap() { + let file_name = if let Some(file_name) = field.file_name() { + file_name.to_owned() + } else { + continue; + }; + + stream_to_file(&file_name, field) + .await + .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?; + } + + Ok(Redirect::to("/")) +} + +// Save a `Stream` to a file +async fn stream_to_file(path: &str, stream: S) -> Result<(), io::Error> +where + S: Stream>, + E: Into, +{ + // Convert the stream into an `AsyncRead`. + let body_with_io_error = stream.map_err(|err| io::Error::new(io::ErrorKind::Other, err)); + let body_reader = StreamReader::new(body_with_io_error); + futures::pin_mut!(body_reader); + + // Create the file. `File` implements `AsyncWrite`. + let mut file = BufWriter::new(File::create(path).await?); + + // Copy the body into the file. + tokio::io::copy(&mut body_reader, &mut file).await?; + + Ok(()) +}