use crate::error::StdError;
use crate::http::Body;
use crate::stream::*;
use std::fmt;
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::Stream;
use hyper::body::Bytes;
pub struct StreamingBlob {
inner: DynByteStream,
}
impl StreamingBlob {
pub fn new<S>(stream: S) -> Self
where
S: ByteStream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
{
Self { inner: Box::pin(stream) }
}
pub fn wrap<S, E>(stream: S) -> Self
where
S: Stream<Item = Result<Bytes, E>> + Send + Sync + 'static,
E: std::error::Error + Send + Sync + 'static,
{
Self { inner: wrap(stream) }
}
fn into_inner(self) -> DynByteStream {
self.inner
}
}
impl fmt::Debug for StreamingBlob {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("StreamingBlob").finish_non_exhaustive()
}
}
impl Stream for StreamingBlob {
type Item = Result<Bytes, StdError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(cx)
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl ByteStream for StreamingBlob {
fn remaining_length(&self) -> RemainingLength {
self.inner.remaining_length()
}
}
impl From<StreamingBlob> for DynByteStream {
fn from(value: StreamingBlob) -> Self {
value.into_inner()
}
}
impl From<StreamingBlob> for Body {
fn from(value: StreamingBlob) -> Self {
Body::from(value.into_inner())
}
}
impl From<Body> for StreamingBlob {
fn from(value: Body) -> Self {
Self::new(value)
}
}