wasi_hyperium/hyperium1/
outgoing.rs

1use std::{
2    pin::Pin,
3    task::{Context, Poll},
4};
5
6use bytes::Buf;
7
8use crate::{
9    outgoing::{Copied, OutgoingBodyCopier},
10    poll::PollableRegistry,
11    wasi::{OutgoingBody, OutgoingRequest, OutgoingResponse},
12    Error,
13};
14
15pub fn outgoing_request<B, Registry>(
16    request: &http1::Request<B>,
17    registry: Registry,
18) -> Result<OutgoingRequest<Registry>, Error>
19where
20    Registry: PollableRegistry,
21{
22    let mut req = OutgoingRequest::from_headers(&request.headers().into(), registry)?;
23    req.set_method(request.method().into())?;
24    if let Some(path_with_query) = request.uri().path_and_query() {
25        req.set_path_with_query(Some(path_with_query.as_str()))?;
26    }
27    if let Some(scheme) = request.uri().scheme() {
28        req.set_scheme(Some(scheme.into()))?;
29    }
30    if let Some(authority) = request.uri().authority() {
31        req.set_authority(Some(authority.as_str()))?;
32    }
33
34    Ok(req)
35}
36
37pub fn outgoing_response<B, Registry>(
38    resp: &http1::Response<B>,
39    registry: Registry,
40) -> Result<OutgoingResponse<Registry>, Error>
41where
42    Registry: PollableRegistry,
43{
44    let mut outgoing = OutgoingResponse::from_headers(&resp.headers().into(), registry)?;
45    outgoing.set_status_code(resp.status().as_u16())?;
46    Ok(outgoing)
47}
48
49pub struct Hyperium1OutgoingBodyCopier<HttpBody, Registry>
50where
51    HttpBody: http_body1::Body,
52    Registry: PollableRegistry,
53{
54    src: HttpBody,
55    dest: Option<OutgoingBody<Registry>>,
56    buf: Option<HttpBody::Data>,
57}
58
59impl<HttpBody, Registry> Hyperium1OutgoingBodyCopier<HttpBody, Registry>
60where
61    HttpBody: http_body1::Body,
62    Registry: PollableRegistry,
63{
64    pub fn new(src: HttpBody, dest: OutgoingBody<Registry>) -> Result<Self, Error> {
65        Ok(Self {
66            src,
67            dest: Some(dest),
68            buf: None,
69        })
70    }
71}
72
73impl<HttpBody, Registry> OutgoingBodyCopier for Hyperium1OutgoingBodyCopier<HttpBody, Registry>
74where
75    HttpBody: http_body1::Body + Unpin,
76    anyhow::Error: From<HttpBody::Error>,
77    Registry: PollableRegistry,
78{
79    fn poll_copy(&mut self, cx: &mut Context) -> Poll<Option<Result<Copied, Error>>> {
80        if self.dest.is_none() {
81            return Poll::Ready(None);
82        }
83
84        if self.buf.is_none() {
85            // Fill buffer
86            match Pin::new(&mut self.src)
87                .poll_frame(cx)
88                .map_err(|err| Error::BodyError(err.into()))?
89            {
90                Poll::Ready(Some(frame)) => {
91                    if frame.is_data() {
92                        self.buf =
93                            Some(frame.into_data().unwrap_or_else(|_| {
94                                panic!("into_data failed when is_data = true")
95                            }));
96                    } else {
97                        // Got trailers; finish outgoing-body
98                        let trailers = frame.into_trailers().unwrap_or_else(|_| {
99                            panic!("into_trailers failed when is_data = false")
100                        });
101                        self.dest.take().unwrap().finish(Some(trailers.into()))?;
102                        return Poll::Ready(Some(Ok(Copied::Trailers)));
103                    }
104                }
105                Poll::Ready(None) => {
106                    // End of body (no trailers); finish outgoing-body
107                    self.dest.take().unwrap().finish(None)?;
108                    return Poll::Ready(None);
109                }
110                Poll::Pending => return Poll::Pending,
111            }
112        }
113
114        // Write buffer
115        let stream = self.dest.as_mut().unwrap().stream();
116        match stream.poll_check_write(cx)? {
117            Poll::Ready(permit) => {
118                let buf = self.buf.as_mut().unwrap();
119                let len = permit.write(buf.chunk())?;
120                buf.advance(len);
121                if !buf.has_remaining() {
122                    self.buf = None;
123                }
124                Poll::Ready(Some(Ok(Copied::Body(len))))
125            }
126            Poll::Pending => Poll::Pending,
127        }
128    }
129}
wasi_hyperium/hyperium1/outgoing.rs

wasi_hyperium/hyperium1/
outgoing.rs
