1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
use core::pin::Pin;
use core::task::{Context, Poll};
use futures::ready;
use futures::stream::*;
use hyper::body::{Bytes, HttpBody};
use garage_util::error::Error;
pub type BodyType = Box<dyn HttpBody<Data = Bytes, Error = Error> + Send + Unpin>;
type StreamType = Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>;
pub struct StreamBody {
stream: StreamType,
}
impl StreamBody {
pub fn new(stream: StreamType) -> Self {
Self { stream }
}
}
impl HttpBody for StreamBody {
type Data = Bytes;
type Error = Error;
fn poll_data(
mut self: Pin<&mut Self>,
cx: &mut Context,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
match ready!(self.stream.as_mut().poll_next(cx)) {
Some(res) => Poll::Ready(Some(res)),
None => Poll::Ready(None),
}
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context,
) -> Poll<Result<Option<hyper::HeaderMap<hyper::header::HeaderValue>>, Self::Error>> {
Poll::Ready(Ok(None))
}
}
pub struct BytesBody {
bytes: Option<Bytes>,
}
impl BytesBody {
pub fn new(bytes: Bytes) -> Self {
Self { bytes: Some(bytes) }
}
}
impl HttpBody for BytesBody {
type Data = Bytes;
type Error = Error;
fn poll_data(
mut self: Pin<&mut Self>,
_cx: &mut Context,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
Poll::Ready(self.bytes.take().map(Ok))
}
fn poll_trailers(
self: Pin<&mut Self>,
_cx: &mut Context,
) -> Poll<Result<Option<hyper::HeaderMap<hyper::header::HeaderValue>>, Self::Error>> {
Poll::Ready(Ok(None))
}
}
impl From<String> for BytesBody {
fn from(x: String) -> BytesBody {
Self::new(Bytes::from(x))
}
}
impl From<Vec<u8>> for BytesBody {
fn from(x: Vec<u8>) -> BytesBody {
Self::new(Bytes::from(x))
}
}
pub fn empty_body() -> BodyType {
Box::new(BytesBody::from(vec![]))
}
|
