remove unpin bounds on other body types

This commit is contained in:
Rob Ede 2021-04-12 21:49:08 +01:00
parent 4f35b7494e
commit 5bdf96aa8c
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
4 changed files with 43 additions and 22 deletions

View File

@ -137,7 +137,7 @@ impl From<BytesMut> for Body {
impl<S> From<SizedStream<S>> for Body
where
S: Stream<Item = Result<Bytes, Error>> + Unpin + 'static,
S: Stream<Item = Result<Bytes, Error>> + 'static,
{
fn from(s: SizedStream<S>) -> Body {
Body::from_message(s)
@ -146,7 +146,7 @@ where
impl<S, E> From<BodyStream<S>> for Body
where
S: Stream<Item = Result<Bytes, E>> + Unpin + 'static,
S: Stream<Item = Result<Bytes, E>> + 'static,
E: Into<Error> + 'static,
{
fn from(s: BodyStream<S>) -> Body {

View File

@ -5,21 +5,25 @@ use std::{
use bytes::Bytes;
use futures_core::{ready, Stream};
use pin_project_lite::pin_project;
use crate::error::Error;
use super::{BodySize, MessageBody};
/// Streaming response wrapper.
///
/// Response does not contain `Content-Length` header and appropriate transfer encoding is used.
pub struct BodyStream<S: Unpin> {
pin_project! {
/// Streaming response wrapper.
///
/// Response does not contain `Content-Length` header and appropriate transfer encoding is used.
pub struct BodyStream<S> {
#[pin]
stream: S,
}
}
impl<S, E> BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
S: Stream<Item = Result<Bytes, E>>,
E: Into<Error>,
{
pub fn new(stream: S) -> Self {
@ -29,7 +33,7 @@ where
impl<S, E> MessageBody for BodyStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
S: Stream<Item = Result<Bytes, E>>,
E: Into<Error>,
{
fn size(&self) -> BodySize {
@ -46,9 +50,9 @@ where
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
loop {
let stream = &mut self.as_mut().stream;
let stream = self.as_mut().project().stream;
let chunk = match ready!(Pin::new(stream).poll_next(cx)) {
let chunk = match ready!(stream.poll_next(cx)) {
Some(Ok(ref bytes)) if bytes.is_empty() => continue,
opt => opt.map(|res| res.map_err(Into::into)),
};

View File

@ -52,6 +52,19 @@ impl<T: MessageBody + Unpin> MessageBody for Box<T> {
}
}
impl<T: MessageBody> MessageBody for Pin<Box<T>> {
fn size(&self) -> BodySize {
self.as_ref().size()
}
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
self.as_mut().poll_next(cx)
}
}
impl MessageBody for Bytes {
fn size(&self) -> BodySize {
BodySize::Sized(self.len() as u64)

View File

@ -5,23 +5,27 @@ use std::{
use bytes::Bytes;
use futures_core::{ready, Stream};
use pin_project_lite::pin_project;
use crate::error::Error;
use super::{BodySize, MessageBody};
/// Known sized streaming response wrapper.
///
/// This body implementation should be used if total size of stream is known. Data get sent as is
/// without using transfer encoding.
pub struct SizedStream<S: Unpin> {
pin_project! {
/// Known sized streaming response wrapper.
///
/// This body implementation should be used if total size of stream is known. Data get sent as is
/// without using transfer encoding.
pub struct SizedStream<S> {
size: u64,
#[pin]
stream: S,
}
}
impl<S> SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + Unpin,
S: Stream<Item = Result<Bytes, Error>>,
{
pub fn new(size: u64, stream: S) -> Self {
SizedStream { size, stream }
@ -30,7 +34,7 @@ where
impl<S> MessageBody for SizedStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + Unpin,
S: Stream<Item = Result<Bytes, Error>>,
{
fn size(&self) -> BodySize {
BodySize::Sized(self.size as u64)
@ -46,9 +50,9 @@ where
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Error>>> {
loop {
let stream = &mut self.as_mut().stream;
let stream = self.as_mut().project().stream;
let chunk = match ready!(Pin::new(stream).poll_next(cx)) {
let chunk = match ready!(stream.poll_next(cx)) {
Some(Ok(ref bytes)) if bytes.is_empty() => continue,
val => val,
};