mirror of
https://github.com/denoland/deno
synced 2024-07-02 16:31:02 +00:00
92 lines
2.6 KiB
Rust
92 lines
2.6 KiB
Rust
// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
|
|
use bytes::Bytes;
|
|
use deno_core::error::AnyError;
|
|
use deno_core::futures::stream::Peekable;
|
|
use deno_core::futures::Stream;
|
|
use deno_core::futures::StreamExt;
|
|
use deno_core::AsyncRefCell;
|
|
use deno_core::AsyncResult;
|
|
use deno_core::BufView;
|
|
use deno_core::RcRef;
|
|
use deno_core::Resource;
|
|
use hyper::body::Body;
|
|
use hyper::body::Incoming;
|
|
use hyper::body::SizeHint;
|
|
use std::borrow::Cow;
|
|
use std::pin::Pin;
|
|
use std::rc::Rc;
|
|
use std::task::ready;
|
|
use std::task::Poll;
|
|
|
|
/// Converts a hyper incoming body stream into a stream of [`Bytes`] that we can use to read in V8.
|
|
struct ReadFuture(Incoming);
|
|
|
|
impl Stream for ReadFuture {
|
|
type Item = Result<Bytes, AnyError>;
|
|
|
|
fn poll_next(
|
|
self: Pin<&mut Self>,
|
|
cx: &mut std::task::Context<'_>,
|
|
) -> Poll<Option<Self::Item>> {
|
|
// Loop until we receive a non-empty frame from Hyper
|
|
let this = self.get_mut();
|
|
loop {
|
|
let res = ready!(Pin::new(&mut this.0).poll_frame(cx));
|
|
break match res {
|
|
Some(Ok(frame)) => {
|
|
if let Ok(data) = frame.into_data() {
|
|
// Ensure that we never yield an empty frame
|
|
if !data.is_empty() {
|
|
break Poll::Ready(Some(Ok::<_, AnyError>(data)));
|
|
}
|
|
}
|
|
// Loop again so we don't lose the waker
|
|
continue;
|
|
}
|
|
Some(Err(e)) => Poll::Ready(Some(Err(e.into()))),
|
|
None => Poll::Ready(None),
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
pub struct HttpRequestBody(AsyncRefCell<Peekable<ReadFuture>>, SizeHint);
|
|
|
|
impl HttpRequestBody {
|
|
pub fn new(body: Incoming) -> Self {
|
|
let size_hint = body.size_hint();
|
|
Self(AsyncRefCell::new(ReadFuture(body).peekable()), size_hint)
|
|
}
|
|
|
|
async fn read(self: Rc<Self>, limit: usize) -> Result<BufView, AnyError> {
|
|
let peekable = RcRef::map(self, |this| &this.0);
|
|
let mut peekable = peekable.borrow_mut().await;
|
|
match Pin::new(&mut *peekable).peek_mut().await {
|
|
None => Ok(BufView::empty()),
|
|
Some(Err(_)) => Err(peekable.next().await.unwrap().err().unwrap()),
|
|
Some(Ok(bytes)) => {
|
|
if bytes.len() <= limit {
|
|
// We can safely take the next item since we peeked it
|
|
return Ok(BufView::from(peekable.next().await.unwrap()?));
|
|
}
|
|
let ret = bytes.split_to(limit);
|
|
Ok(BufView::from(ret))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Resource for HttpRequestBody {
|
|
fn name(&self) -> Cow<str> {
|
|
"requestBody".into()
|
|
}
|
|
|
|
fn read(self: Rc<Self>, limit: usize) -> AsyncResult<BufView> {
|
|
Box::pin(HttpRequestBody::read(self, limit))
|
|
}
|
|
|
|
fn size_hint(&self) -> (u64, Option<u64>) {
|
|
(self.1.lower(), self.1.upper())
|
|
}
|
|
}
|