OpenSource/axum
OpenSource/axum
1
0
Fork 0
mirror of https://github.com/tokio-rs/axum.git synced 2024-11-21 06:36:32 +01:00
Code Issues Projects Releases Packages Wiki Activity

Update to latest versions of hyper and http-body (#1882)

Browse source 
Co-authored-by: Michael Scofield <mscofield0@tutanota.com>
Co-authored-by: Jonas Platte <jplatte+git@posteo.de>
This commit is contained in:
David Pedersen 2023-11-23 12:03:03 +01:00 committed by GitHub
parent 2f4720907a
commit 43b14a5f02
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
93 changed files with 1463 additions and 1448 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/CI.yml vendored
View file

@ -2,7 +2,7 @@ name: CI
env:
CARGO_TERM_COLOR: always
MSRV: '1.65'
MSRV: '1.66'
on:
push:

4
Cargo.toml
View file

@ -5,3 +5,7 @@ default-members = ["axum", "axum-*"]
# Example has been deleted, but README.md remains
exclude = ["examples/async-graphql"]
resolver = "2"
[patch.crates-io]
# for http 1.0. PR to update is merged but not published
headers = { git = "https://github.com/hyperium/headers", rev = "4400aa90c47a7" }

6
_typos.toml Normal file
View file

@ -0,0 +1,6 @@
[files]
extend-exclude = ["Cargo.toml"]
[default.extend-identifiers]
DefaultOnFailedUpdgrade = "DefaultOnFailedUpdgrade"
OnFailedUpdgrade = "OnFailedUpdgrade"

11
axum-core/Cargo.toml
View file

@ -21,8 +21,9 @@ __private_docs = ["dep:tower-http"]
async-trait = "0.1.67"
bytes = "1.0"
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
http = "0.2.7"
http-body = "0.4.5"
http = "1.0.0"
http-body = "1.0.0"
http-body-util = "0.1.0"
mime = "0.3.16"
pin-project-lite = "0.2.7"
sync_wrapper = "0.1.1"
@ -30,7 +31,7 @@ tower-layer = "0.3"
tower-service = "0.3"
# optional dependencies
tower-http = { version = "0.4", optional = true, features = ["limit"] }
tower-http = { version = "0.5.0", optional = true, features = ["limit"] }
tracing = { version = "0.1.37", default-features = false, optional = true }
[build-dependencies]
@ -40,9 +41,9 @@ rustversion = "1.0.9"
axum = { path = "../axum", version = "0.6.0" }
axum-extra = { path = "../axum-extra", features = ["typed-header"] }
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
hyper = "0.14.24"
hyper = "1.0.0"
tokio = { version = "1.25.0", features = ["macros"] }
tower-http = { version = "0.4", features = ["limit"] }
tower-http = { version = "0.5.0", features = ["limit"] }
[package.metadata.cargo-public-api-crates]
allowed = [

146
axum-core/src/body.rs
View file

@ -2,17 +2,16 @@
use crate::{BoxError, Error};
use bytes::Bytes;
use bytes::{Buf, BufMut};
use futures_util::stream::Stream;
use futures_util::TryStream;
use http::HeaderMap;
use http_body::Body as _;
use http_body::{Body as _, Frame};
use http_body_util::BodyExt;
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
use sync_wrapper::SyncWrapper;
type BoxBody = http_body::combinators::UnsyncBoxBody<Bytes, Error>;
type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, Error>;
fn boxed<B>(body: B) -> BoxBody
where
@ -35,58 +34,6 @@ where
}
}
// copied from hyper under the following license:
// Copyright (c) 2014-2021 Sean McArthur
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
pub(crate) async fn to_bytes<T>(body: T) -> Result<Bytes, T::Error>
where
T: http_body::Body,
{
futures_util::pin_mut!(body);
// If there's only 1 chunk, we can just return Buf::to_bytes()
let mut first = if let Some(buf) = body.data().await {
buf?
} else {
return Ok(Bytes::new());
};
let second = if let Some(buf) = body.data().await {
buf?
} else {
return Ok(first.copy_to_bytes(first.remaining()));
};
// With more than 1 buf, we gotta flatten into a Vec first.
let cap = first.remaining() + second.remaining() + body.size_hint().lower() as usize;
let mut vec = Vec::with_capacity(cap);
vec.put(first);
vec.put(second);
while let Some(buf) = body.data().await {
vec.put(buf?);
}
Ok(vec.into())
}
/// The body type used in axum requests and responses.
#[derive(Debug)]
pub struct Body(BoxBody);
@ -103,7 +50,7 @@ impl Body {
/// Create an empty body.
pub fn empty() -> Self {
Self::new(http_body::Empty::new())
Self::new(http_body_util::Empty::new())
}
/// Create a new `Body` from a [`Stream`].
@ -119,6 +66,16 @@ impl Body {
stream: SyncWrapper::new(stream),
})
}
/// Convert the body into a [`Stream`] of data frames.
///
/// Non-data frames (such as trailers) will be discarded. Use [`http_body_util::BodyStream`] if
/// you need a [`Stream`] of all frame types.
///
/// [`http_body_util::BodyStream`]: https://docs.rs/http-body-util/latest/http_body_util/struct.BodyStream.html
pub fn into_data_stream(self) -> BodyDataStream {
BodyDataStream { inner: self }
}
}
impl Default for Body {
@ -131,7 +88,7 @@ macro_rules! body_from_impl {
($ty:ty) => {
impl From<$ty> for Body {
fn from(buf: $ty) -> Self {
Self::new(http_body::Full::from(buf))
Self::new(http_body_util::Full::from(buf))
}
}
};
@ -152,19 +109,11 @@ impl http_body::Body for Body {
type Error = Error;
#[inline]
fn poll_data(
fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> std::task::Poll<Option<Result<Self::Data, Self::Error>>> {
Pin::new(&mut self.0).poll_data(cx)
}
#[inline]
fn poll_trailers(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> std::task::Poll<Result<Option<HeaderMap>, Self::Error>> {
Pin::new(&mut self.0).poll_trailers(cx)
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
Pin::new(&mut self.0).poll_frame(cx)
}
#[inline]
@ -178,12 +127,51 @@ impl http_body::Body for Body {
}
}
impl Stream for Body {
/// A stream of data frames.
///
/// Created with [`Body::into_data_stream`].
#[derive(Debug)]
pub struct BodyDataStream {
inner: Body,
}
impl Stream for BodyDataStream {
type Item = Result<Bytes, Error>;
#[inline]
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.poll_data(cx)
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
match futures_util::ready!(Pin::new(&mut self.inner).poll_frame(cx)?) {
Some(frame) => match frame.into_data() {
Ok(data) => return Poll::Ready(Some(Ok(data))),
Err(_frame) => {}
},
None => return Poll::Ready(None),
}
}
}
}
impl http_body::Body for BodyDataStream {
type Data = Bytes;
type Error = Error;
#[inline]
fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
Pin::new(&mut self.inner).poll_frame(cx)
}
#[inline]
fn is_end_stream(&self) -> bool {
self.inner.is_end_stream()
}
#[inline]
fn size_hint(&self) -> http_body::SizeHint {
self.inner.size_hint()
}
}
@ -203,25 +191,17 @@ where
type Data = Bytes;
type Error = Error;
fn poll_data(
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, Self::Error>>> {
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let stream = self.project().stream.get_pin_mut();
match futures_util::ready!(stream.try_poll_next(cx)) {
Some(Ok(chunk)) => Poll::Ready(Some(Ok(chunk.into()))),
Some(Ok(chunk)) => Poll::Ready(Some(Ok(Frame::data(chunk.into())))),
Some(Err(err)) => Poll::Ready(Some(Err(Error::new(err)))),
None => Poll::Ready(None),
}
}
#[inline]
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
Poll::Ready(Ok(None))
}
}
#[test]

23
axum-core/src/ext_traits/request.rs
View file

@ -1,7 +1,6 @@
use crate::body::Body;
use crate::extract::{DefaultBodyLimitKind, FromRequest, FromRequestParts, Request};
use futures_util::future::BoxFuture;
use http_body::Limited;
mod sealed {
pub trait Sealed {}
@ -258,13 +257,13 @@ pub trait RequestExt: sealed::Sealed + Sized {
/// Apply the [default body limit](crate::extract::DefaultBodyLimit).
///
/// If it is disabled, return the request as-is in `Err`.
fn with_limited_body(self) -> Result<Request<Limited<Body>>, Request>;
/// If it is disabled, the request is returned as-is.
fn with_limited_body(self) -> Request;
/// Consumes the request, returning the body wrapped in [`Limited`] if a
/// Consumes the request, returning the body wrapped in [`http_body_util::Limited`] if a
/// [default limit](crate::extract::DefaultBodyLimit) is in place, or not wrapped if the
/// default limit is disabled.
fn into_limited_body(self) -> Result<Limited<Body>, Body>;
fn into_limited_body(self) -> Body;
}
impl RequestExt for Request {
@ -320,24 +319,22 @@ impl RequestExt for Request {
})
}
fn with_limited_body(self) -> Result<Request<Limited<Body>>, Request> {
fn with_limited_body(self) -> Request {
// update docs in `axum-core/src/extract/default_body_limit.rs` and
// `axum/src/docs/extract.md` if this changes
const DEFAULT_LIMIT: usize = 2_097_152; // 2 mb
match self.extensions().get::<DefaultBodyLimitKind>().copied() {
Some(DefaultBodyLimitKind::Disable) => Err(self),
Some(DefaultBodyLimitKind::Disable) => self,
Some(DefaultBodyLimitKind::Limit(limit)) => {
Ok(self.map(|b| http_body::Limited::new(b, limit)))
self.map(|b| Body::new(http_body_util::Limited::new(b, limit)))
}
None => Ok(self.map(|b| http_body::Limited::new(b, DEFAULT_LIMIT))),
None => self.map(|b| Body::new(http_body_util::Limited::new(b, DEFAULT_LIMIT))),
}
}
fn into_limited_body(self) -> Result<Limited<Body>, Body> {
self.with_limited_body()
.map(Request::into_body)
.map_err(Request::into_body)
fn into_limited_body(self) -> Body {
self.with_limited_body().into_body()
}
}

34
axum-core/src/extract/default_body_limit.rs
View file

@ -8,7 +8,7 @@ use tower_layer::Layer;
///
/// This middleware provides ways to configure that.
///
/// Note that if an extractor consumes the body directly with [`Body::data`], or similar, the
/// Note that if an extractor consumes the body directly with [`Body::poll_frame`], or similar, the
/// default limit is _not_ applied.
///
/// # Difference between `DefaultBodyLimit` and [`RequestBodyLimit`]
@ -22,8 +22,7 @@ use tower_layer::Layer;
/// [`RequestBodyLimit`] is applied globally to all requests, regardless of which extractors are
/// used or how the body is consumed.
///
/// `DefaultBodyLimit` is also easier to integrate into an existing setup since it doesn't change
/// the request body type:
/// # Example
///
/// ```
/// use axum::{
@ -34,31 +33,12 @@ use tower_layer::Layer;
/// };
///
/// let app = Router::new()
/// .route(
/// "/",
/// // even with `DefaultBodyLimit` the request body is still just `Body`
/// post(|request: Request| async {}),
/// )
/// .route("/", post(|request: Request| async {}))
/// // change the default limit
/// .layer(DefaultBodyLimit::max(1024));
/// # let _: Router = app;
/// ```
///
/// ```
/// use axum::{Router, routing::post, body::Body, extract::Request};
/// use tower_http::limit::RequestBodyLimitLayer;
/// use http_body::Limited;
///
/// let app = Router::new()
/// .route(
/// "/",
/// // `RequestBodyLimitLayer` changes the request body type to `Limited<Body>`
/// // extracting a different body type wont work
/// post(|request: Request| async {}),
/// )
/// .layer(RequestBodyLimitLayer::new(1024));
/// # let _: Router = app;
/// ```
///
/// In general using `DefaultBodyLimit` is recommended but if you need to use third party
/// extractors and want to sure a limit is also applied there then [`RequestBodyLimit`] should be
/// used.
@ -84,7 +64,7 @@ use tower_layer::Layer;
/// # let _: Router = app;
/// ```
///
/// [`Body::data`]: http_body::Body::data
/// [`Body::poll_frame`]: http_body::Body::poll_frame
/// [`Bytes`]: bytes::Bytes
/// [`Json`]: https://docs.rs/axum/0.6.0/axum/struct.Json.html
/// [`Form`]: https://docs.rs/axum/0.6.0/axum/struct.Form.html
@ -123,7 +103,7 @@ impl DefaultBodyLimit {
/// extract::DefaultBodyLimit,
/// };
/// use tower_http::limit::RequestBodyLimitLayer;
/// use http_body::Limited;
/// use http_body_util::Limited;
///
/// let app: Router<()> = Router::new()
/// .route("/", get(|body: Bytes| async {}))
@ -158,7 +138,7 @@ impl DefaultBodyLimit {
/// extract::DefaultBodyLimit,
/// };
/// use tower_http::limit::RequestBodyLimitLayer;
/// use http_body::Limited;
/// use http_body_util::Limited;
///
/// let app: Router<()> = Router::new()
/// .route("/", get(|body: Bytes| async {}))

11
axum-core/src/extract/rejection.rs
View file

@ -19,11 +19,18 @@ impl FailedToBufferBody {
where
E: Into<BoxError>,
{
// two layers of boxes here because `with_limited_body`
// wraps the `http_body_util::Limited` in a `axum_core::Body`
// which also wraps the error type
let box_error = match err.into().downcast::<Error>() {
Ok(err) => err.into_inner(),
Err(err) => err,
};
match box_error.downcast::<http_body::LengthLimitError>() {
let box_error = match box_error.downcast::<Error>() {
Ok(err) => err.into_inner(),
Err(err) => err,
};
match box_error.downcast::<http_body_util::LengthLimitError>() {
Ok(err) => Self::LengthLimitError(LengthLimitError::from_err(err)),
Err(err) => Self::UnknownBodyError(UnknownBodyError::from_err(err)),
}
@ -36,7 +43,7 @@ define_rejection! {
/// Encountered some other error when buffering the body.
///
/// This can _only_ happen when you're using [`tower_http::limit::RequestBodyLimitLayer`] or
/// otherwise wrapping request bodies in [`http_body::Limited`].
/// otherwise wrapping request bodies in [`http_body_util::Limited`].
pub struct LengthLimitError(Error);
}

15
axum-core/src/extract/request_parts.rs
View file

@ -3,6 +3,7 @@ use crate::{body::Body, RequestExt};
use async_trait::async_trait;
use bytes::Bytes;
use http::{request::Parts, HeaderMap, Method, Uri, Version};
use http_body_util::BodyExt;
use std::convert::Infallible;
#[async_trait]
@ -78,14 +79,12 @@ where
type Rejection = BytesRejection;
async fn from_request(req: Request, _: &S) -> Result<Self, Self::Rejection> {
let bytes = match req.into_limited_body() {
Ok(limited_body) => crate::body::to_bytes(limited_body)
.await
.map_err(FailedToBufferBody::from_err)?,
Err(unlimited_body) => crate::body::to_bytes(unlimited_body)
.await
.map_err(FailedToBufferBody::from_err)?,
};
let bytes = req
.into_limited_body()
.collect()
.await
.map_err(FailedToBufferBody::from_err)?
.to_bytes();
Ok(bytes)
}

36
axum-core/src/response/into_response.rs
View file

@ -5,7 +5,7 @@ use http::{
header::{self, HeaderMap, HeaderName, HeaderValue},
Extensions, StatusCode,
};
use http_body::SizeHint;
use http_body::{Frame, SizeHint};
use std::{
borrow::Cow,
convert::Infallible,
@ -58,9 +58,7 @@ use std::{
/// async fn handler() -> Result<(), MyError> {
/// Err(MyError::SomethingWentWrong)
/// }
/// # async {
/// # hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
/// # };
/// # let _: Router = app;
/// ```
///
/// Or if you have a custom body type you'll also need to implement
@ -76,6 +74,7 @@ use std::{
/// };
/// use http::HeaderMap;
/// use bytes::Bytes;
/// use http_body::Frame;
/// use std::{
/// convert::Infallible,
/// task::{Poll, Context},
@ -90,18 +89,10 @@ use std::{
/// type Data = Bytes;
/// type Error = Infallible;
///
/// fn poll_data(
/// fn poll_frame(
/// self: Pin<&mut Self>,
/// cx: &mut Context<'_>
/// ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
/// # unimplemented!()
/// // ...
/// }
///
/// fn poll_trailers(
/// self: Pin<&mut Self>,
/// cx: &mut Context<'_>
/// ) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
/// cx: &mut Context<'_>,
/// ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
/// # unimplemented!()
/// // ...
/// }
@ -256,30 +247,23 @@ where
type Data = Bytes;
type Error = Infallible;
fn poll_data(
fn poll_frame(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, Self::Error>>> {
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
if let Some(mut buf) = self.first.take() {
let bytes = buf.copy_to_bytes(buf.remaining());
return Poll::Ready(Some(Ok(bytes)));
return Poll::Ready(Some(Ok(Frame::data(bytes))));
}
if let Some(mut buf) = self.second.take() {
let bytes = buf.copy_to_bytes(buf.remaining());
return Poll::Ready(Some(Ok(bytes)));
return Poll::Ready(Some(Ok(Frame::data(bytes))));
}
Poll::Ready(None)
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
Poll::Ready(Ok(None))
}
fn is_end_stream(&self) -> bool {
self.first.is_none() && self.second.is_none()
}

13
axum-extra/Cargo.toml
View file

@ -2,7 +2,7 @@
categories = ["asynchronous", "network-programming", "web-programming"]
description = "Extra utilities for axum"
edition = "2021"
rust-version = "1.65"
rust-version = "1.66"
homepage = "https://github.com/tokio-rs/axum"
keywords = ["http", "web", "framework"]
license = "MIT"
@ -40,8 +40,9 @@ axum = { path = "../axum", version = "0.6.13", default-features = false }
axum-core = { path = "../axum-core", version = "0.3.4" }
bytes = "1.1.0"
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
http = "0.2"
http-body = "0.4.4"
http = "1.0.0"
http-body = "1.0.0"
http-body-util = "0.1.0"
mime = "0.3"
pin-project-lite = "0.2"
serde = "1.0"
@ -65,15 +66,13 @@ tokio-util = { version = "0.7", optional = true }
[dev-dependencies]
axum = { path = "../axum", version = "0.6.0" }
futures = "0.3"
http-body = "0.4.4"
hyper = "0.14"
hyper = "1.0.0"
reqwest = { version = "0.11", default-features = false, features = ["json", "stream", "multipart"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0.71"
tokio = { version = "1.14", features = ["full"] }
tower = { version = "0.4", features = ["util"] }
tower-http = { version = "0.4", features = ["map-response-body", "timeout"] }
tower-http = { version = "0.5.0", features = ["map-response-body", "timeout"] }
[package.metadata.docs.rs]
all-features = true

2
axum-extra/README.md
View file

@ -14,7 +14,7 @@ This crate uses `#![forbid(unsafe_code)]` to ensure everything is implemented in
## Minimum supported Rust version
axum-extra's MSRV is 1.65.
axum-extra's MSRV is 1.66.
## Getting Help

21
axum-extra/src/body/async_read_body.rs
View file

@ -1,6 +1,5 @@
use axum::{
body::{Body, Bytes, HttpBody},
http::HeaderMap,
response::{IntoResponse, Response},
Error,
};
@ -69,18 +68,22 @@ impl HttpBody for AsyncReadBody {
type Data = Bytes;
type Error = Error;
fn poll_data(
#[inline]
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, Self::Error>>> {
self.project().body.poll_data(cx)
) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
self.project().body.poll_frame(cx)
}
fn poll_trailers(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
self.project().body.poll_trailers(cx)
#[inline]
fn is_end_stream(&self) -> bool {
self.body.is_end_stream()
}
#[inline]
fn size_hint(&self) -> http_body::SizeHint {
self.body.size_hint()
}
}

3
axum-extra/src/extract/cookie/mod.rs
View file

@ -234,6 +234,7 @@ fn set_cookies(jar: cookie::CookieJar, headers: &mut HeaderMap) {
mod tests {
use super::*;
use axum::{body::Body, extract::FromRef, http::Request, routing::get, Router};
use http_body_util::BodyExt;
use tower::ServiceExt;
macro_rules! cookie_test {
@ -378,7 +379,7 @@ mod tests {
B: axum::body::HttpBody,
B::Error: std::fmt::Debug,
{
let bytes = hyper::body::to_bytes(body).await.unwrap();
let bytes = body.collect().await.unwrap().to_bytes();
String::from_utf8(bytes.to_vec()).unwrap()
}
}

31
axum-extra/src/extract/multipart.rs
View file

@ -12,7 +12,7 @@ use axum::{
use futures_util::stream::Stream;
use http::{
header::{HeaderMap, CONTENT_TYPE},
Request, StatusCode,
HeaderName, HeaderValue, Request, StatusCode,
};
use std::{
error::Error,
@ -99,11 +99,8 @@ where
async fn from_request(req: Request<Body>, _state: &S) -> Result<Self, Self::Rejection> {
let boundary = parse_boundary(req.headers()).ok_or(InvalidBoundary)?;
let stream = match req.with_limited_body() {
Ok(limited) => Body::new(limited),
Err(unlimited) => unlimited.into_body(),
};
let multipart = multer::Multipart::new(stream, boundary);
let stream = req.with_limited_body().into_body();
let multipart = multer::Multipart::new(stream.into_data_stream(), boundary);
Ok(Self { inner: multipart })
}
}
@ -118,7 +115,22 @@ impl Multipart {
.map_err(MultipartError::from_multer)?;
if let Some(field) = field {
Ok(Some(Field { inner: field }))
// multer still uses http 0.2 which means we cannot directly expose
// `multer::Field::headers`. Instead we have to eagerly convert the headers into http
// 1.0
//
// When the next major version of multer is published we can remove this.
let mut headers = HeaderMap::with_capacity(field.headers().len());
headers.extend(field.headers().into_iter().map(|(name, value)| {
let name = HeaderName::from_bytes(name.as_ref()).unwrap();
let value = HeaderValue::from_bytes(value.as_ref()).unwrap();
(name, value)
}));
Ok(Some(Field {
inner: field,
headers,
}))
} else {
Ok(None)
}
@ -137,6 +149,7 @@ impl Multipart {
#[derive(Debug)]
pub struct Field {
inner: multer::Field<'static>,
headers: HeaderMap,
}
impl Stream for Field {
@ -171,7 +184,7 @@ impl Field {
/// Get a map of headers as [`HeaderMap`].
pub fn headers(&self) -> &HeaderMap {
self.inner.headers()
&self.headers
}
/// Get the full data of the field as [`Bytes`].
@ -283,7 +296,7 @@ fn status_code_from_multer_error(err: &multer::Error) -> StatusCode {
if err
.downcast_ref::<axum::Error>()
.and_then(|err| err.source())
.and_then(|err| err.downcast_ref::<http_body::LengthLimitError>())
.and_then(|err| err.downcast_ref::<http_body_util::LengthLimitError>())
.is_some()
{
return StatusCode::PAYLOAD_TOO_LARGE;

4
axum-extra/src/json_lines.rs
View file

@ -111,8 +111,8 @@ where
// `Stream::lines` isn't a thing so we have to convert it into an `AsyncRead`
// so we can call `AsyncRead::lines` and then convert it back to a `Stream`
let body = req.into_body();
let stream = TryStreamExt::map_err(body, |err| io::Error::new(io::ErrorKind::Other, err));
let stream = body.into_data_stream();
let stream = stream.map_err(|err| io::Error::new(io::ErrorKind::Other, err));
let read = StreamReader::new(stream);
let lines_stream = LinesStream::new(read.lines());

3
axum-extra/src/routing/resource.rs
View file

@ -151,6 +151,7 @@ mod tests {
use super::*;
use axum::{body::Body, extract::Path, http::Method, Router};
use http::Request;
use http_body_util::BodyExt;
use tower::ServiceExt;
#[tokio::test]
@ -216,7 +217,7 @@ mod tests {
)
.await
.unwrap();
let bytes = hyper::body::to_bytes(res).await.unwrap();
let bytes = res.collect().await.unwrap().to_bytes();
String::from_utf8(bytes.to_vec()).unwrap()
}
}

2
axum-macros/Cargo.toml
View file

@ -2,7 +2,7 @@
categories = ["asynchronous", "network-programming", "web-programming"]
description = "Macros for axum"
edition = "2021"
rust-version = "1.65"
rust-version = "1.66"
homepage = "https://github.com/tokio-rs/axum"
keywords = ["axum"]
license = "MIT"

2
axum-macros/README.md
View file

@ -14,7 +14,7 @@ This crate uses `#![forbid(unsafe_code)]` to ensure everything is implemented in
## Minimum supported Rust version
axum-macros's MSRV is 1.65.
axum-macros's MSRV is 1.66.
## Getting Help

1
axum-macros/tests/from_request/pass/override_rejection.rs
View file

@ -4,6 +4,7 @@ use axum::{
http::StatusCode,
response::{IntoResponse, Response},
routing::get,
body::Body,
Extension, Router,
};

5
axum/CHANGELOG.md
View file

@ -61,11 +61,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **fixed:** Fix `.source()` of composite rejections ([#2030])
- **fixed:** Allow unreachable code in `#[debug_handler]` ([#2014])
- **change:** Update tokio-tungstenite to 0.19 ([#2021])
- **change:** axum's MSRV is now 1.65 ([#2021])
- **change:** axum's MSRV is now 1.66 ([#1882])
- **added:** Implement `Handler` for `T: IntoResponse` ([#2140])
- **added:** Implement `IntoResponse` for `(R,) where R: IntoResponse` ([#2143])
- **changed:** For SSE, add space between field and value for compatibility ([#2149])
- **added:** Add `NestedPath` extractor ([#1924])
- **breaking:** `impl<T> IntoResponse(Parts) for Extension<T>` now requires
`T: Clone`, as that is required by the http crate ([#1882])
- **added:** Add `axum::Json::from_bytes` ([#2244])
[#1664]: https://github.com/tokio-rs/axum/pull/1664
@ -75,6 +77,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
[#1835]: https://github.com/tokio-rs/axum/pull/1835
[#1850]: https://github.com/tokio-rs/axum/pull/1850
[#1868]: https://github.com/tokio-rs/axum/pull/1868
[#1882]: https://github.com/tokio-rs/axum/pull/1882
[#1924]: https://github.com/tokio-rs/axum/pull/1924
[#1956]: https://github.com/tokio-rs/axum/pull/1956
[#1972]: https://github.com/tokio-rs/axum/pull/1972

21
axum/Cargo.toml
View file

@ -4,7 +4,7 @@ version = "0.6.16"
categories = ["asynchronous", "network-programming", "web-programming::http-server"]
description = "Web framework that focuses on ergonomics and modularity"
edition = "2021"
rust-version = "1.65"
rust-version = "1.66"
homepage = "https://github.com/tokio-rs/axum"
keywords = ["http", "web", "framework"]
license = "MIT"
@ -22,7 +22,7 @@ matched-path = []
multipart = ["dep:multer"]
original-uri = []
query = ["dep:serde_urlencoded"]
tokio = ["dep:tokio", "dep:hyper-util", "hyper/server", "hyper/tcp", "hyper/runtime", "tower/make"]
tokio = ["dep:hyper-util", "dep:tokio", "tokio/net", "tokio/rt", "tower/make"]
tower-log = ["tower/log"]
tracing = ["dep:tracing", "axum-core/tracing"]
ws = ["tokio", "dep:tokio-tungstenite", "dep:sha1", "dep:base64"]
@ -35,9 +35,10 @@ async-trait = "0.1.67"
axum-core = { path = "../axum-core", version = "0.3.4" }
bytes = "1.0"
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
http = "0.2.9"
http-body = "0.4.4"
hyper = { version = "0.14.24", features = ["stream"] }
http = "1.0.0"
http-body = "1.0.0"
http-body-util = "0.1.0"
hyper = { package = "hyper", version = "1.0.0", features = ["server", "http1"] }
itoa = "1.0.5"
matchit = "0.7"
memchr = "2.4.1"
@ -50,14 +51,10 @@ tower = { version = "0.4.13", default-features = false, features = ["util"] }
tower-layer = "0.3.2"
tower-service = "0.3"
# wont need this when axum uses http-body 1.0
hyper1 = { package = "hyper", version = "=1.0.0-rc.4", features = ["server", "http1", "http2"] }
tower-hyper-http-body-compat = { version = "0.2", features = ["server", "http1"] }
# optional dependencies
axum-macros = { path = "../axum-macros", version = "0.3.7", optional = true }
base64 = { version = "0.21.0", optional = true }
hyper-util = { git = "https://github.com/hyperium/hyper-util", rev = "d97181a", features = ["auto"], optional = true }
hyper-util = { version = "0.1.1", features = ["tokio", "server", "server-auto"], optional = true }
multer = { version = "2.0.0", optional = true }
serde_json = { version = "1.0", features = ["raw_value"], optional = true }
serde_path_to_error = { version = "0.1.8", optional = true }
@ -68,7 +65,7 @@ tokio-tungstenite = { version = "0.20", optional = true }
tracing = { version = "0.1", default-features = false, optional = true }
[dependencies.tower-http]
version = "0.4"
version = "0.5.0"
optional = true
features = [
# all tower-http features except (de)?compression-zstd which doesn't
@ -138,7 +135,7 @@ features = [
]
[dev-dependencies.tower-http]
version = "0.4"
version = "0.5.0"
features = [
# all tower-http features except (de)?compression-zstd which doesn't
# build on `--target armv5te-unknown-linux-musleabi`

2
axum/README.md
View file

@ -112,7 +112,7 @@ This crate uses `#![forbid(unsafe_code)]` to ensure everything is implemented in
## Minimum supported Rust version
axum's MSRV is 1.65.
axum's MSRV is 1.66.
## Examples

9
axum/benches/benches.rs
View file

@ -3,7 +3,6 @@ use axum::{
routing::{get, post},
Extension, Json, Router,
};
use hyper::server::conn::AddrIncoming;
use serde::{Deserialize, Serialize};
use std::{
io::BufRead,
@ -162,13 +161,7 @@ impl BenchmarkBuilder {
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || {
rt.block_on(async move {
let incoming = AddrIncoming::from_listener(listener).unwrap();
hyper::Server::builder(incoming)
.serve(app.into_make_service())
.await
.unwrap();
});
rt.block_on(axum::serve(listener, app)).unwrap();
});
let mut cmd = Command::new("rewrk");

9
axum/src/docs/method_routing/fallback.md
View file

@ -18,9 +18,7 @@ let app = Router::new().route("/", handler);
async fn fallback(method: Method, uri: Uri) -> (StatusCode, String) {
(StatusCode::NOT_FOUND, format!("`{method}` not allowed for {uri}"))
}
# async {
# hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
# };
# let _: Router = app;
```
## When used with `MethodRouter::merge`
@ -44,10 +42,7 @@ let method_route = one.merge(two);
async fn fallback_one() -> impl IntoResponse { /* ... */ }
async fn fallback_two() -> impl IntoResponse { /* ... */ }
# let app = axum::Router::new().route("/", method_route);
# async {
# hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
# };
# let app: axum::Router = axum::Router::new().route("/", method_route);
```
## Setting the `Allow` header

4
axum/src/docs/method_routing/merge.md
View file

@ -19,7 +19,5 @@ let app = Router::new().route("/", merged);
// Our app now accepts
// - GET /
// - POST /
# async {
# hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
# };
# let _: Router = app;
```

7
axum/src/docs/middleware.md
View file

@ -64,14 +64,11 @@ Some commonly used middleware are:
- [`TraceLayer`](tower_http::trace) for high level tracing/logging.
- [`CorsLayer`](tower_http::cors) for handling CORS.
- [`CompressionLayer`](tower_http::compression) for automatic compression of
responses.
- [`CompressionLayer`](tower_http::compression) for automatic compression of responses.
- [`RequestIdLayer`](tower_http::request_id) and
[`PropagateRequestIdLayer`](tower_http::request_id) set and propagate request
ids.
- [`TimeoutLayer`](tower::timeout::TimeoutLayer) for timeouts. Note this
requires using [`HandleErrorLayer`](crate::error_handling::HandleErrorLayer)
to convert timeouts to responses.
- [`TimeoutLayer`](tower_http::timeout::TimeoutLayer) for timeouts.
# Ordering

1
axum/src/docs/response.md
View file

@ -127,6 +127,7 @@ async fn with_status_extensions() -> impl IntoResponse {
)
}
#[derive(Clone)]
struct Foo(&'static str);
// Or mix and match all the things

4
axum/src/docs/routing/merge.md
View file

@ -32,9 +32,7 @@ let app = Router::new()
// - GET /users
// - GET /users/:id
// - GET /teams
# async {
# hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
# };
# let _: Router = app;
```
# Merging routers with state

4
axum/src/extension.rs
View file

@ -98,7 +98,7 @@ axum_core::__impl_deref!(Extension);
impl<T> IntoResponseParts for Extension<T>
where
T: Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
type Error = Infallible;
@ -110,7 +110,7 @@ where
impl<T> IntoResponse for Extension<T>
where
T: Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
fn into_response(self) -> Response {
let mut res = ().into_response();

18
axum/src/extract/connect_info.rs
View file

@ -4,8 +4,9 @@
//!
//! [`Router::into_make_service_with_connect_info`]: crate::routing::Router::into_make_service_with_connect_info
use crate::extension::AddExtension;
use super::{Extension, FromRequestParts};
use crate::{middleware::AddExtension, serve::IncomingStream};
use async_trait::async_trait;
use http::request::Parts;
use std::{
@ -82,11 +83,16 @@ pub trait Connected<T>: Clone + Send + Sync + 'static {
fn connect_info(target: T) -> Self;
}
impl Connected<IncomingStream<'_>> for SocketAddr {
fn connect_info(target: IncomingStream<'_>) -> Self {
target.remote_addr()
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
const _: () = {
use crate::serve::IncomingStream;
impl Connected<IncomingStream<'_>> for SocketAddr {
fn connect_info(target: IncomingStream<'_>) -> Self {
target.remote_addr()
}
}
}
};
impl<S, C, T> Service<T> for IntoMakeServiceWithConnectInfo<S, C>
where
@ -212,7 +218,7 @@ where
#[cfg(test)]
mod tests {
use super::*;
use crate::{routing::get, test_helpers::TestClient, Router};
use crate::{routing::get, serve::IncomingStream, test_helpers::TestClient, Router};
use std::net::SocketAddr;
use tokio::net::TcpListener;

50
axum/src/extract/multipart.rs
View file

@ -5,16 +5,18 @@
use super::{FromRequest, Request};
use crate::body::Bytes;
use async_trait::async_trait;
use axum_core::__composite_rejection as composite_rejection;
use axum_core::__define_rejection as define_rejection;
use axum_core::body::Body;
use axum_core::response::{IntoResponse, Response};
use axum_core::RequestExt;
use axum_core::{
__composite_rejection as composite_rejection, __define_rejection as define_rejection,
response::{IntoResponse, Response},
RequestExt,
};
use futures_util::stream::Stream;
use http::header::{HeaderMap, CONTENT_TYPE};
use http::StatusCode;
use std::error::Error;
use http::{
header::{HeaderMap, CONTENT_TYPE},
HeaderName, HeaderValue, StatusCode,
};
use std::{
error::Error,
fmt,
pin::Pin,
task::{Context, Poll},
@ -72,11 +74,8 @@ where
async fn from_request(req: Request, _state: &S) -> Result<Self, Self::Rejection> {
let boundary = parse_boundary(req.headers()).ok_or(InvalidBoundary)?;
let stream = match req.with_limited_body() {
Ok(limited) => Body::new(limited),
Err(unlimited) => unlimited.into_body(),
};
let multipart = multer::Multipart::new(stream, boundary);
let stream = req.with_limited_body().into_body();
let multipart = multer::Multipart::new(stream.into_data_stream(), boundary);
Ok(Self { inner: multipart })
}
}
@ -91,8 +90,21 @@ impl Multipart {
.map_err(MultipartError::from_multer)?;
if let Some(field) = field {
// multer still uses http 0.2 which means we cannot directly expose
// `multer::Field::headers`. Instead we have to eagerly convert the headers into http
// 1.0
//
// When the next major version of multer is published we can remove this.
let mut headers = HeaderMap::with_capacity(field.headers().len());
headers.extend(field.headers().into_iter().map(|(name, value)| {
let name = HeaderName::from_bytes(name.as_ref()).unwrap();
let value = HeaderValue::from_bytes(value.as_ref()).unwrap();
(name, value)
}));
Ok(Some(Field {
inner: field,
headers,
_multipart: self,
}))
} else {
@ -105,6 +117,7 @@ impl Multipart {
#[derive(Debug)]
pub struct Field<'a> {
inner: multer::Field<'static>,
headers: HeaderMap,
// multer requires there to only be one live `multer::Field` at any point. This enforces that
// statically, which multer does not do, it returns an error instead.
_multipart: &'a mut Multipart,
@ -142,7 +155,7 @@ impl<'a> Field<'a> {
/// Get a map of headers as [`HeaderMap`].
pub fn headers(&self) -> &HeaderMap {
self.inner.headers()
&self.headers
}
/// Get the full data of the field as [`Bytes`].
@ -249,7 +262,7 @@ fn status_code_from_multer_error(err: &multer::Error) -> StatusCode {
if err
.downcast_ref::<crate::Error>()
.and_then(|err| err.source())
.and_then(|err| err.downcast_ref::<http_body::LengthLimitError>())
.and_then(|err| err.downcast_ref::<http_body_util::LengthLimitError>())
.is_some()
{
return StatusCode::PAYLOAD_TOO_LARGE;
@ -324,6 +337,7 @@ mod tests {
assert_eq!(field.file_name().unwrap(), FILE_NAME);
assert_eq!(field.content_type().unwrap(), CONTENT_TYPE);
assert_eq!(field.headers()["foo"], "bar");
assert_eq!(field.bytes().await.unwrap(), BYTES);
assert!(multipart.next_field().await.unwrap().is_none());
@ -338,7 +352,11 @@ mod tests {
reqwest::multipart::Part::bytes(BYTES)
.file_name(FILE_NAME)
.mime_str(CONTENT_TYPE)
.unwrap(),
.unwrap()
.headers(reqwest::header::HeaderMap::from_iter([(
reqwest::header::HeaderName::from_static("foo"),
reqwest::header::HeaderValue::from_static("bar"),
)])),
);
client.post("/").multipart(form).send().await;

6
axum/src/extract/ws.rs
View file

@ -133,7 +133,7 @@ pub struct WebSocketUpgrade<F = DefaultOnFailedUpgrade> {
/// The chosen protocol sent in the `Sec-WebSocket-Protocol` header of the response.
protocol: Option<HeaderValue>,
sec_websocket_key: HeaderValue,
on_upgrade: hyper1::upgrade::OnUpgrade,
on_upgrade: hyper::upgrade::OnUpgrade,
on_failed_upgrade: F,
sec_websocket_protocol: Option<HeaderValue>,
}
@ -413,7 +413,7 @@ where
let on_upgrade = parts
.extensions
.remove::<hyper1::upgrade::OnUpgrade>()
.remove::<hyper::upgrade::OnUpgrade>()
.ok_or(ConnectionNotUpgradable)?;
let sec_websocket_protocol = parts.headers.get(header::SEC_WEBSOCKET_PROTOCOL).cloned();
@ -456,7 +456,7 @@ fn header_contains(headers: &HeaderMap, key: HeaderName, value: &'static str) ->
/// See [the module level documentation](self) for more details.
#[derive(Debug)]
pub struct WebSocket {
inner: WebSocketStream<TokioIo<hyper1::upgrade::Upgraded>>,
inner: WebSocketStream<TokioIo<hyper::upgrade::Upgraded>>,
protocol: Option<HeaderValue>,
}

6
axum/src/handler/mod.rs
View file

@ -391,9 +391,8 @@ mod tests {
use http::StatusCode;
use std::time::Duration;
use tower_http::{
compression::CompressionLayer, limit::RequestBodyLimitLayer,
map_request_body::MapRequestBodyLayer, map_response_body::MapResponseBodyLayer,
timeout::TimeoutLayer,
limit::RequestBodyLimitLayer, map_request_body::MapRequestBodyLayer,
map_response_body::MapResponseBodyLayer, timeout::TimeoutLayer,
};
#[crate::test]
@ -420,7 +419,6 @@ mod tests {
RequestBodyLimitLayer::new(1024),
TimeoutLayer::new(Duration::from_secs(10)),
MapResponseBodyLayer::new(Body::new),
CompressionLayer::new(),
))
.layer(MapRequestBodyLayer::new(Body::new))
.with_state("foo");

2
axum/src/handler/service.rs
View file

@ -178,7 +178,7 @@ where
}
// for `axum::serve(listener, handler)`
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
const _: () = {
use crate::serve::IncomingStream;

11
axum/src/lib.rs
View file

@ -308,16 +308,11 @@
//! ```toml
//! [dependencies]
//! axum = "<latest-version>"
//! hyper = { version = "<latest-version>", features = ["full"] }
//! tokio = { version = "<latest-version>", features = ["full"] }
//! tower = "<latest-version>"
//! ```
//!
//! The `"full"` feature for hyper and tokio isn't strictly necessary but it's
//! the easiest way to get started.
//!
//! Note that [`hyper::Server`] is re-exported by axum so if that's all you need
//! then you don't have to explicitly depend on hyper.
//! The `"full"` feature for tokio isn't necessary but it's the easiest way to get started.
//!
//! Tower isn't strictly necessary either but helpful for testing. See the
//! testing example in the repo to learn more about testing axum apps.
@ -444,7 +439,7 @@ pub mod handler;
pub mod middleware;
pub mod response;
pub mod routing;
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
pub mod serve;
#[cfg(test)]
@ -473,7 +468,7 @@ pub use axum_core::{BoxError, Error, RequestExt, RequestPartsExt};
#[cfg(feature = "macros")]
pub use axum_macros::debug_handler;
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
#[doc(inline)]
pub use self::serve::serve;

3
axum/src/middleware/from_fn.rs
View file

@ -383,6 +383,7 @@ mod tests {
use super::*;
use crate::{body::Body, routing::get, Router};
use http::{HeaderMap, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt;
#[crate::test]
@ -407,7 +408,7 @@ mod tests {
.await
.unwrap();
assert_eq!(res.status(), StatusCode::OK);
let body = hyper::body::to_bytes(res).await.unwrap();
let body = res.collect().await.unwrap().to_bytes();
assert_eq!(&body[..], b"ok");
}
}

4
axum/src/response/redirect.rs
View file

@ -15,9 +15,7 @@ use http::{header::LOCATION, HeaderValue, StatusCode};
/// let app = Router::new()
/// .route("/old", get(|| async { Redirect::permanent("/new") }))
/// .route("/new", get(|| async { "Hello!" }));
/// # async {
/// # hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
/// # };
/// # let _: Router = app;
/// ```
#[must_use = "needs to be returned from a handler or otherwise turned into a Response to be useful"]
#[derive(Debug, Clone)]

20
axum/src/response/sse.rs
View file

@ -22,9 +22,7 @@
//!
//! Sse::new(stream).keep_alive(KeepAlive::default())
//! }
//! # async {
//! # hyper::Server::bind(&"".parse().unwrap()).serve(app.into_make_service()).await.unwrap();
//! # };
//! # let _: Router = app;
//! ```
use crate::{
@ -40,6 +38,7 @@ use futures_util::{
ready,
stream::{Stream, TryStream},
};
use http_body::Frame;
use pin_project_lite::pin_project;
use std::{
fmt,
@ -129,16 +128,16 @@ where
type Data = Bytes;
type Error = E;
fn poll_data(
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Data, Self::Error>>> {
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
let this = self.project();
match this.event_stream.get_pin_mut().poll_next(cx) {
Poll::Pending => {
if let Some(keep_alive) = this.keep_alive.as_pin_mut() {
keep_alive.poll_event(cx).map(|e| Some(Ok(e)))
keep_alive.poll_event(cx).map(|e| Some(Ok(Frame::data(e))))
} else {
Poll::Pending
}
@ -147,19 +146,12 @@ where
if let Some(keep_alive) = this.keep_alive.as_pin_mut() {
keep_alive.reset();
}
Poll::Ready(Some(Ok(event.finalize())))
Poll::Ready(Some(Ok(Frame::data(event.finalize()))))
}
Poll::Ready(Some(Err(error))) => Poll::Ready(Some(Err(error))),
Poll::Ready(None) => Poll::Ready(None),
}
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
Poll::Ready(Ok(None))
}
}
/// Server-sent event

6
axum/src/routing/method_routing.rs
View file

@ -1225,7 +1225,7 @@ where
}
// for `axum::serve(listener, router)`
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
const _: () = {
use crate::serve::IncomingStream;
@ -1250,6 +1250,7 @@ mod tests {
use crate::{body::Body, extract::State, handler::HandlerWithoutStateExt};
use axum_core::response::IntoResponse;
use http::{header::ALLOW, HeaderMap};
use http_body_util::BodyExt;
use std::time::Duration;
use tower::{Service, ServiceExt};
use tower_http::{
@ -1539,7 +1540,8 @@ mod tests {
.unwrap()
.into_response();
let (parts, body) = response.into_parts();
let body = String::from_utf8(hyper::body::to_bytes(body).await.unwrap().to_vec()).unwrap();
let body =
String::from_utf8(BodyExt::collect(body).await.unwrap().to_bytes().to_vec()).unwrap();
(parts.status, parts.headers, body)
}

5
axum/src/routing/mod.rs
View file

@ -397,9 +397,6 @@ impl Router {
/// Convert this router into a [`MakeService`], that is a [`Service`] whose
/// response is another service.
///
/// This is useful when running your application with hyper's
/// [`Server`](hyper::server::Server):
///
/// ```
/// use axum::{
/// routing::get,
@ -431,7 +428,7 @@ impl Router {
}
// for `axum::serve(listener, router)`
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
const _: () = {
use crate::serve::IncomingStream;

4
axum/src/routing/tests/get_to_head.rs
View file

@ -32,7 +32,7 @@ mod for_handlers {
assert_eq!(res.status(), StatusCode::OK);
assert_eq!(res.headers()["x-some-header"], "foobar");
let body = hyper::body::to_bytes(res.into_body()).await.unwrap();
let body = BodyExt::collect(res.into_body()).await.unwrap().to_bytes();
assert_eq!(body.len(), 0);
}
}
@ -67,7 +67,7 @@ mod for_services {
assert_eq!(res.status(), StatusCode::OK);
assert_eq!(res.headers()["x-some-header"], "foobar");
let body = hyper::body::to_bytes(res.into_body()).await.unwrap();
let body = BodyExt::collect(res.into_body()).await.unwrap().to_bytes();
assert_eq!(body.len(), 0);
}
}

21
axum/src/routing/tests/mod.rs
View file

@ -17,10 +17,10 @@ use crate::{
use axum_core::extract::Request;
use futures_util::stream::StreamExt;
use http::{
header::CONTENT_LENGTH,
header::{ALLOW, HOST},
header::{ALLOW, CONTENT_LENGTH, HOST},
HeaderMap, Method, StatusCode, Uri,
};
use http_body_util::BodyExt;
use serde::Deserialize;
use serde_json::json;
use std::{
@ -119,7 +119,7 @@ async fn router_type_doesnt_change() {
on(MethodFilter::GET, |_: Request| async { "hi from GET" })
.on(MethodFilter::POST, |_: Request| async { "hi from POST" }),
)
.layer(tower_http::compression::CompressionLayer::new());
.layer(tower_http::trace::TraceLayer::new_for_http());
let client = TestClient::new(app);
@ -180,16 +180,13 @@ async fn routing_between_services() {
#[crate::test]
async fn middleware_on_single_route() {
use tower_http::{compression::CompressionLayer, trace::TraceLayer};
use tower_http::trace::TraceLayer;
async fn handle(_: Request) -> &'static str {
"Hello, World!"
}
let app = Router::new().route(
"/",
get(handle.layer((TraceLayer::new_for_http(), CompressionLayer::new()))),
);
let app = Router::new().route("/", get(handle.layer(TraceLayer::new_for_http())));
let client = TestClient::new(app);
@ -934,7 +931,7 @@ async fn state_isnt_cloned_too_much() {
impl Clone for AppState {
fn clone(&self) -> Self {
#[rustversion::since(1.65)]
#[rustversion::since(1.66)]
#[track_caller]
fn count() {
if SETUP_DONE.load(Ordering::SeqCst) {
@ -950,7 +947,7 @@ async fn state_isnt_cloned_too_much() {
}
}
#[rustversion::not(since(1.65))]
#[rustversion::not(since(1.66))]
fn count() {
if SETUP_DONE.load(Ordering::SeqCst) {
COUNT.fetch_add(1, Ordering::SeqCst);
@ -1058,7 +1055,7 @@ async fn connect_going_to_custom_fallback() {
let res = app.oneshot(req).await.unwrap();
assert_eq!(res.status(), StatusCode::NOT_FOUND);
let text = String::from_utf8(hyper::body::to_bytes(res).await.unwrap().to_vec()).unwrap();
let text = String::from_utf8(res.collect().await.unwrap().to_bytes().to_vec()).unwrap();
assert_eq!(text, "custom fallback");
}
@ -1076,7 +1073,7 @@ async fn connect_going_to_default_fallback() {
let res = app.oneshot(req).await.unwrap();
assert_eq!(res.status(), StatusCode::NOT_FOUND);
let body = hyper::body::to_bytes(res).await.unwrap();
let body = res.collect().await.unwrap().to_bytes();
assert!(body.is_empty());
}

1
axum/src/routing/url_params.rs
View file

@ -3,6 +3,7 @@ use http::Extensions;
use matchit::Params;
use std::sync::Arc;
#[derive(Clone)]
pub(crate) enum UrlParams {
Params(Vec<(Arc<str>, PercentDecodedStr)>),
InvalidUtf8InPathParam { key: Arc<str> },

106
axum/src/serve.rs
View file

@ -1,15 +1,24 @@
//! Serve services.
use std::{convert::Infallible, io, net::SocketAddr};
use std::{
convert::Infallible,
future::Future,
io,
net::SocketAddr,
pin::Pin,
task::{Context, Poll},
};
use axum_core::{body::Body, extract::Request, response::Response};
use futures_util::{future::poll_fn, FutureExt};
use futures_util::future::poll_fn;
use hyper::body::Incoming;
use hyper_util::{
rt::{TokioExecutor, TokioIo},
server::conn::auto::Builder,
};
use pin_project_lite::pin_project;
use tokio::net::{TcpListener, TcpStream};
use tower_hyper_http_body_compat::{HttpBody04ToHttpBody1, HttpBody1ToHttpBody04};
use tower::util::{Oneshot, ServiceExt};
use tower_service::Service;
/// Serve the service with the supplied listener.
@ -76,7 +85,7 @@ use tower_service::Service;
/// [`Handler`]: crate::handler::Handler
/// [`HandlerWithoutStateExt::into_make_service_with_connect_info`]: crate::handler::HandlerWithoutStateExt::into_make_service_with_connect_info
/// [`HandlerService::into_make_service_with_connect_info`]: crate::handler::HandlerService::into_make_service_with_connect_info
#[cfg(feature = "tokio")]
#[cfg(all(feature = "tokio", any(feature = "http1", feature = "http2")))]
pub async fn serve<M, S>(tcp_listener: TcpListener, mut make_service: M) -> io::Result<()>
where
M: for<'a> Service<IncomingStream<'a>, Error = Infallible, Response = S>,
@ -91,7 +100,7 @@ where
.await
.unwrap_or_else(|err| match err {});
let service = make_service
let tower_service = make_service
.call(IncomingStream {
tcp_stream: &tcp_stream,
remote_addr,
@ -99,50 +108,14 @@ where
.await
.unwrap_or_else(|err| match err {});
let service = hyper1::service::service_fn(move |req: Request<hyper1::body::Incoming>| {
// `hyper1::service::service_fn` takes an `Fn` closure. So we need an owned service in
// order to call `poll_ready` and `call` which need `&mut self`
let mut service = service.clone();
let req = req.map(|body| {
// wont need this when axum uses http-body 1.0
let http_body_04 = HttpBody1ToHttpBody04::new(body);
Body::new(http_body_04)
});
// doing this saves cloning the service just to await the service being ready
//
// services like `Router` are always ready, so assume the service
// we're running here is also always ready...
match poll_fn(|cx| service.poll_ready(cx)).now_or_never() {
Some(Ok(())) => {}
Some(Err(err)) => match err {},
None => {
// ...otherwise load shed
let mut res = Response::new(HttpBody04ToHttpBody1::new(Body::empty()));
*res.status_mut() = http::StatusCode::SERVICE_UNAVAILABLE;
return std::future::ready(Ok(res)).left_future();
}
}
let future = service.call(req);
async move {
let response = future
.await
.unwrap_or_else(|err| match err {})
// wont need this when axum uses http-body 1.0
.map(HttpBody04ToHttpBody1::new);
Ok::<_, Infallible>(response)
}
.right_future()
});
let hyper_service = TowerToHyperService {
service: tower_service,
};
tokio::task::spawn(async move {
match Builder::new(TokioExecutor::new())
// upgrades needed for websockets
.serve_connection_with_upgrades(tcp_stream.into_inner(), service)
.serve_connection_with_upgrades(tcp_stream, hyper_service)
.await
{
Ok(()) => {}
@ -158,6 +131,49 @@ where
}
}
#[derive(Debug, Copy, Clone)]
struct TowerToHyperService<S> {
service: S,
}
impl<S> hyper::service::Service<Request<Incoming>> for TowerToHyperService<S>
where
S: tower_service::Service<Request> + Clone,
{
type Response = S::Response;
type Error = S::Error;
type Future = TowerToHyperServiceFuture<S, Request>;
fn call(&self, req: Request<Incoming>) -> Self::Future {
let req = req.map(Body::new);
TowerToHyperServiceFuture {
future: self.service.clone().oneshot(req),
}
}
}
pin_project! {
struct TowerToHyperServiceFuture<S, R>
where
S: tower_service::Service<R>,
{
#[pin]
future: Oneshot<S, R>,
}
}
impl<S, R> Future for TowerToHyperServiceFuture<S, R>
where
S: tower_service::Service<R>,
{
type Output = Result<S::Response, S::Error>;
#[inline]
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.project().future.poll(cx)
}
}
/// An incoming stream.
///
/// Used with [`serve`] and [`IntoMakeServiceWithConnectInfo`].

23
axum/src/test_helpers/test_client.rs
View file

@ -4,7 +4,7 @@ use http::{
header::{HeaderName, HeaderValue},
StatusCode,
};
use std::{convert::Infallible, net::SocketAddr};
use std::{convert::Infallible, net::SocketAddr, str::FromStr};
use tokio::net::TcpListener;
use tower::make::Shared;
use tower_service::Service;
@ -105,7 +105,15 @@ impl RequestBuilder {
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
{
// reqwest still uses http 0.2
let key: HeaderName = key.try_into().map_err(Into::into).unwrap();
let key = reqwest::header::HeaderName::from_bytes(key.as_ref()).unwrap();
let value: HeaderValue = value.try_into().map_err(Into::into).unwrap();
let value = reqwest::header::HeaderValue::from_bytes(value.as_bytes()).unwrap();
self.builder = self.builder.header(key, value);
self
}
@ -140,11 +148,18 @@ impl TestResponse {
}
pub(crate) fn status(&self) -> StatusCode {
self.response.status()
StatusCode::from_u16(self.response.status().as_u16()).unwrap()
}
pub(crate) fn headers(&self) -> &http::HeaderMap {
self.response.headers()
pub(crate) fn headers(&self) -> http::HeaderMap {
// reqwest still uses http 0.2 so have to convert into http 1.0
let mut headers = http::HeaderMap::new();
for (key, value) in self.response.headers() {
let key = http::HeaderName::from_str(key.as_str()).unwrap();
let value = http::HeaderValue::from_bytes(value.as_bytes()).unwrap();
headers.insert(key, value);
}
headers
}
pub(crate) async fn chunk(&mut self) -> Option<Bytes> {

5
examples/consume-body-in-extractor-or-middleware/Cargo.toml
View file

@ -6,9 +6,10 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = "0.14"
http-body-util = "0.1.0"
hyper = "1.0.0"
tokio = { version = "1.0", features = ["full"] }
tower = "0.4"
tower-http = { version = "0.4.0", features = ["map-request-body", "util"] }
tower-http = { version = "0.5.0", features = ["map-request-body", "util"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

7
examples/consume-body-in-extractor-or-middleware/src/main.rs
View file

@ -14,6 +14,7 @@ use axum::{
routing::post,
Router,
};
use http_body_util::BodyExt;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
@ -50,9 +51,11 @@ async fn buffer_request_body(request: Request) -> Result<Request, Response> {
let (parts, body) = request.into_parts();
// this wont work if the body is an long running stream
let bytes = hyper::body::to_bytes(body)
let bytes = body
.collect()
.await
.map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())?;
.map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())?
.to_bytes();
do_thing_with_request_body(bytes.clone());

2
examples/cors/Cargo.toml
View file

@ -7,4 +7,4 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.4.0", features = ["cors"] }
tower-http = { version = "0.5.0", features = ["cors"] }

2
examples/graceful-shutdown/Cargo.toml
View file

@ -6,5 +6,5 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }

87
examples/graceful-shutdown/src/main.rs
View file

@ -5,51 +5,56 @@
//! kill or ctrl-c
//! ```
use axum::{response::Html, routing::get, Router};
use std::net::SocketAddr;
use tokio::signal;
#[tokio::main]
async fn main() {
// build our application with a route
let app = Router::new().route("/", get(handler));
// run it
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
println!("listening on {addr}");
hyper::Server::bind(&addr)
.serve(app.into_make_service())
.with_graceful_shutdown(shutdown_signal())
.await
.unwrap();
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
async fn handler() -> Html<&'static str> {
Html("<h1>Hello, World!</h1>")
}
// use axum::{response::Html, routing::get, Router};
// use std::net::SocketAddr;
// use tokio::signal;
async fn shutdown_signal() {
let ctrl_c = async {
signal::ctrl_c()
.await
.expect("failed to install Ctrl+C handler");
};
// #[tokio::main]
// async fn main() {
// // build our application with a route
// let app = Router::new().route("/", get(handler));
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install signal handler")
.recv()
.await;
};
// // run it
// let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
// println!("listening on {}", addr);
// hyper::Server::bind(&addr)
// .serve(app.into_make_service())
// .with_graceful_shutdown(shutdown_signal())
// .await
// .unwrap();
// }
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
// async fn handler() -> Html<&'static str> {
// Html("<h1>Hello, World!</h1>")
// }
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
// async fn shutdown_signal() {
// let ctrl_c = async {
// signal::ctrl_c()
// .await
// .expect("failed to install Ctrl+C handler");
// };
println!("signal received, starting graceful shutdown");
}
// #[cfg(unix)]
// let terminate = async {
// signal::unix::signal(signal::unix::SignalKind::terminate())
// .expect("failed to install signal handler")
// .recv()
// .await;
// };
// #[cfg(not(unix))]
// let terminate = std::future::pending::<()>();
// tokio::select! {
// _ = ctrl_c => {},
// _ = terminate => {},
// }
// println!("signal received, starting graceful shutdown");
// }

3
examples/handle-head-request/Cargo.toml
View file

@ -9,5 +9,6 @@ axum = { path = "../../axum" }
tokio = { version = "1.0", features = ["full"] }
[dev-dependencies]
hyper = { version = "0.14", features = ["full"] }
http-body-util = "0.1.0"
hyper = { version = "1.0.0", features = ["full"] }
tower = { version = "0.4", features = ["util"] }

5
examples/handle-head-request/src/main.rs
View file

@ -44,6 +44,7 @@ mod tests {
use super::*;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt;
#[tokio::test]
@ -58,7 +59,7 @@ mod tests {
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["x-some-header"], "header from GET");
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
let body = response.collect().await.unwrap().to_bytes();
assert_eq!(&body[..], b"body from GET");
}
@ -74,7 +75,7 @@ mod tests {
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["x-some-header"], "header from HEAD");
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
let body = response.collect().await.unwrap().to_bytes();
assert!(body.is_empty());
}
}

2
examples/http-proxy/Cargo.toml
View file

@ -6,7 +6,7 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["make"] }
tracing = "0.1"

161
examples/http-proxy/src/main.rs
View file

@ -12,87 +12,96 @@
//!
//! Example is based on <https://github.com/hyperium/hyper/blob/master/examples/http_proxy.rs>
use axum::{
body::Body,
extract::Request,
http::{Method, StatusCode},
response::{IntoResponse, Response},
routing::get,
Router,
};
use hyper::upgrade::Upgraded;
use std::net::SocketAddr;
use tokio::net::TcpStream;
use tower::{make::Shared, ServiceExt};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_http_proxy=trace,tower_http=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let router_svc = Router::new().route("/", get(|| async { "Hello, World!" }));
let service = tower::service_fn(move |req: Request<_>| {
let router_svc = router_svc.clone();
let req = req.map(Body::new);
async move {
if req.method() == Method::CONNECT {
proxy(req).await
} else {
router_svc.oneshot(req).await.map_err(|err| match err {})
}
}
});
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
tracing::debug!("listening on {addr}");
hyper::Server::bind(&addr)
.http1_preserve_header_case(true)
.http1_title_case_headers(true)
.serve(Shared::new(service))
.await
.unwrap();
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
async fn proxy(req: Request) -> Result<Response, hyper::Error> {
tracing::trace!(?req);
// use axum::{
// body::Body,
// extract::Request,
// http::{Method, StatusCode},
// response::{IntoResponse, Response},
// routing::get,
// Router,
// };
// use hyper::upgrade::Upgraded;
// use std::net::SocketAddr;
// use tokio::net::TcpStream;
// use tower::{make::Shared, ServiceExt};
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
if let Some(host_addr) = req.uri().authority().map(|auth| auth.to_string()) {
tokio::task::spawn(async move {
match hyper::upgrade::on(req).await {
Ok(upgraded) => {
if let Err(e) = tunnel(upgraded, host_addr).await {
tracing::warn!("server io error: {e}");
};
}
Err(e) => tracing::warn!("upgrade error: {e}"),
}
});
// #[tokio::main]
// async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_http_proxy=trace,tower_http=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
Ok(Response::new(Body::empty()))
} else {
tracing::warn!("CONNECT host is not socket addr: {:?}", req.uri());
Ok((
StatusCode::BAD_REQUEST,
"CONNECT must be to a socket address",
)
.into_response())
}
}
// let router_svc = Router::new().route("/", get(|| async { "Hello, World!" }));
async fn tunnel(mut upgraded: Upgraded, addr: String) -> std::io::Result<()> {
let mut server = TcpStream::connect(addr).await?;
// let service = tower::service_fn(move |req: Request<_>| {
// let router_svc = router_svc.clone();
// let req = req.map(Body::new);
// async move {
// if req.method() == Method::CONNECT {
// proxy(req).await
// } else {
// router_svc.oneshot(req).await.map_err(|err| match err {})
// }
// }
// });
let (from_client, from_server) =
tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
// let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
// tracing::debug!("listening on {}", addr);
// hyper::Server::bind(&addr)
// .http1_preserve_header_case(true)
// .http1_title_case_headers(true)
// .serve(Shared::new(service))
// .await
// .unwrap();
// }
tracing::debug!("client wrote {from_client} bytes and received {from_server} bytes");
// async fn proxy(req: Request) -> Result<Response, hyper::Error> {
// tracing::trace!(?req);
Ok(())
}
// if let Some(host_addr) = req.uri().authority().map(|auth| auth.to_string()) {
// tokio::task::spawn(async move {
// match hyper::upgrade::on(req).await {
// Ok(upgraded) => {
// if let Err(e) = tunnel(upgraded, host_addr).await {
// tracing::warn!("server io error: {}", e);
// };
// }
// Err(e) => tracing::warn!("upgrade error: {}", e),
// }
// });
// Ok(Response::new(Body::empty()))
// } else {
// tracing::warn!("CONNECT host is not socket addr: {:?}", req.uri());
// Ok((
// StatusCode::BAD_REQUEST,
// "CONNECT must be to a socket address",
// )
// .into_response())
// }
// }
// async fn tunnel(mut upgraded: Upgraded, addr: String) -> std::io::Result<()> {
// let mut server = TcpStream::connect(addr).await?;
// let (from_client, from_server) =
// tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
// tracing::debug!(
// "client wrote {} bytes and received {} bytes",
// from_client,
// from_server
// );
// Ok(())
// }

15
examples/hyper-1-0/Cargo.toml
View file

@ -1,15 +0,0 @@
[package]
name = "example-hyper-1-0"
version = "0.1.0"
edition = "2021"
publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "=1.0.0-rc.4", features = ["full"] }
hyper-util = { git = "https://github.com/hyperium/hyper-util", rev = "f898015", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.4", features = ["trace"] }
tower-hyper-http-body-compat = { version = "0.2", features = ["http1", "server"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

53
examples/hyper-1-0/src/main.rs
View file

@ -1,53 +0,0 @@
//! Run with
//!
//! ```not_rust
//! cargo run -p example-hyper-1-0
//! ```
use axum::{routing::get, Router};
use std::net::SocketAddr;
use tokio::net::TcpListener;
use tower_http::trace::TraceLayer;
use tower_hyper_http_body_compat::TowerService03HttpServiceAsHyper1HttpService;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
// this is hyper 1.0
use hyper::server::conn::http1;
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_hyper_1_0=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let app = Router::new()
.route("/", get(|| async { "Hello, World!" }))
// we can still add regular tower middleware
.layer(TraceLayer::new_for_http());
// `Router` implements tower-service 0.3's `Service` trait. Convert that to something
// that implements hyper 1.0's `Service` trait.
let service = TowerService03HttpServiceAsHyper1HttpService::new(app);
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
let tcp_listener = TcpListener::bind(addr).await.unwrap();
tracing::debug!("listening on {addr}");
loop {
let (tcp_stream, _) = tcp_listener.accept().await.unwrap();
let tcp_stream = hyper_util::rt::TokioIo::new(tcp_stream);
let service = service.clone();
tokio::task::spawn(async move {
if let Err(http_err) = http1::Builder::new()
.keep_alive(true)
.serve_connection(tcp_stream, service)
.await
{
eprintln!("Error while serving HTTP connection: {http_err}");
}
});
}
}

2
examples/key-value-store/Cargo.toml
View file

@ -8,7 +8,7 @@ publish = false
axum = { path = "../../axum" }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util", "timeout", "load-shed", "limit"] }
tower-http = { version = "0.4.0", features = [
tower-http = { version = "0.5.0", features = [
"add-extension",
"auth",
"compression-full",

2
examples/listen-multiple-addrs/Cargo.toml
View file

@ -6,5 +6,5 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1", features = ["full"] }

104
examples/listen-multiple-addrs/src/main.rs
View file

@ -5,56 +5,68 @@
//! listen on both IPv4 and IPv6 when the IPv6 catch-all listener is used (`::`),
//! [like older versions of Windows.](https://docs.microsoft.com/en-us/windows/win32/winsock/dual-stack-sockets)
use axum::{routing::get, Router};
use hyper::server::{accept::Accept, conn::AddrIncoming};
use std::{
net::{Ipv4Addr, Ipv6Addr, SocketAddr},
pin::Pin,
task::{Context, Poll},
};
//! Showcases how listening on multiple addrs is possible by
//! implementing Accept for a custom struct.
//!
//! This may be useful in cases where the platform does not
//! listen on both IPv4 and IPv6 when the IPv6 catch-all listener is used (`::`),
//! [like older versions of Windows.](https://docs.microsoft.com/en-us/windows/win32/winsock/dual-stack-sockets)
#[tokio::main]
async fn main() {
let app = Router::new().route("/", get(|| async { "Hello, World!" }));
let localhost_v4 = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 8080);
let incoming_v4 = AddrIncoming::bind(&localhost_v4).unwrap();
let localhost_v6 = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 8080);
let incoming_v6 = AddrIncoming::bind(&localhost_v6).unwrap();
let combined = CombinedIncoming {
a: incoming_v4,
b: incoming_v6,
};
hyper::Server::builder(combined)
.serve(app.into_make_service())
.await
.unwrap();
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
struct CombinedIncoming {
a: AddrIncoming,
b: AddrIncoming,
}
// use axum::{routing::get, Router};
// use hyper::server::{accept::Accept, conn::AddrIncoming};
// use std::{
// net::{Ipv4Addr, Ipv6Addr, SocketAddr},
// pin::Pin,
// task::{Context, Poll},
// };
impl Accept for CombinedIncoming {
type Conn = <AddrIncoming as Accept>::Conn;
type Error = <AddrIncoming as Accept>::Error;
// #[tokio::main]
// async fn main() {
// let app = Router::new().route("/", get(|| async { "Hello, World!" }));
fn poll_accept(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
if let Poll::Ready(Some(value)) = Pin::new(&mut self.a).poll_accept(cx) {
return Poll::Ready(Some(value));
}
// let localhost_v4 = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 8080);
// let incoming_v4 = AddrIncoming::bind(&localhost_v4).unwrap();
if let Poll::Ready(Some(value)) = Pin::new(&mut self.b).poll_accept(cx) {
return Poll::Ready(Some(value));
}
// let localhost_v6 = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 8080);
// let incoming_v6 = AddrIncoming::bind(&localhost_v6).unwrap();
Poll::Pending
}
}
// let combined = CombinedIncoming {
// a: incoming_v4,
// b: incoming_v6,
// };
// hyper::Server::builder(combined)
// .serve(app.into_make_service())
// .await
// .unwrap();
// }
// struct CombinedIncoming {
// a: AddrIncoming,
// b: AddrIncoming,
// }
// impl Accept for CombinedIncoming {
// type Conn = <AddrIncoming as Accept>::Conn;
// type Error = <AddrIncoming as Accept>::Error;
// fn poll_accept(
// mut self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
// if let Poll::Ready(Some(value)) = Pin::new(&mut self.a).poll_accept(cx) {
// return Poll::Ready(Some(value));
// }
// if let Poll::Ready(Some(value)) = Pin::new(&mut self.b).poll_accept(cx) {
// return Poll::Ready(Some(value));
// }
// Poll::Pending
// }
// }

2
examples/low-level-openssl/Cargo.toml
View file

@ -7,7 +7,7 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
openssl = "0.10"
tokio = { version = "1", features = ["full"] }
tokio-openssl = "0.6"

173
examples/low-level-openssl/src/main.rs
View file

@ -1,86 +1,91 @@
use openssl::ssl::{Ssl, SslAcceptor, SslFiletype, SslMethod};
use tokio_openssl::SslStream;
use axum::{body::Body, http::Request, routing::get, Router};
use futures_util::future::poll_fn;
use hyper::server::{
accept::Accept,
conn::{AddrIncoming, Http},
};
use std::{path::PathBuf, pin::Pin, sync::Arc};
use tokio::net::TcpListener;
use tower::MakeService;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_low_level_openssl=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let mut tls_builder = SslAcceptor::mozilla_modern_v5(SslMethod::tls()).unwrap();
tls_builder
.set_certificate_file(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("cert.pem"),
SslFiletype::PEM,
)
.unwrap();
tls_builder
.set_private_key_file(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("key.pem"),
SslFiletype::PEM,
)
.unwrap();
tls_builder.check_private_key().unwrap();
let acceptor = tls_builder.build();
let listener = TcpListener::bind("127.0.0.1:3000").await.unwrap();
let mut listener = AddrIncoming::from_listener(listener).unwrap();
let protocol = Arc::new(Http::new());
let mut app = Router::new().route("/", get(handler)).into_make_service();
tracing::info!("listening on https://localhost:3000");
loop {
let stream = poll_fn(|cx| Pin::new(&mut listener).poll_accept(cx))
.await
.unwrap()
.unwrap();
let acceptor = acceptor.clone();
let protocol = protocol.clone();
let svc = MakeService::<_, Request<Body>>::make_service(&mut app, &stream);
tokio::spawn(async move {
let ssl = Ssl::new(acceptor.context()).unwrap();
let mut tls_stream = SslStream::new(ssl, stream).unwrap();
SslStream::accept(Pin::new(&mut tls_stream)).await.unwrap();
let _ = protocol
.serve_connection(tls_stream, svc.await.unwrap())
.await;
});
}
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
async fn handler() -> &'static str {
"Hello, World!"
}
// use openssl::ssl::{Ssl, SslAcceptor, SslFiletype, SslMethod};
// use tokio_openssl::SslStream;
// use axum::{body::Body, http::Request, routing::get, Router};
// use futures_util::future::poll_fn;
// use hyper::server::{
// accept::Accept,
// conn::{AddrIncoming, Http},
// };
// use std::{path::PathBuf, pin::Pin, sync::Arc};
// use tokio::net::TcpListener;
// use tower::MakeService;
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
// #[tokio::main]
// async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_low_level_openssl=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
// let mut tls_builder = SslAcceptor::mozilla_modern_v5(SslMethod::tls()).unwrap();
// tls_builder
// .set_certificate_file(
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("cert.pem"),
// SslFiletype::PEM,
// )
// .unwrap();
// tls_builder
// .set_private_key_file(
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("key.pem"),
// SslFiletype::PEM,
// )
// .unwrap();
// tls_builder.check_private_key().unwrap();
// let acceptor = tls_builder.build();
// let listener = TcpListener::bind("127.0.0.1:3000").await.unwrap();
// let mut listener = AddrIncoming::from_listener(listener).unwrap();
// let protocol = Arc::new(Http::new());
// let mut app = Router::new().route("/", get(handler)).into_make_service();
// tracing::info!("listening on https://localhost:3000");
// loop {
// let stream = poll_fn(|cx| Pin::new(&mut listener).poll_accept(cx))
// .await
// .unwrap()
// .unwrap();
// let acceptor = acceptor.clone();
// let protocol = protocol.clone();
// let svc = MakeService::<_, Request<Body>>::make_service(&mut app, &stream);
// tokio::spawn(async move {
// let ssl = Ssl::new(acceptor.context()).unwrap();
// let mut tls_stream = SslStream::new(ssl, stream).unwrap();
// SslStream::accept(Pin::new(&mut tls_stream)).await.unwrap();
// let _ = protocol
// .serve_connection(tls_stream, svc.await.unwrap())
// .await;
// });
// }
// }
// async fn handler() -> &'static str {
// "Hello, World!"
// }

2
examples/low-level-rustls/Cargo.toml
View file

@ -7,7 +7,7 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
rustls-pemfile = "0.3"
tokio = { version = "1", features = ["full"] }
tokio-rustls = "0.23"

185
examples/low-level-rustls/src/main.rs
View file

@ -4,100 +4,105 @@
//! cargo run -p example-low-level-rustls
//! ```
use axum::{extract::Request, routing::get, Router};
use futures_util::future::poll_fn;
use hyper::server::{
accept::Accept,
conn::{AddrIncoming, Http},
};
use rustls_pemfile::{certs, pkcs8_private_keys};
use std::{
fs::File,
io::BufReader,
path::{Path, PathBuf},
pin::Pin,
sync::Arc,
};
use tokio::net::TcpListener;
use tokio_rustls::{
rustls::{Certificate, PrivateKey, ServerConfig},
TlsAcceptor,
};
use tower::make::MakeService;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_tls_rustls=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let rustls_config = rustls_server_config(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("key.pem"),
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("cert.pem"),
);
let acceptor = TlsAcceptor::from(rustls_config);
let listener = TcpListener::bind("127.0.0.1:3000").await.unwrap();
let mut listener = AddrIncoming::from_listener(listener).unwrap();
let protocol = Arc::new(Http::new());
let mut app = Router::<()>::new()
.route("/", get(handler))
.into_make_service();
loop {
let stream = poll_fn(|cx| Pin::new(&mut listener).poll_accept(cx))
.await
.unwrap()
.unwrap();
let acceptor = acceptor.clone();
let protocol = protocol.clone();
let svc = MakeService::<_, Request<hyper::Body>>::make_service(&mut app, &stream);
tokio::spawn(async move {
if let Ok(stream) = acceptor.accept(stream).await {
let _ = protocol.serve_connection(stream, svc.await.unwrap()).await;
}
});
}
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
async fn handler() -> &'static str {
"Hello, World!"
}
// use axum::{extract::Request, routing::get, Router};
// use futures_util::future::poll_fn;
// use hyper::server::{
// accept::Accept,
// conn::{AddrIncoming, Http},
// };
// use rustls_pemfile::{certs, pkcs8_private_keys};
// use std::{
// fs::File,
// io::BufReader,
// path::{Path, PathBuf},
// pin::Pin,
// sync::Arc,
// };
// use tokio::net::TcpListener;
// use tokio_rustls::{
// rustls::{Certificate, PrivateKey, ServerConfig},
// TlsAcceptor,
// };
// use tower::make::MakeService;
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
fn rustls_server_config(key: impl AsRef<Path>, cert: impl AsRef<Path>) -> Arc<ServerConfig> {
let mut key_reader = BufReader::new(File::open(key).unwrap());
let mut cert_reader = BufReader::new(File::open(cert).unwrap());
// #[tokio::main]
// async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_tls_rustls=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
let key = PrivateKey(pkcs8_private_keys(&mut key_reader).unwrap().remove(0));
let certs = certs(&mut cert_reader)
.unwrap()
.into_iter()
.map(Certificate)
.collect();
// let rustls_config = rustls_server_config(
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("key.pem"),
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("cert.pem"),
// );
let mut config = ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth()
.with_single_cert(certs, key)
.expect("bad certificate/key");
// let acceptor = TlsAcceptor::from(rustls_config);
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// let listener = TcpListener::bind("127.0.0.1:3000").await.unwrap();
// let mut listener = AddrIncoming::from_listener(listener).unwrap();
Arc::new(config)
}
// let protocol = Arc::new(Http::new());
// let mut app = Router::<()>::new()
// .route("/", get(handler))
// .into_make_service();
// loop {
// let stream = poll_fn(|cx| Pin::new(&mut listener).poll_accept(cx))
// .await
// .unwrap()
// .unwrap();
// let acceptor = acceptor.clone();
// let protocol = protocol.clone();
// let svc = MakeService::<_, Request<hyper::Body>>::make_service(&mut app, &stream);
// tokio::spawn(async move {
// if let Ok(stream) = acceptor.accept(stream).await {
// let _ = protocol.serve_connection(stream, svc.await.unwrap()).await;
// }
// });
// }
// }
// async fn handler() -> &'static str {
// "Hello, World!"
// }
// fn rustls_server_config(key: impl AsRef<Path>, cert: impl AsRef<Path>) -> Arc<ServerConfig> {
// let mut key_reader = BufReader::new(File::open(key).unwrap());
// let mut cert_reader = BufReader::new(File::open(cert).unwrap());
// let key = PrivateKey(pkcs8_private_keys(&mut key_reader).unwrap().remove(0));
// let certs = certs(&mut cert_reader)
// .unwrap()
// .into_iter()
// .map(Certificate)
// .collect();
// let mut config = ServerConfig::builder()
// .with_safe_defaults()
// .with_no_client_auth()
// .with_single_cert(certs, key)
// .expect("bad certificate/key");
// config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// Arc::new(config)
// }

2
examples/multipart-form/Cargo.toml
View file

@ -7,6 +7,6 @@ publish = false
[dependencies]
axum = { path = "../../axum", features = ["multipart"] }
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.4.0", features = ["limit", "trace"] }
tower-http = { version = "0.5.0", features = ["limit", "trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

2
examples/oauth/Cargo.toml
View file

@ -9,7 +9,7 @@ anyhow = "1"
async-session = "3.0.0"
axum = { path = "../../axum" }
axum-extra = { path = "../../axum-extra", features = ["typed-header"] }
http = "0.2"
http = "1.0.0"
oauth2 = "4.1"
# Use Rustls because it makes it easier to cross-compile on CI
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "json"] }

5
examples/oauth/src/main.rs
View file

@ -69,10 +69,7 @@ async fn main() {
.unwrap()
);
axum::serve(listener, app)
.await
.context("failed to serve service")
.unwrap();
axum::serve(listener, app).await.unwrap();
}
#[derive(Clone)]

3
examples/print-request-response/Cargo.toml
View file

@ -6,7 +6,8 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
http-body-util = "0.1.0"
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util", "filter"] }
tracing = "0.1"

5
examples/print-request-response/src/main.rs
View file

@ -13,6 +13,7 @@ use axum::{
routing::post,
Router,
};
use http_body_util::BodyExt;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
@ -58,8 +59,8 @@ where
B: axum::body::HttpBody<Data = Bytes>,
B::Error: std::fmt::Display,
{
let bytes = match hyper::body::to_bytes(body).await {
Ok(bytes) => bytes,
let bytes = match body.collect().await {
Ok(collected) => collected.to_bytes(),
Err(err) => {
return Err((
StatusCode::BAD_REQUEST,

3
examples/query-params-with-empty-strings/Cargo.toml
View file

@ -6,7 +6,8 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = "0.14"
http-body-util = "0.1.0"
hyper = "1.0.0"
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util"] }

3
examples/query-params-with-empty-strings/src/main.rs
View file

@ -58,6 +58,7 @@ where
mod tests {
use super::*;
use axum::{body::Body, http::Request};
use http_body_util::BodyExt;
use tower::ServiceExt;
#[tokio::test]
@ -114,7 +115,7 @@ mod tests {
.await
.unwrap()
.into_body();
let bytes = hyper::body::to_bytes(body).await.unwrap();
let bytes = body.collect().await.unwrap().to_bytes();
String::from_utf8(bytes.to_vec()).unwrap()
}
}

2
examples/reqwest-response/Cargo.toml
View file

@ -9,6 +9,6 @@ axum = { path = "../../axum" }
reqwest = { version = "0.11", features = ["stream"] }
tokio = { version = "1.0", features = ["full"] }
tokio-stream = "0.1"
tower-http = { version = "0.4", features = ["trace"] }
tower-http = { version = "0.5.0", features = ["trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

138
examples/reqwest-response/src/main.rs
View file

@ -4,76 +4,80 @@
//! cargo run -p example-reqwest-response
//! ```
use std::{convert::Infallible, time::Duration};
use axum::{
body::{Body, Bytes},
extract::State,
response::{IntoResponse, Response},
routing::get,
Router,
};
use reqwest::{Client, StatusCode};
use tokio_stream::StreamExt;
use tower_http::trace::TraceLayer;
use tracing::Span;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_reqwest_response=debug,tower_http=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let client = Client::new();
let app = Router::new()
.route("/", get(proxy_via_reqwest))
.route("/stream", get(stream_some_data))
// Add some logging so we can see the streams going through
.layer(TraceLayer::new_for_http().on_body_chunk(
|chunk: &Bytes, _latency: Duration, _span: &Span| {
tracing::debug!("streaming {} bytes", chunk.len());
},
))
.with_state(client);
let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
.await
.unwrap();
tracing::debug!("listening on {}", listener.local_addr().unwrap());
axum::serve(listener, app).await.unwrap();
fn main() {
// this examples requires reqwest to be updated to hyper and http 1.0
}
async fn proxy_via_reqwest(State(client): State<Client>) -> Response {
let reqwest_response = match client.get("http://127.0.0.1:3000/stream").send().await {
Ok(res) => res,
Err(err) => {
tracing::error!(%err, "request failed");
return StatusCode::BAD_GATEWAY.into_response();
}
};
// use std::{convert::Infallible, time::Duration};
let mut response_builder = Response::builder().status(reqwest_response.status());
// use axum::{
// body::{Body, Bytes},
// extract::State,
// response::{IntoResponse, Response},
// routing::get,
// Router,
// };
// use reqwest::{Client, StatusCode};
// use tokio_stream::StreamExt;
// use tower_http::trace::TraceLayer;
// use tracing::Span;
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
// This unwrap is fine because we haven't insert any headers yet so there can't be any invalid
// headers
*response_builder.headers_mut().unwrap() = reqwest_response.headers().clone();
// #[tokio::main]
// async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_reqwest_response=debug,tower_http=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
response_builder
.body(Body::from_stream(reqwest_response.bytes_stream()))
// Same goes for this unwrap
.unwrap()
}
// let client = Client::new();
async fn stream_some_data() -> Body {
let stream = tokio_stream::iter(0..5)
.throttle(Duration::from_secs(1))
.map(|n| n.to_string())
.map(Ok::<_, Infallible>);
Body::from_stream(stream)
}
// let app = Router::new()
// .route("/", get(proxy_via_reqwest))
// .route("/stream", get(stream_some_data))
// // Add some logging so we can see the streams going through
// .layer(TraceLayer::new_for_http().on_body_chunk(
// |chunk: &Bytes, _latency: Duration, _span: &Span| {
// tracing::debug!("streaming {} bytes", chunk.len());
// },
// ))
// .with_state(client);
// let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
// .await
// .unwrap();
// tracing::debug!("listening on {}", listener.local_addr().unwrap());
// axum::serve(listener, app).await.unwrap();
// }
// async fn proxy_via_reqwest(State(client): State<Client>) -> Response {
// let reqwest_response = match client.get("http://127.0.0.1:3000/stream").send().await {
// Ok(res) => res,
// Err(err) => {
// tracing::error!(%err, "request failed");
// return StatusCode::BAD_GATEWAY.into_response();
// }
// };
// let mut response_builder = Response::builder().status(reqwest_response.status());
// // This unwrap is fine because we haven't insert any headers yet so there can't be any invalid
// // headers
// *response_builder.headers_mut().unwrap() = reqwest_response.headers().clone();
// response_builder
// .body(Body::from_stream(reqwest_response.bytes_stream()))
// // Same goes for this unwrap
// .unwrap()
// }
// async fn stream_some_data() -> Body {
// let stream = tokio_stream::iter(0..5)
// .throttle(Duration::from_secs(1))
// .map(|n| n.to_string())
// .map(Ok::<_, Infallible>);
// Body::from_stream(stream)
// }

2
examples/rest-grpc-multiplex/Cargo.toml
View file

@ -7,7 +7,7 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
futures = "0.3"
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
prost = "0.11"
tokio = { version = "1", features = ["full"] }
tonic = { version = "0.9" }

135
examples/rest-grpc-multiplex/src/main.rs
View file

@ -4,79 +4,84 @@
//! cargo run -p example-rest-grpc-multiplex
//! ```
use self::multiplex_service::MultiplexService;
use axum::{routing::get, Router};
use proto::{
greeter_server::{Greeter, GreeterServer},
HelloReply, HelloRequest,
};
use std::net::SocketAddr;
use tonic::{Response as TonicResponse, Status};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
mod multiplex_service;
mod proto {
tonic::include_proto!("helloworld");
pub(crate) const FILE_DESCRIPTOR_SET: &[u8] =
tonic::include_file_descriptor_set!("helloworld_descriptor");
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
#[derive(Default)]
struct GrpcServiceImpl {}
// use self::multiplex_service::MultiplexService;
// use axum::{routing::get, Router};
// use proto::{
// greeter_server::{Greeter, GreeterServer},
// HelloReply, HelloRequest,
// };
// use std::net::SocketAddr;
// use tonic::{Response as TonicResponse, Status};
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tonic::async_trait]
impl Greeter for GrpcServiceImpl {
async fn say_hello(
&self,
request: tonic::Request<HelloRequest>,
) -> Result<TonicResponse<HelloReply>, Status> {
tracing::info!("Got a request from {:?}", request.remote_addr());
// mod multiplex_service;
let reply = HelloReply {
message: format!("Hello {}!", request.into_inner().name),
};
// mod proto {
// tonic::include_proto!("helloworld");
Ok(TonicResponse::new(reply))
}
}
// pub(crate) const FILE_DESCRIPTOR_SET: &[u8] =
// tonic::include_file_descriptor_set!("helloworld_descriptor");
// }
async fn web_root() -> &'static str {
"Hello, World!"
}
// #[derive(Default)]
// struct GrpcServiceImpl {}
#[tokio::main]
async fn main() {
// initialize tracing
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_rest_grpc_multiplex=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
// #[tonic::async_trait]
// impl Greeter for GrpcServiceImpl {
// async fn say_hello(
// &self,
// request: tonic::Request<HelloRequest>,
// ) -> Result<TonicResponse<HelloReply>, Status> {
// tracing::info!("Got a request from {:?}", request.remote_addr());
// build the rest service
let rest = Router::new().route("/", get(web_root));
// let reply = HelloReply {
// message: format!("Hello {}!", request.into_inner().name),
// };
// build the grpc service
let reflection_service = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(proto::FILE_DESCRIPTOR_SET)
.build()
.unwrap();
let grpc = tonic::transport::Server::builder()
.add_service(reflection_service)
.add_service(GreeterServer::new(GrpcServiceImpl::default()))
.into_service();
// Ok(TonicResponse::new(reply))
// }
// }
// combine them into one service
let service = MultiplexService::new(rest, grpc);
// async fn web_root() -> &'static str {
// "Hello, World!"
// }
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
tracing::debug!("listening on {addr}");
hyper::Server::bind(&addr)
.serve(tower::make::Shared::new(service))
.await
.unwrap();
}
// #[tokio::main]
// async fn main() {
// // initialize tracing
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_rest_grpc_multiplex=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
// // build the rest service
// let rest = Router::new().route("/", get(web_root));
// // build the grpc service
// let reflection_service = tonic_reflection::server::Builder::configure()
// .register_encoded_file_descriptor_set(proto::FILE_DESCRIPTOR_SET)
// .build()
// .unwrap();
// let grpc = tonic::transport::Server::builder()
// .add_service(reflection_service)
// .add_service(GreeterServer::new(GrpcServiceImpl::default()))
// .into_service();
// // combine them into one service
// let service = MultiplexService::new(rest, grpc);
// let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
// tracing::debug!("listening on {}", addr);
// hyper::Server::bind(&addr)
// .serve(tower::make::Shared::new(service))
// .await
// .unwrap();
// }

1
examples/rest-grpc-multiplex/src/multiplex_service.rs
View file

@ -1,4 +1,5 @@
use axum::{
body::Body,
extract::Request,
http::header::CONTENT_TYPE,
response::{IntoResponse, Response},

2
examples/reverse-proxy/Cargo.toml
View file

@ -5,5 +5,5 @@ edition = "2021"
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1", features = ["full"] }

101
examples/reverse-proxy/src/main.rs
View file

@ -7,58 +7,63 @@
//! cargo run -p example-reverse-proxy
//! ```
use axum::{
body::Body,
extract::{Request, State},
http::uri::Uri,
response::{IntoResponse, Response},
routing::get,
Router,
};
use hyper::{client::HttpConnector, StatusCode};
type Client = hyper::client::Client<HttpConnector, Body>;
#[tokio::main]
async fn main() {
tokio::spawn(server());
let client: Client = hyper::Client::builder().build(HttpConnector::new());
let app = Router::new().route("/", get(handler)).with_state(client);
let listener = tokio::net::TcpListener::bind("127.0.0.1:4000")
.await
.unwrap();
println!("listening on {}", listener.local_addr().unwrap());
axum::serve(listener, app).await.unwrap();
// TODO
fn main() {
eprint!("this example has not yet been updated to hyper 1.0");
}
async fn handler(State(client): State<Client>, mut req: Request) -> Result<Response, StatusCode> {
let path = req.uri().path();
let path_query = req
.uri()
.path_and_query()
.map(|v| v.as_str())
.unwrap_or(path);
// use axum::{
// body::Body,
// extract::{Request, State},
// http::uri::Uri,
// response::{IntoResponse, Response},
// routing::get,
// Router,
// };
// use hyper::{client::HttpConnector, StatusCode};
let uri = format!("http://127.0.0.1:3000{path_query}");
// type Client = hyper::client::Client<HttpConnector, Body>;
*req.uri_mut() = Uri::try_from(uri).unwrap();
// #[tokio::main]
// async fn main() {
// tokio::spawn(server());
Ok(client
.request(req)
.await
.map_err(|_| StatusCode::BAD_REQUEST)?
.into_response())
}
// let client: Client = hyper::Client::builder().build(HttpConnector::new());
async fn server() {
let app = Router::new().route("/", get(|| async { "Hello, world!" }));
// let app = Router::new().route("/", get(handler)).with_state(client);
let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
.await
.unwrap();
println!("listening on {}", listener.local_addr().unwrap());
axum::serve(listener, app).await.unwrap();
}
// let listener = tokio::net::TcpListener::bind("127.0.0.1:4000")
// .await
// .unwrap();
// println!("listening on {}", listener.local_addr().unwrap());
// axum::serve(listener, app).await.unwrap();
// }
// async fn handler(State(client): State<Client>, mut req: Request) -> Result<Response, StatusCode> {
// let path = req.uri().path();
// let path_query = req
// .uri()
// .path_and_query()
// .map(|v| v.as_str())
// .unwrap_or(path);
// let uri = format!("http://127.0.0.1:3000{}", path_query);
// *req.uri_mut() = Uri::try_from(uri).unwrap();
// Ok(client
// .request(req)
// .await
// .map_err(|_| StatusCode::BAD_REQUEST)?
// .into_response())
// }
// async fn server() {
// let app = Router::new().route("/", get(|| async { "Hello, world!" }));
// let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
// .await
// .unwrap();
// println!("listening on {}", listener.local_addr().unwrap());
// axum::serve(listener, app).await.unwrap();
// }

2
examples/simple-router-wasm/Cargo.toml
View file

@ -12,5 +12,5 @@ axum = { path = "../../axum", default-features = false }
# works in wasm as well
axum-extra = { path = "../../axum-extra", default-features = false }
futures-executor = "0.3.21"
http = "0.2.7"
http = "1.0.0"
tower-service = "0.3.1"

2
examples/sse/Cargo.toml
View file

@ -11,6 +11,6 @@ futures = "0.3"
headers = "0.3"
tokio = { version = "1.0", features = ["full"] }
tokio-stream = "0.1"
tower-http = { version = "0.4.0", features = ["fs", "trace"] }
tower-http = { version = "0.5.0", features = ["fs", "trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

2
examples/static-file-server/Cargo.toml
View file

@ -9,6 +9,6 @@ axum = { path = "../../axum" }
axum-extra = { path = "../../axum-extra" }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util"] }
tower-http = { version = "0.4.0", features = ["fs", "trace"] }
tower-http = { version = "0.5.0", features = ["fs", "trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

2
examples/stream-to-file/src/main.rs
View file

@ -53,7 +53,7 @@ async fn save_request_body(
Path(file_name): Path<String>,
request: Request,
) -> Result<(), (StatusCode, String)> {
stream_to_file(&file_name, request.into_body()).await
stream_to_file(&file_name, request.into_body().into_data_stream()).await
}
// Handler that returns HTML for a multipart form.

2
examples/testing-websockets/Cargo.toml
View file

@ -7,6 +7,6 @@ publish = false
[dependencies]
axum = { path = "../../axum", features = ["ws"] }
futures = "0.3"
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }
tokio-tungstenite = "0.20"

5
examples/testing/Cargo.toml
View file

@ -6,11 +6,12 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
hyper = { version = "0.14", features = ["full"] }
http-body-util = "0.1.0"
hyper = { version = "1.0.0", features = ["full"] }
mime = "0.3"
serde_json = "1.0"
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.4.0", features = ["trace"] }
tower-http = { version = "0.5.0", features = ["trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

348
examples/testing/src/main.rs
View file

@ -4,194 +4,208 @@
//! cargo test -p example-testing
//! ```
use std::net::SocketAddr;
use axum::{
extract::ConnectInfo,
routing::{get, post},
Json, Router,
};
use tower_http::trace::TraceLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_testing=debug,tower_http=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
.await
.unwrap();
tracing::debug!("listening on {}", listener.local_addr().unwrap());
axum::serve(listener, app()).await.unwrap();
fn main() {
// This example has not yet been updated to Hyper 1.0
}
/// Having a function that produces our app makes it easy to call it from tests
/// without having to create an HTTP server.
fn app() -> Router {
Router::new()
.route("/", get(|| async { "Hello, World!" }))
.route(
"/json",
post(|payload: Json<serde_json::Value>| async move {
Json(serde_json::json!({ "data": payload.0 }))
}),
)
.route(
"/requires-connect-into",
get(|ConnectInfo(addr): ConnectInfo<SocketAddr>| async move { format!("Hi {addr}") }),
)
// We can still add middleware
.layer(TraceLayer::new_for_http())
}
//use std::net::SocketAddr;
#[cfg(test)]
mod tests {
use super::*;
use axum::{
body::Body,
extract::connect_info::MockConnectInfo,
http::{self, Request, StatusCode},
};
use serde_json::{json, Value};
use std::net::SocketAddr;
use tokio::net::TcpListener;
use tower::Service; // for `call`
use tower::ServiceExt; // for `oneshot` and `ready`
//use axum::{
// extract::ConnectInfo,
// routing::{get, post},
// Json, Router,
//};
//use tower_http::trace::TraceLayer;
//use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::test]
async fn hello_world() {
let app = app();
//#[tokio::main]
//async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_testing=debug,tower_http=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
// `Router` implements `tower::Service<Request<Body>>` so we can
// call it like any tower service, no need to run an HTTP server.
let response = app
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
// let listener = tokio::net::TcpListener::bind("127.0.0.1:3000")
// .await
// .unwrap();
// tracing::debug!("listening on {}", listener.local_addr().unwrap());
// axum::serve(listener, app()).await.unwrap();
//}
assert_eq!(response.status(), StatusCode::OK);
///// Having a function that produces our app makes it easy to call it from tests
///// without having to create an HTTP server.
//fn app() -> Router {
// Router::new()
// .route("/", get(|| async { "Hello, World!" }))
// .route(
// "/json",
// post(|payload: Json<serde_json::Value>| async move {
// Json(serde_json::json!({ "data": payload.0 }))
// }),
// )
// .route(
// "/requires-connect-into",
// get(|ConnectInfo(addr): ConnectInfo<SocketAddr>| async move { format!("Hi {addr}") }),
// )
// // We can still add middleware
// .layer(TraceLayer::new_for_http())
//}
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
assert_eq!(&body[..], b"Hello, World!");
}
//#[cfg(test)]
//mod tests {
// use super::*;
// use axum::{
// body::Body,
// extract::connect_info::MockConnectInfo,
// http::{self, Request, StatusCode},
// };
// use http_body_util::BodyExt;
// use serde_json::{json, Value};
// use std::net::SocketAddr;
// use tokio::net::{TcpListener, TcpStream};
// use tower::Service; // for `call`
// use tower::ServiceExt; // for `oneshot` and `ready` // for `collect`
#[tokio::test]
async fn json() {
let app = app();
// #[tokio::test]
// async fn hello_world() {
// let app = app();
let response = app
.oneshot(
Request::builder()
.method(http::Method::POST)
.uri("/json")
.header(http::header::CONTENT_TYPE, mime::APPLICATION_JSON.as_ref())
.body(Body::from(
serde_json::to_vec(&json!([1, 2, 3, 4])).unwrap(),
))
.unwrap(),
)
.await
.unwrap();
// // `Router` implements `tower::Service<Request<Body>>` so we can
// // call it like any tower service, no need to run an HTTP server.
// let response = app
// .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
// .await
// .unwrap();
assert_eq!(response.status(), StatusCode::OK);
// assert_eq!(response.status(), StatusCode::OK);
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
let body: Value = serde_json::from_slice(&body).unwrap();
assert_eq!(body, json!({ "data": [1, 2, 3, 4] }));
}
// let body = response.into_body().collect().await.unwrap().to_bytes();
// assert_eq!(&body[..], b"Hello, World!");
// }
#[tokio::test]
async fn not_found() {
let app = app();
// #[tokio::test]
// async fn json() {
// let app = app();
let response = app
.oneshot(
Request::builder()
.uri("/does-not-exist")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
// let response = app
// .oneshot(
// Request::builder()
// .method(http::Method::POST)
// .uri("/json")
// .header(http::header::CONTENT_TYPE, mime::APPLICATION_JSON.as_ref())
// .body(Body::from(
// serde_json::to_vec(&json!([1, 2, 3, 4])).unwrap(),
// ))
// .unwrap(),
// )
// .await
// .unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
assert!(body.is_empty());
}
// assert_eq!(response.status(), StatusCode::OK);
// You can also spawn a server and talk to it like any other HTTP server:
#[tokio::test]
async fn the_real_deal() {
let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
let addr = listener.local_addr().unwrap();
// let body = response.into_body().collect().await.unwrap().to_bytes();
// let body: Value = serde_json::from_slice(&body).unwrap();
// assert_eq!(body, json!({ "data": [1, 2, 3, 4] }));
// }
tokio::spawn(async move {
axum::serve(listener, app()).await.unwrap();
});
// #[tokio::test]
// async fn not_found() {
// let app = app();
let client = hyper::Client::new();
// let response = app
// .oneshot(
// Request::builder()
// .uri("/does-not-exist")
// .body(Body::empty())
// .unwrap(),
// )
// .await
// .unwrap();
let response = client
.request(
Request::builder()
.uri(format!("http://{addr}"))
.body(hyper::Body::empty())
.unwrap(),
)
.await
.unwrap();
// assert_eq!(response.status(), StatusCode::NOT_FOUND);
// let body = response.into_body().collect().await.unwrap().to_bytes();
// assert!(body.is_empty());
// }
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
assert_eq!(&body[..], b"Hello, World!");
}
// // You can also spawn a server and talk to it like any other HTTP server:
// #[tokio::test]
// async fn the_real_deal() {
// // TODO(david): convert this to hyper-util when thats published
// You can use `ready()` and `call()` to avoid using `clone()`
// in multiple request
#[tokio::test]
async fn multiple_request() {
let mut app = app().into_service();
// use hyper::client::conn;
let request = Request::builder().uri("/").body(Body::empty()).unwrap();
let response = ServiceExt::<Request<Body>>::ready(&mut app)
.await
.unwrap()
.call(request)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
// let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
// let addr = listener.local_addr().unwrap();
let request = Request::builder().uri("/").body(Body::empty()).unwrap();
let response = ServiceExt::<Request<Body>>::ready(&mut app)
.await
.unwrap()
.call(request)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
// tokio::spawn(async move {
// axum::serve(listener, app()).await.unwrap();
// });
// Here we're calling `/requires-connect-into` which requires `ConnectInfo`
//
// That is normally set with `Router::into_make_service_with_connect_info` but we can't easily
// use that during tests. The solution is instead to set the `MockConnectInfo` layer during
// tests.
#[tokio::test]
async fn with_into_make_service_with_connect_info() {
let mut app = app()
.layer(MockConnectInfo(SocketAddr::from(([0, 0, 0, 0], 3000))))
.into_service();
// let target_stream = TcpStream::connect(addr).await.unwrap();
let request = Request::builder()
.uri("/requires-connect-into")
.body(Body::empty())
.unwrap();
let response = app.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
}
// let (mut request_sender, connection) = conn::http1::handshake(target_stream).await.unwrap();
// tokio::spawn(async move { connection.await.unwrap() });
// let response = request_sender
// .send_request(
// Request::builder()
// .uri(format!("http://{addr}"))
// .header("Host", "localhost")
// .body(Body::empty())
// .unwrap(),
// )
// .await
// .unwrap();
// let body = response.into_body().collect().await.unwrap().to_bytes();
// assert_eq!(&body[..], b"Hello, World!");
// }
// // You can use `ready()` and `call()` to avoid using `clone()`
// // in multiple request
// #[tokio::test]
// async fn multiple_request() {
// let mut app = app().into_service();
// let request = Request::builder().uri("/").body(Body::empty()).unwrap();
// let response = ServiceExt::<Request<Body>>::ready(&mut app)
// .await
// .unwrap()
// .call(request)
// .await
// .unwrap();
// assert_eq!(response.status(), StatusCode::OK);
// let request = Request::builder().uri("/").body(Body::empty()).unwrap();
// let response = ServiceExt::<Request<Body>>::ready(&mut app)
// .await
// .unwrap()
// .call(request)
// .await
// .unwrap();
// assert_eq!(response.status(), StatusCode::OK);
// }
// // Here we're calling `/requires-connect-into` which requires `ConnectInfo`
// //
// // That is normally set with `Router::into_make_service_with_connect_info` but we can't easily
// // use that during tests. The solution is instead to set the `MockConnectInfo` layer during
// // tests.
// #[tokio::test]
// async fn with_into_make_service_with_connect_info() {
// let mut app = app()
// .layer(MockConnectInfo(SocketAddr::from(([0, 0, 0, 0], 3000))))
// .into_service();
// let request = Request::builder()
// .uri("/requires-connect-into")
// .body(Body::empty())
// .unwrap();
// let response = app.ready().await.unwrap().call(request).await.unwrap();
// assert_eq!(response.status(), StatusCode::OK);
// }
//}

228
examples/tls-graceful-shutdown/src/main.rs
View file

@ -4,136 +4,140 @@
//! cargo run -p example-tls-graceful-shutdown
//! ```
use axum::{
extract::Host,
handler::HandlerWithoutStateExt,
http::{StatusCode, Uri},
response::Redirect,
routing::get,
BoxError, Router,
};
use axum_server::tls_rustls::RustlsConfig;
use std::{future::Future, net::SocketAddr, path::PathBuf, time::Duration};
use tokio::signal;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[derive(Clone, Copy)]
struct Ports {
http: u16,
https: u16,
fn main() {
// This example has not yet been updated to Hyper 1.0
}
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_tls_graceful_shutdown=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
//use axum::{
// extract::Host,
// handler::HandlerWithoutStateExt,
// http::{StatusCode, Uri},
// response::Redirect,
// routing::get,
// BoxError, Router,
//};
//use axum_server::tls_rustls::RustlsConfig;
//use std::{future::Future, net::SocketAddr, path::PathBuf, time::Duration};
//use tokio::signal;
//use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
let ports = Ports {
http: 7878,
https: 3000,
};
//#[derive(Clone, Copy)]
//struct Ports {
// http: u16,
// https: u16,
//}
//Create a handle for our TLS server so the shutdown signal can all shutdown
let handle = axum_server::Handle::new();
//save the future for easy shutting down of redirect server
let shutdown_future = shutdown_signal(handle.clone());
//#[tokio::main]
//async fn main() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_tls_graceful_shutdown=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
// optional: spawn a second server to redirect http requests to this server
tokio::spawn(redirect_http_to_https(ports, shutdown_future));
// let ports = Ports {
// http: 7878,
// https: 3000,
// };
// configure certificate and private key used by https
let config = RustlsConfig::from_pem_file(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("cert.pem"),
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("key.pem"),
)
.await
.unwrap();
// //Create a handle for our TLS server so the shutdown signal can all shutdown
// let handle = axum_server::Handle::new();
// //save the future for easy shutting down of redirect server
// let shutdown_future = shutdown_signal(handle.clone());
let app = Router::new().route("/", get(handler));
// // optional: spawn a second server to redirect http requests to this server
// tokio::spawn(redirect_http_to_https(ports, shutdown_future));
// run https server
let addr = SocketAddr::from(([127, 0, 0, 1], ports.https));
tracing::debug!("listening on {addr}");
axum_server::bind_rustls(addr, config)
.handle(handle)
.serve(app.into_make_service())
.await
.unwrap();
}
// // configure certificate and private key used by https
// let config = RustlsConfig::from_pem_file(
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("cert.pem"),
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("key.pem"),
// )
// .await
// .unwrap();
async fn shutdown_signal(handle: axum_server::Handle) {
let ctrl_c = async {
signal::ctrl_c()
.await
.expect("failed to install Ctrl+C handler");
};
// let app = Router::new().route("/", get(handler));
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install signal handler")
.recv()
.await;
};
// // run https server
// let addr = SocketAddr::from(([127, 0, 0, 1], ports.https));
// tracing::debug!("listening on {addr}");
// axum_server::bind_rustls(addr, config)
// .handle(handle)
// .serve(app.into_make_service())
// .await
// .unwrap();
//}
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
//async fn shutdown_signal(handle: axum_server::Handle) {
// let ctrl_c = async {
// signal::ctrl_c()
// .await
// .expect("failed to install Ctrl+C handler");
// };
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
// #[cfg(unix)]
// let terminate = async {
// signal::unix::signal(signal::unix::SignalKind::terminate())
// .expect("failed to install signal handler")
// .recv()
// .await;
// };
tracing::info!("Received termination signal shutting down");
handle.graceful_shutdown(Some(Duration::from_secs(10))); // 10 secs is how long docker will wait
// to force shutdown
}
// #[cfg(not(unix))]
// let terminate = std::future::pending::<()>();
async fn handler() -> &'static str {
"Hello, World!"
}
// tokio::select! {
// _ = ctrl_c => {},
// _ = terminate => {},
// }
async fn redirect_http_to_https(ports: Ports, signal: impl Future<Output = ()>) {
fn make_https(host: String, uri: Uri, ports: Ports) -> Result<Uri, BoxError> {
let mut parts = uri.into_parts();
// tracing::info!("Received termination signal shutting down");
// handle.graceful_shutdown(Some(Duration::from_secs(10))); // 10 secs is how long docker will wait
// // to force shutdown
//}
parts.scheme = Some(axum::http::uri::Scheme::HTTPS);
//async fn handler() -> &'static str {
// "Hello, World!"
//}
if parts.path_and_query.is_none() {
parts.path_and_query = Some("/".parse().unwrap());
}
//async fn redirect_http_to_https(ports: Ports, signal: impl Future<Output = ()>) {
// fn make_https(host: String, uri: Uri, ports: Ports) -> Result<Uri, BoxError> {
// let mut parts = uri.into_parts();
let https_host = host.replace(&ports.http.to_string(), &ports.https.to_string());
parts.authority = Some(https_host.parse()?);
// parts.scheme = Some(axum::http::uri::Scheme::HTTPS);
Ok(Uri::from_parts(parts)?)
}
// if parts.path_and_query.is_none() {
// parts.path_and_query = Some("/".parse().unwrap());
// }
let redirect = move |Host(host): Host, uri: Uri| async move {
match make_https(host, uri, ports) {
Ok(uri) => Ok(Redirect::permanent(&uri.to_string())),
Err(error) => {
tracing::warn!(%error, "failed to convert URI to HTTPS");
Err(StatusCode::BAD_REQUEST)
}
}
};
// let https_host = host.replace(&ports.http.to_string(), &ports.https.to_string());
// parts.authority = Some(https_host.parse()?);
let addr = SocketAddr::from(([127, 0, 0, 1], ports.http));
//let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
tracing::debug!("listening on {addr}");
hyper::Server::bind(&addr)
.serve(redirect.into_make_service())
.with_graceful_shutdown(signal)
.await
.unwrap();
}
// Ok(Uri::from_parts(parts)?)
// }
// let redirect = move |Host(host): Host, uri: Uri| async move {
// match make_https(host, uri, ports) {
// Ok(uri) => Ok(Redirect::permanent(&uri.to_string())),
// Err(error) => {
// tracing::warn!(%error, "failed to convert URI to HTTPS");
// Err(StatusCode::BAD_REQUEST)
// }
// }
// };
// let addr = SocketAddr::from(([127, 0, 0, 1], ports.http));
// //let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
// tracing::debug!("listening on {addr}");
// hyper::Server::bind(&addr)
// .serve(redirect.into_make_service())
// .with_graceful_shutdown(signal)
// .await
// .unwrap();
//}

71
examples/tls-rustls/src/main.rs
View file

@ -4,6 +4,8 @@
//! cargo run -p example-tls-rustls
//! ```
#![allow(unused_imports)]
use axum::{
extract::Host,
handler::HandlerWithoutStateExt,
@ -16,6 +18,7 @@ use axum_server::tls_rustls::RustlsConfig;
use std::{net::SocketAddr, path::PathBuf};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
#[allow(dead_code)]
#[derive(Clone, Copy)]
struct Ports {
http: u16,
@ -24,48 +27,52 @@ struct Ports {
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "example_tls_rustls=debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
// Updating this example to hyper 1.0 requires axum_server to update first
let ports = Ports {
http: 7878,
https: 3000,
};
// optional: spawn a second server to redirect http requests to this server
tokio::spawn(redirect_http_to_https(ports));
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "example_tls_rustls=debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
// configure certificate and private key used by https
let config = RustlsConfig::from_pem_file(
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("cert.pem"),
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("self_signed_certs")
.join("key.pem"),
)
.await
.unwrap();
// let ports = Ports {
// http: 7878,
// https: 3000,
// };
// // optional: spawn a second server to redirect http requests to this server
// tokio::spawn(redirect_http_to_https(ports));
let app = Router::new().route("/", get(handler));
// // configure certificate and private key used by https
// let config = RustlsConfig::from_pem_file(
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("cert.pem"),
// PathBuf::from(env!("CARGO_MANIFEST_DIR"))
// .join("self_signed_certs")
// .join("key.pem"),
// )
// .await
// .unwrap();
// run https server
let addr = SocketAddr::from(([127, 0, 0, 1], ports.https));
tracing::debug!("listening on {addr}");
axum_server::bind_rustls(addr, config)
.serve(app.into_make_service())
.await
.unwrap();
// let app = Router::new().route("/", get(handler));
// // run https server
// let addr = SocketAddr::from(([127, 0, 0, 1], ports.https));
// tracing::debug!("listening on {}", addr);
// axum_server::bind_rustls(addr, config)
// .await
// .unwrap();
}
#[allow(dead_code)]
async fn handler() -> &'static str {
"Hello, World!"
}
#[allow(dead_code)]
async fn redirect_http_to_https(ports: Ports) {
fn make_https(host: String, uri: Uri, ports: Ports) -> Result<Uri, BoxError> {
let mut parts = uri.into_parts();

2
examples/todos/Cargo.toml
View file

@ -9,7 +9,7 @@ axum = { path = "../../axum" }
serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util", "timeout"] }
tower-http = { version = "0.4.0", features = ["add-extension", "trace"] }
tower-http = { version = "0.5.0", features = ["add-extension", "trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
uuid = { version = "1.0", features = ["serde", "v4"] }

2
examples/tracing-aka-logging/Cargo.toml
View file

@ -7,6 +7,6 @@ publish = false
[dependencies]
axum = { path = "../../axum", features = ["tracing"] }
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.4.0", features = ["trace"] }
tower-http = { version = "0.5.0", features = ["trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

2
examples/unix-domain-socket/Cargo.toml
View file

@ -7,7 +7,7 @@ publish = false
[dependencies]
axum = { path = "../../axum" }
futures = "0.3"
hyper = { version = "0.14", features = ["full"] }
hyper = { version = "1.0.0", features = ["full"] }
tokio = { version = "1.0", features = ["full"] }
tower = { version = "0.4", features = ["util"] }
tracing = "0.1"

301
examples/unix-domain-socket/src/main.rs
View file

@ -4,178 +4,183 @@
//! cargo run -p example-unix-domain-socket
//! ```
#[cfg(unix)]
#[tokio::main]
async fn main() {
unix::server().await;
}
#[cfg(not(unix))]
// TODO
fn main() {
println!("This example requires unix")
eprint!("this example has not yet been updated to hyper 1.0");
}
#[cfg(unix)]
mod unix {
use axum::{
body::Body,
extract::connect_info::{self, ConnectInfo},
http::{Method, Request, StatusCode, Uri},
routing::get,
Router,
};
use futures::ready;
use hyper::{
client::connect::{Connected, Connection},
server::accept::Accept,
};
use std::{
io,
path::PathBuf,
pin::Pin,
sync::Arc,
task::{Context, Poll},
};
use tokio::{
io::{AsyncRead, AsyncWrite},
net::{unix::UCred, UnixListener, UnixStream},
};
use tower::BoxError;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
// #[cfg(unix)]
// #[tokio::main]
// async fn main() {
// unix::server().await;
// }
pub async fn server() {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "debug".into()),
)
.with(tracing_subscriber::fmt::layer())
.init();
// #[cfg(not(unix))]
// fn main() {
// println!("This example requires unix")
// }
let path = PathBuf::from("/tmp/axum/helloworld");
// #[cfg(unix)]
// mod unix {
// use axum::{
// body::Body,
// extract::connect_info::{self, ConnectInfo},
// http::{Method, Request, StatusCode, Uri},
// routing::get,
// Router,
// };
// use futures::ready;
// use hyper::{
// client::connect::{Connected, Connection},
// server::accept::Accept,
// };
// use std::{
// io,
// path::PathBuf,
// pin::Pin,
// sync::Arc,
// task::{Context, Poll},
// };
// use tokio::{
// io::{AsyncRead, AsyncWrite},
// net::{unix::UCred, UnixListener, UnixStream},
// };
// use tower::BoxError;
// use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
let _ = tokio::fs::remove_file(&path).await;
tokio::fs::create_dir_all(path.parent().unwrap())
.await
.unwrap();
// pub async fn server() {
// tracing_subscriber::registry()
// .with(
// tracing_subscriber::EnvFilter::try_from_default_env()
// .unwrap_or_else(|_| "debug".into()),
// )
// .with(tracing_subscriber::fmt::layer())
// .init();
let uds = UnixListener::bind(path.clone()).unwrap();
tokio::spawn(async {
let app = Router::new().route("/", get(handler));
// let path = PathBuf::from("/tmp/axum/helloworld");
hyper::Server::builder(ServerAccept { uds })
.serve(app.into_make_service_with_connect_info::<UdsConnectInfo>())
.await
.unwrap();
});
// let _ = tokio::fs::remove_file(&path).await;
// tokio::fs::create_dir_all(path.parent().unwrap())
// .await
// .unwrap();
let connector = tower::service_fn(move |_: Uri| {
let path = path.clone();
Box::pin(async move {
let stream = UnixStream::connect(path).await?;
Ok::<_, io::Error>(ClientConnection { stream })
})
});
let client = hyper::Client::builder().build(connector);
// let uds = UnixListener::bind(path.clone()).unwrap();
// tokio::spawn(async {
// let app = Router::new().route("/", get(handler));
let request = Request::builder()
.method(Method::GET)
.uri("http://uri-doesnt-matter.com")
.body(Body::empty())
.unwrap();
// hyper::Server::builder(ServerAccept { uds })
// .serve(app.into_make_service_with_connect_info::<UdsConnectInfo>())
// .await
// .unwrap();
// });
let response = client.request(request).await.unwrap();
// let connector = tower::service_fn(move |_: Uri| {
// let path = path.clone();
// Box::pin(async move {
// let stream = UnixStream::connect(path).await?;
// Ok::<_, io::Error>(ClientConnection { stream })
// })
// });
// let client = hyper::Client::builder().build(connector);
assert_eq!(response.status(), StatusCode::OK);
// let request = Request::builder()
// .method(Method::GET)
// .uri("http://uri-doesnt-matter.com")
// .body(Body::empty())
// .unwrap();
let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
let body = String::from_utf8(body.to_vec()).unwrap();
assert_eq!(body, "Hello, World!");
}
// let response = client.request(request).await.unwrap();
async fn handler(ConnectInfo(info): ConnectInfo<UdsConnectInfo>) -> &'static str {
println!("new connection from `{info:?}`");
// assert_eq!(response.status(), StatusCode::OK);
"Hello, World!"
}
// let body = hyper::body::to_bytes(response.into_body()).await.unwrap();
// let body = String::from_utf8(body.to_vec()).unwrap();
// assert_eq!(body, "Hello, World!");
// }
struct ServerAccept {
uds: UnixListener,
}
// async fn handler(ConnectInfo(info): ConnectInfo<UdsConnectInfo>) -> &'static str {
// println!("new connection from `{:?}`", info);
impl Accept for ServerAccept {
type Conn = UnixStream;
type Error = BoxError;
// "Hello, World!"
// }
fn poll_accept(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
let (stream, _addr) = ready!(self.uds.poll_accept(cx))?;
Poll::Ready(Some(Ok(stream)))
}
}
// struct ServerAccept {
// uds: UnixListener,
// }
struct ClientConnection {
stream: UnixStream,
}
// impl Accept for ServerAccept {
// type Conn = UnixStream;
// type Error = BoxError;
impl AsyncWrite for ClientConnection {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, io::Error>> {
Pin::new(&mut self.stream).poll_write(cx, buf)
}
// fn poll_accept(
// self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
// let (stream, _addr) = ready!(self.uds.poll_accept(cx))?;
// Poll::Ready(Some(Ok(stream)))
// }
// }
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), io::Error>> {
Pin::new(&mut self.stream).poll_flush(cx)
}
// struct ClientConnection {
// stream: UnixStream,
// }
fn poll_shutdown(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), io::Error>> {
Pin::new(&mut self.stream).poll_shutdown(cx)
}
}
// impl AsyncWrite for ClientConnection {
// fn poll_write(
// mut self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// buf: &[u8],
// ) -> Poll<Result<usize, io::Error>> {
// Pin::new(&mut self.stream).poll_write(cx, buf)
// }
impl AsyncRead for ClientConnection {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.stream).poll_read(cx, buf)
}
}
// fn poll_flush(
// mut self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// ) -> Poll<Result<(), io::Error>> {
// Pin::new(&mut self.stream).poll_flush(cx)
// }
impl Connection for ClientConnection {
fn connected(&self) -> Connected {
Connected::new()
}
}
// fn poll_shutdown(
// mut self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// ) -> Poll<Result<(), io::Error>> {
// Pin::new(&mut self.stream).poll_shutdown(cx)
// }
// }
#[derive(Clone, Debug)]
#[allow(dead_code)]
struct UdsConnectInfo {
peer_addr: Arc<tokio::net::unix::SocketAddr>,
peer_cred: UCred,
}
// impl AsyncRead for ClientConnection {
// fn poll_read(
// mut self: Pin<&mut Self>,
// cx: &mut Context<'_>,
// buf: &mut tokio::io::ReadBuf<'_>,
// ) -> Poll<io::Result<()>> {
// Pin::new(&mut self.stream).poll_read(cx, buf)
// }
// }
impl connect_info::Connected<&UnixStream> for UdsConnectInfo {
fn connect_info(target: &UnixStream) -> Self {
let peer_addr = target.peer_addr().unwrap();
let peer_cred = target.peer_cred().unwrap();
// impl Connection for ClientConnection {
// fn connected(&self) -> Connected {
// Connected::new()
// }
// }
Self {
peer_addr: Arc::new(peer_addr),
peer_cred,
}
}
}
}
// #[derive(Clone, Debug)]
// #[allow(dead_code)]
// struct UdsConnectInfo {
// peer_addr: Arc<tokio::net::unix::SocketAddr>,
// peer_cred: UCred,
// }
// impl connect_info::Connected<&UnixStream> for UdsConnectInfo {
// fn connect_info(target: &UnixStream) -> Self {
// let peer_addr = target.peer_addr().unwrap();
// let peer_cred = target.peer_cred().unwrap();
// Self {
// peer_addr: Arc::new(peer_addr),
// peer_cred,
// }
// }
// }
// }

2
examples/validator/Cargo.toml
View file

@ -7,7 +7,7 @@ version = "0.1.0"
[dependencies]
async-trait = "0.1.67"
axum = { path = "../../axum" }
http-body = "0.4.3"
http-body = "1.0.0"
serde = { version = "1.0", features = ["derive"] }
thiserror = "1.0.29"
tokio = { version = "1.0", features = ["full"] }

2
examples/websockets/Cargo.toml
View file

@ -13,7 +13,7 @@ headers = "0.3"
tokio = { version = "1.0", features = ["full"] }
tokio-tungstenite = "0.20"
tower = { version = "0.4", features = ["util"] }
tower-http = { version = "0.4.0", features = ["fs", "trace"] }
tower-http = { version = "0.5.0", features = ["fs", "trace"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }