Various fixes and updates for aws-smithy-http including errors and impl futures::Stream for SdkBody
#1733
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,18 +4,17 @@ | |
| */ | ||
|
|
||
| use bytes::Bytes; | ||
| use futures::Stream; | ||
| use http::{HeaderMap, HeaderValue}; | ||
| use http_body::{Body, SizeHint}; | ||
| use pin_project_lite::pin_project; | ||
|
|
||
| use std::error::Error as StdError; | ||
| use std::fmt::{self, Debug, Formatter}; | ||
| use std::pin::Pin; | ||
| use std::sync::Arc; | ||
| use std::task::{Context, Poll}; | ||
|
|
||
| pub type Error = Box<dyn StdError + Send + Sync>; | ||
|
|
||
| pin_project! { | ||
|
|
@@ -35,9 +34,6 @@ pin_project! { | |
| // In the event of retry, this function will be called to generate a new body. See | ||
| // [`try_clone()`](SdkBody::try_clone) | ||
| rebuild: Option<Arc<dyn (Fn() -> Inner) + Send + Sync>>, | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -96,7 +92,6 @@ impl SdkBody { | |
| Self { | ||
| inner: Inner::Dyn { inner: body }, | ||
| rebuild: None, | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -113,31 +108,28 @@ impl SdkBody { | |
| SdkBody { | ||
| inner: initial.inner, | ||
| rebuild: Some(Arc::new(move || f().inner)), | ||
| } | ||
| } | ||
|
|
||
| pub fn taken() -> Self { | ||
| Self { | ||
| inner: Inner::Taken, | ||
| rebuild: None, | ||
| } | ||
| } | ||
|
|
||
| pub fn empty() -> Self { | ||
| Self { | ||
| inner: Inner::Once { inner: None }, | ||
| rebuild: Some(Arc::new(|| Inner::Once { inner: None })), | ||
| } | ||
| } | ||
|
|
||
| fn poll_inner( | ||
| self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| ) -> Poll<Option<Result<Bytes, Error>>> { | ||
| let this = self.project(); | ||
| let polling_result = match this.inner.project() { | ||
| InnerProj::Once { ref mut inner } => { | ||
| let data = inner.take(); | ||
|
|
@@ -154,26 +146,6 @@ impl SdkBody { | |
| } | ||
| }; | ||
|
|
||
| polling_result | ||
| } | ||
|
|
||
|
|
@@ -192,12 +164,9 @@ impl SdkBody { | |
| pub fn try_clone(&self) -> Option<Self> { | ||
| self.rebuild.as_ref().map(|rebuild| { | ||
| let next = rebuild(); | ||
| Self { | ||
| inner: next, | ||
| rebuild: self.rebuild.clone(), | ||
| } | ||
| }) | ||
| } | ||
|
|
@@ -206,11 +175,6 @@ impl SdkBody { | |
| http_body::Body::size_hint(self).exact() | ||
| } | ||
|
|
||
| pub fn map(self, f: impl Fn(SdkBody) -> SdkBody + Sync + Send + 'static) -> SdkBody { | ||
| if self.rebuild.is_some() { | ||
| SdkBody::retryable(move || f(self.try_clone().unwrap())) | ||
|
|
@@ -235,7 +199,6 @@ impl From<Bytes> for SdkBody { | |
| rebuild: Some(Arc::new(move || Inner::Once { | ||
| inner: Some(bytes.clone()), | ||
| })), | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -245,7 +208,6 @@ impl From<hyper::Body> for SdkBody { | |
| SdkBody { | ||
| inner: Inner::Streaming { inner: body }, | ||
| rebuild: None, | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -283,38 +245,17 @@ impl http_body::Body for SdkBody { | |
| self: Pin<&mut Self>, | ||
| _cx: &mut Context<'_>, | ||
| ) -> Poll<Result<Option<HeaderMap<HeaderValue>>, Self::Error>> { | ||
| // `SdkBody`s have no trailers. If returning trailers is necessary, | ||
| // use `SdkBody::map` to wrap this body with another body that does return trailers. | ||
| Poll::Ready(Ok(None)) | ||
| } | ||
|
|
||
| fn is_end_stream(&self) -> bool { | ||
| match &self.inner { | ||
| Inner::Once { inner: None } => true, | ||
| Inner::Once { inner: Some(bytes) } => bytes.is_empty(), | ||
| Inner::Streaming { inner } => inner.is_end_stream(), | ||
| Inner::Dyn { inner } => inner.is_end_stream(), | ||
| Inner::Taken => true, | ||
| } | ||
| } | ||
|
|
@@ -323,13 +264,47 @@ impl http_body::Body for SdkBody { | |
| match &self.inner { | ||
| Inner::Once { inner: None } => SizeHint::with_exact(0), | ||
| Inner::Once { inner: Some(bytes) } => SizeHint::with_exact(bytes.len() as u64), | ||
| Inner::Streaming { inner } => http_body::Body::size_hint(inner), | ||
| Inner::Dyn { inner } => http_body::Body::size_hint(inner), | ||
| Inner::Taken => SizeHint::new(), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| const SIZE_HINT_32_BIT_PANIC_MESSAGE: &str = r#" | ||
| You're running a 32-bit system and this stream's length is too large to be represented with a usize. | ||
| Please limit stream length to less than 4.294Gb or run this program on a 64-bit computer architecture. | ||
| "#; | ||
|
|
||
| impl Stream for SdkBody { | ||
|
There was a problem hiding this comment. The reasoning for not doing this up until now was that There are a couple of crates that need to get revisited as we get closer to a stable release: |
||
| type Item = Result<Bytes, std::io::Error>; | ||
|
|
||
| fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { | ||
| // By converting to IO errors here, we get access to a bunch of `futures` extension traits. | ||
| // We get access to extension traits from `tokio` too! | ||
| // | ||
| // Convert an `SdkBody` into a `tokio::io::AsyncRead` like so: | ||
| // let body = tokio_util::io::StreamReader::new(SdkBody::from(input)); | ||
| // | ||
| // Convert an `SdkBody` into a `futures::io::AsyncRead` like so: | ||
| // let body = SdkBody::from(input).into_async_read(); | ||
| self.poll_data(cx).map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) | ||
| } | ||
|
|
||
| fn size_hint(&self) -> (usize, Option<usize>) { | ||
| let size_hint = http_body::Body::size_hint(self); | ||
| match ( | ||
| size_hint.lower().try_into(), | ||
| size_hint.upper().map(TryInto::try_into).transpose(), | ||
| ) { | ||
| (Ok(lower), Ok(upper)) => (lower, upper), | ||
| (Err(_), _) | (_, Err(_)) => { | ||
| panic!("{}", SIZE_HINT_32_BIT_PANIC_MESSAGE) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod test { | ||
| use crate::body::{BoxBody, SdkBody}; | ||
|
|
@@ -395,4 +370,36 @@ mod test { | |
| fn is_send<T: Send>() {} | ||
| is_send::<SdkBody>() | ||
| } | ||
|
|
||
|
|
||
| #[cfg(feature = "rt-tokio")] | ||
| #[tokio::test] | ||
| async fn sdk_body_can_be_tokio_async_read() { | ||
| use tokio::io::AsyncBufReadExt; | ||
|
|
||
| let body = SdkBody::from("data 1\ndata 2\ndata 3\n"); | ||
| let body = tokio_util::io::StreamReader::new(body); | ||
| let async_buf_read = tokio::io::BufReader::new(body); | ||
|
|
||
| let mut lines = async_buf_read.lines(); | ||
|
|
||
| assert_eq!(lines.next_line().await.unwrap(), Some("data 1".to_owned())); | ||
| assert_eq!(lines.next_line().await.unwrap(), Some("data 2".to_owned())); | ||
| assert_eq!(lines.next_line().await.unwrap(), Some("data 3".to_owned())); | ||
| assert_eq!(lines.next_line().await.unwrap(), None); | ||
| } | ||
|
|
||
| #[cfg(feature = "rt-tokio")] | ||
| #[tokio::test] | ||
| async fn sdk_body_can_be_futures_async_read() { | ||
| use futures::AsyncReadExt; | ||
| use futures::stream::TryStreamExt; | ||
|
|
||
| let input: &[u8] = b"This is a test body."; | ||
| let mut body = SdkBody::from(input).into_async_read(); | ||
| let mut output = [0u8; 20]; | ||
| let _ = body.read(&mut output).await.unwrap(); | ||
|
|
||
| assert_eq!(&output, input); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why was the hyper upgrade necessary?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It wasn't, I just saw that the server folks were using
.20so I figured I'd update our stuff too. I can remove this if you like.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Follow the RFC on this: https://awslabs.github.io/smithy-rs/design/rfcs/rfc0021_dependency_versions.html