mirror of
https://github.com/denoland/deno.git
synced 2024-10-29 08:58:01 -04:00
689 lines
20 KiB
Rust
689 lines
20 KiB
Rust
// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
|
|
use crate::request_properties::HttpConnectionProperties;
|
|
use crate::response_body::ResponseBytesInner;
|
|
use crate::response_body::ResponseStreamResult;
|
|
use deno_core::error::AnyError;
|
|
use deno_core::futures::ready;
|
|
use deno_core::BufView;
|
|
use deno_core::OpState;
|
|
use deno_core::ResourceId;
|
|
use http_1::request::Parts;
|
|
use hyper1::body::Body;
|
|
use hyper1::body::Frame;
|
|
use hyper1::body::Incoming;
|
|
use hyper1::body::SizeHint;
|
|
use hyper1::header::HeaderMap;
|
|
use hyper1::upgrade::OnUpgrade;
|
|
|
|
use scopeguard::guard;
|
|
use scopeguard::ScopeGuard;
|
|
use std::cell::Cell;
|
|
use std::cell::Ref;
|
|
use std::cell::RefCell;
|
|
use std::cell::RefMut;
|
|
use std::future::Future;
|
|
use std::mem::ManuallyDrop;
|
|
use std::pin::Pin;
|
|
use std::rc::Rc;
|
|
use std::task::Context;
|
|
use std::task::Poll;
|
|
use std::task::Waker;
|
|
|
|
pub type Request = hyper1::Request<Incoming>;
|
|
pub type Response = hyper1::Response<HttpRecordResponse>;
|
|
|
|
#[cfg(feature = "__http_tracing")]
|
|
pub static RECORD_COUNT: std::sync::atomic::AtomicUsize =
|
|
std::sync::atomic::AtomicUsize::new(0);
|
|
|
|
macro_rules! http_general_trace {
|
|
($($args:expr),*) => {
|
|
#[cfg(feature = "__http_tracing")]
|
|
{
|
|
let count = $crate::service::RECORD_COUNT
|
|
.load(std::sync::atomic::Ordering::SeqCst);
|
|
|
|
println!(
|
|
"HTTP [+{count}]: {}",
|
|
format!($($args),*),
|
|
);
|
|
}
|
|
};
|
|
}
|
|
|
|
macro_rules! http_trace {
|
|
($record:expr $(, $args:expr)*) => {
|
|
#[cfg(feature = "__http_tracing")]
|
|
{
|
|
let count = $crate::service::RECORD_COUNT
|
|
.load(std::sync::atomic::Ordering::SeqCst);
|
|
|
|
println!(
|
|
"HTTP [+{count}] id={:p} strong={}: {}",
|
|
$record,
|
|
std::rc::Rc::strong_count(&$record),
|
|
format!($($args),*),
|
|
);
|
|
}
|
|
};
|
|
}
|
|
|
|
pub(crate) use http_general_trace;
|
|
pub(crate) use http_trace;
|
|
|
|
pub(crate) struct HttpServerStateInner {
|
|
pool: Vec<(Rc<HttpRecord>, HeaderMap)>,
|
|
}
|
|
|
|
/// A signalling version of `Rc` that allows one to poll for when all other references
|
|
/// to the `Rc` have been dropped.
|
|
#[repr(transparent)]
|
|
pub(crate) struct SignallingRc<T>(Rc<(T, Cell<Option<Waker>>)>);
|
|
|
|
impl<T> SignallingRc<T> {
|
|
#[inline]
|
|
pub fn new(t: T) -> Self {
|
|
Self(Rc::new((t, Default::default())))
|
|
}
|
|
|
|
#[inline]
|
|
pub fn strong_count(&self) -> usize {
|
|
Rc::strong_count(&self.0)
|
|
}
|
|
|
|
/// Resolves when this is the only remaining reference.
|
|
#[inline]
|
|
pub fn poll_complete(&self, cx: &mut Context<'_>) -> Poll<()> {
|
|
if Rc::strong_count(&self.0) == 1 {
|
|
Poll::Ready(())
|
|
} else {
|
|
self.0 .1.set(Some(cx.waker().clone()));
|
|
Poll::Pending
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> Clone for SignallingRc<T> {
|
|
#[inline]
|
|
fn clone(&self) -> Self {
|
|
Self(self.0.clone())
|
|
}
|
|
}
|
|
|
|
impl<T> Drop for SignallingRc<T> {
|
|
#[inline]
|
|
fn drop(&mut self) {
|
|
// Trigger the waker iff the refcount is about to become 1.
|
|
if Rc::strong_count(&self.0) == 2 {
|
|
if let Some(waker) = self.0 .1.take() {
|
|
waker.wake();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> std::ops::Deref for SignallingRc<T> {
|
|
type Target = T;
|
|
#[inline]
|
|
fn deref(&self) -> &Self::Target {
|
|
&self.0 .0
|
|
}
|
|
}
|
|
|
|
pub(crate) struct HttpServerState(RefCell<HttpServerStateInner>);
|
|
|
|
impl HttpServerState {
|
|
pub fn new() -> SignallingRc<Self> {
|
|
SignallingRc::new(Self(RefCell::new(HttpServerStateInner {
|
|
pool: Vec::new(),
|
|
})))
|
|
}
|
|
}
|
|
|
|
impl std::ops::Deref for HttpServerState {
|
|
type Target = RefCell<HttpServerStateInner>;
|
|
|
|
fn deref(&self) -> &Self::Target {
|
|
&self.0
|
|
}
|
|
}
|
|
|
|
enum RequestBodyState {
|
|
Incoming(Incoming),
|
|
Resource(HttpRequestBodyAutocloser),
|
|
}
|
|
|
|
impl From<Incoming> for RequestBodyState {
|
|
fn from(value: Incoming) -> Self {
|
|
RequestBodyState::Incoming(value)
|
|
}
|
|
}
|
|
|
|
/// Ensures that the request body closes itself when no longer needed.
|
|
pub struct HttpRequestBodyAutocloser(ResourceId, Rc<RefCell<OpState>>);
|
|
|
|
impl HttpRequestBodyAutocloser {
|
|
pub fn new(res: ResourceId, op_state: Rc<RefCell<OpState>>) -> Self {
|
|
Self(res, op_state)
|
|
}
|
|
}
|
|
|
|
impl Drop for HttpRequestBodyAutocloser {
|
|
fn drop(&mut self) {
|
|
if let Ok(res) = self.1.borrow_mut().resource_table.take_any(self.0) {
|
|
res.close();
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) async fn handle_request(
|
|
request: Request,
|
|
request_info: HttpConnectionProperties,
|
|
server_state: SignallingRc<HttpServerState>, // Keep server alive for duration of this future.
|
|
tx: tokio::sync::mpsc::Sender<Rc<HttpRecord>>,
|
|
) -> Result<Response, hyper::Error> {
|
|
// If the underlying TCP connection is closed, this future will be dropped
|
|
// and execution could stop at any await point.
|
|
// The HttpRecord must live until JavaScript is done processing so is wrapped
|
|
// in an Rc. The guard ensures unneeded resources are freed at cancellation.
|
|
let guarded_record = guard(
|
|
HttpRecord::new(request, request_info, server_state),
|
|
HttpRecord::cancel,
|
|
);
|
|
|
|
// Clone HttpRecord and send to JavaScript for processing.
|
|
// Safe to unwrap as channel receiver is never closed.
|
|
tx.send(guarded_record.clone()).await.unwrap();
|
|
|
|
// Wait for JavaScript handler to return request.
|
|
http_trace!(*guarded_record, "handle_request response_ready.await");
|
|
guarded_record.response_ready().await;
|
|
|
|
// Defuse the guard. Must not await after this point.
|
|
let record = ScopeGuard::into_inner(guarded_record);
|
|
http_trace!(record, "handle_request complete");
|
|
let response = record.into_response();
|
|
Ok(response)
|
|
}
|
|
|
|
struct HttpRecordInner {
|
|
server_state: SignallingRc<HttpServerState>,
|
|
request_info: HttpConnectionProperties,
|
|
request_parts: http_1::request::Parts,
|
|
request_body: Option<RequestBodyState>,
|
|
response_parts: Option<http_1::response::Parts>,
|
|
response_ready: bool,
|
|
response_waker: Option<Waker>,
|
|
response_body: ResponseBytesInner,
|
|
response_body_finished: bool,
|
|
response_body_waker: Option<Waker>,
|
|
trailers: Option<HeaderMap>,
|
|
been_dropped: bool,
|
|
finished: bool,
|
|
needs_close_after_finish: bool,
|
|
}
|
|
|
|
pub struct HttpRecord(RefCell<Option<HttpRecordInner>>);
|
|
|
|
#[cfg(feature = "__http_tracing")]
|
|
impl Drop for HttpRecord {
|
|
fn drop(&mut self) {
|
|
RECORD_COUNT
|
|
.fetch_sub(1, std::sync::atomic::Ordering::SeqCst)
|
|
.checked_sub(1)
|
|
.expect("Count went below zero");
|
|
http_general_trace!("HttpRecord::drop");
|
|
}
|
|
}
|
|
|
|
impl HttpRecord {
|
|
fn new(
|
|
request: Request,
|
|
request_info: HttpConnectionProperties,
|
|
server_state: SignallingRc<HttpServerState>,
|
|
) -> Rc<Self> {
|
|
let (request_parts, request_body) = request.into_parts();
|
|
let request_body = Some(request_body.into());
|
|
let (mut response_parts, _) = http_1::Response::new(()).into_parts();
|
|
let record =
|
|
if let Some((record, headers)) = server_state.borrow_mut().pool.pop() {
|
|
response_parts.headers = headers;
|
|
http_trace!(record, "HttpRecord::reuse");
|
|
record
|
|
} else {
|
|
#[cfg(feature = "__http_tracing")]
|
|
{
|
|
RECORD_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
}
|
|
|
|
#[allow(clippy::let_and_return)]
|
|
let record = Rc::new(Self(RefCell::new(None)));
|
|
http_trace!(record, "HttpRecord::new");
|
|
record
|
|
};
|
|
*record.0.borrow_mut() = Some(HttpRecordInner {
|
|
server_state,
|
|
request_info,
|
|
request_parts,
|
|
request_body,
|
|
response_parts: Some(response_parts),
|
|
response_ready: false,
|
|
response_waker: None,
|
|
response_body: ResponseBytesInner::Empty,
|
|
response_body_finished: false,
|
|
response_body_waker: None,
|
|
trailers: None,
|
|
been_dropped: false,
|
|
finished: false,
|
|
needs_close_after_finish: false,
|
|
});
|
|
record
|
|
}
|
|
|
|
fn finish(self: Rc<Self>) {
|
|
http_trace!(self, "HttpRecord::finish");
|
|
let mut inner = self.self_mut();
|
|
inner.response_body_finished = true;
|
|
let response_body_waker = inner.response_body_waker.take();
|
|
let needs_close_after_finish = inner.needs_close_after_finish;
|
|
drop(inner);
|
|
if let Some(waker) = response_body_waker {
|
|
waker.wake();
|
|
}
|
|
if !needs_close_after_finish {
|
|
self.recycle();
|
|
}
|
|
}
|
|
|
|
pub fn close_after_finish(self: Rc<Self>) {
|
|
debug_assert!(self.self_ref().needs_close_after_finish);
|
|
let mut inner = self.self_mut();
|
|
inner.needs_close_after_finish = false;
|
|
if !inner.finished {
|
|
drop(inner);
|
|
self.recycle();
|
|
}
|
|
}
|
|
|
|
pub fn needs_close_after_finish(&self) -> RefMut<'_, bool> {
|
|
RefMut::map(self.self_mut(), |inner| &mut inner.needs_close_after_finish)
|
|
}
|
|
|
|
fn recycle(self: Rc<Self>) {
|
|
assert!(
|
|
Rc::strong_count(&self) == 1,
|
|
"HTTP state error: Expected to be last strong reference"
|
|
);
|
|
let HttpRecordInner {
|
|
server_state,
|
|
request_parts: Parts { mut headers, .. },
|
|
..
|
|
} = self.0.borrow_mut().take().unwrap();
|
|
|
|
let inflight = server_state.strong_count();
|
|
http_trace!(self, "HttpRecord::recycle inflight={}", inflight);
|
|
|
|
// Keep a buffer of allocations on hand to be reused by incoming requests.
|
|
// Estimated target size is 16 + 1/8 the number of inflight requests.
|
|
let target = 16 + (inflight >> 3);
|
|
let pool = &mut server_state.borrow_mut().pool;
|
|
if target > pool.len() {
|
|
headers.clear();
|
|
pool.push((self, headers));
|
|
} else if target < pool.len() - 8 {
|
|
pool.truncate(target);
|
|
}
|
|
}
|
|
|
|
fn self_ref(&self) -> Ref<'_, HttpRecordInner> {
|
|
Ref::map(self.0.borrow(), |option| option.as_ref().unwrap())
|
|
}
|
|
|
|
fn self_mut(&self) -> RefMut<'_, HttpRecordInner> {
|
|
RefMut::map(self.0.borrow_mut(), |option| option.as_mut().unwrap())
|
|
}
|
|
|
|
/// Perform the Hyper upgrade on this record.
|
|
pub fn upgrade(&self) -> Result<OnUpgrade, AnyError> {
|
|
// Manually perform the upgrade. We're peeking into hyper's underlying machinery here a bit
|
|
self
|
|
.self_mut()
|
|
.request_parts
|
|
.extensions
|
|
.remove::<OnUpgrade>()
|
|
.ok_or_else(|| AnyError::msg("upgrade unavailable"))
|
|
}
|
|
|
|
/// Take the Hyper body from this record.
|
|
pub fn take_request_body(&self) -> Option<Incoming> {
|
|
let body_holder = &mut self.self_mut().request_body;
|
|
let body = body_holder.take();
|
|
match body {
|
|
Some(RequestBodyState::Incoming(body)) => Some(body),
|
|
x => {
|
|
*body_holder = x;
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Replace the request body with a resource ID and the OpState we'll need to shut it down.
|
|
/// We cannot keep just the resource itself, as JS code might be reading from the resource ID
|
|
/// to generate the response data (requiring us to keep it in the resource table).
|
|
pub fn put_resource(&self, res: HttpRequestBodyAutocloser) {
|
|
self.self_mut().request_body = Some(RequestBodyState::Resource(res));
|
|
}
|
|
|
|
/// Cleanup resources not needed after the future is dropped.
|
|
fn cancel(self: Rc<Self>) {
|
|
http_trace!(self, "HttpRecord::cancel");
|
|
let mut inner = self.self_mut();
|
|
if inner.response_ready {
|
|
// Future dropped between wake() and async fn resuming.
|
|
drop(inner);
|
|
self.finish();
|
|
return;
|
|
}
|
|
inner.been_dropped = true;
|
|
// The request body might include actual resources.
|
|
inner.request_body.take();
|
|
}
|
|
|
|
/// Complete this record, potentially expunging it if it is fully complete (ie: cancelled as well).
|
|
pub fn complete(self: Rc<Self>) {
|
|
http_trace!(self, "HttpRecord::complete");
|
|
let mut inner = self.self_mut();
|
|
assert!(
|
|
!inner.response_ready,
|
|
"HTTP state error: Entry has already been completed"
|
|
);
|
|
if inner.been_dropped {
|
|
drop(inner);
|
|
self.finish();
|
|
return;
|
|
}
|
|
inner.response_ready = true;
|
|
if let Some(waker) = inner.response_waker.take() {
|
|
drop(inner);
|
|
waker.wake();
|
|
}
|
|
}
|
|
|
|
fn take_response_body(&self) -> ResponseBytesInner {
|
|
let mut inner = self.self_mut();
|
|
debug_assert!(
|
|
!matches!(inner.response_body, ResponseBytesInner::Done),
|
|
"HTTP state error: response body already complete"
|
|
);
|
|
std::mem::replace(&mut inner.response_body, ResponseBytesInner::Done)
|
|
}
|
|
|
|
/// Has the future for this record been dropped? ie, has the underlying TCP connection
|
|
/// been closed?
|
|
pub fn cancelled(&self) -> bool {
|
|
self.self_ref().been_dropped
|
|
}
|
|
|
|
/// Get a mutable reference to the response status and headers.
|
|
pub fn response_parts(&self) -> RefMut<'_, http_1::response::Parts> {
|
|
RefMut::map(self.self_mut(), |inner| {
|
|
inner.response_parts.as_mut().unwrap()
|
|
})
|
|
}
|
|
|
|
/// Get a mutable reference to the trailers.
|
|
pub fn trailers(&self) -> RefMut<'_, Option<HeaderMap>> {
|
|
RefMut::map(self.self_mut(), |inner| &mut inner.trailers)
|
|
}
|
|
|
|
pub fn set_response_body(&self, response_body: ResponseBytesInner) {
|
|
let mut inner = self.self_mut();
|
|
debug_assert!(matches!(inner.response_body, ResponseBytesInner::Empty));
|
|
inner.response_body = response_body;
|
|
}
|
|
|
|
/// Take the response.
|
|
fn into_response(self: Rc<Self>) -> Response {
|
|
let parts = self.self_mut().response_parts.take().unwrap();
|
|
let body = HttpRecordResponse(ManuallyDrop::new(self));
|
|
Response::from_parts(parts, body)
|
|
}
|
|
|
|
/// Get a reference to the connection properties.
|
|
pub fn request_info(&self) -> Ref<'_, HttpConnectionProperties> {
|
|
Ref::map(self.self_ref(), |inner| &inner.request_info)
|
|
}
|
|
|
|
/// Get a reference to the request parts.
|
|
pub fn request_parts(&self) -> Ref<'_, Parts> {
|
|
Ref::map(self.self_ref(), |inner| &inner.request_parts)
|
|
}
|
|
|
|
/// Resolves when response head is ready.
|
|
fn response_ready(&self) -> impl Future<Output = ()> + '_ {
|
|
struct HttpRecordReady<'a>(&'a HttpRecord);
|
|
|
|
impl<'a> Future for HttpRecordReady<'a> {
|
|
type Output = ();
|
|
|
|
fn poll(
|
|
self: Pin<&mut Self>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Self::Output> {
|
|
let mut mut_self = self.0.self_mut();
|
|
if mut_self.response_ready {
|
|
return Poll::Ready(());
|
|
}
|
|
mut_self.response_waker = Some(cx.waker().clone());
|
|
Poll::Pending
|
|
}
|
|
}
|
|
|
|
HttpRecordReady(self)
|
|
}
|
|
|
|
/// Resolves when response body has finished streaming.
|
|
pub fn response_body_finished(&self) -> impl Future<Output = ()> + '_ {
|
|
struct HttpRecordFinished<'a>(&'a HttpRecord);
|
|
|
|
impl<'a> Future for HttpRecordFinished<'a> {
|
|
type Output = ();
|
|
|
|
fn poll(
|
|
self: Pin<&mut Self>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Self::Output> {
|
|
let mut mut_self = self.0.self_mut();
|
|
if mut_self.response_body_finished {
|
|
return Poll::Ready(());
|
|
}
|
|
mut_self.response_body_waker = Some(cx.waker().clone());
|
|
Poll::Pending
|
|
}
|
|
}
|
|
|
|
HttpRecordFinished(self)
|
|
}
|
|
}
|
|
|
|
#[repr(transparent)]
|
|
pub struct HttpRecordResponse(ManuallyDrop<Rc<HttpRecord>>);
|
|
|
|
impl Body for HttpRecordResponse {
|
|
type Data = BufView;
|
|
type Error = AnyError;
|
|
|
|
fn poll_frame(
|
|
self: Pin<&mut Self>,
|
|
cx: &mut Context<'_>,
|
|
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
|
|
use crate::response_body::PollFrame;
|
|
let record = &self.0;
|
|
|
|
let res = loop {
|
|
let mut inner = record.self_mut();
|
|
let res = match &mut inner.response_body {
|
|
ResponseBytesInner::Done | ResponseBytesInner::Empty => {
|
|
if let Some(trailers) = inner.trailers.take() {
|
|
return Poll::Ready(Some(Ok(Frame::trailers(trailers))));
|
|
}
|
|
unreachable!()
|
|
}
|
|
ResponseBytesInner::Bytes(..) => {
|
|
drop(inner);
|
|
let ResponseBytesInner::Bytes(data) = record.take_response_body()
|
|
else {
|
|
unreachable!();
|
|
};
|
|
return Poll::Ready(Some(Ok(Frame::data(data))));
|
|
}
|
|
ResponseBytesInner::UncompressedStream(stm) => {
|
|
ready!(Pin::new(stm).poll_frame(cx))
|
|
}
|
|
ResponseBytesInner::GZipStream(stm) => {
|
|
ready!(Pin::new(stm).poll_frame(cx))
|
|
}
|
|
ResponseBytesInner::BrotliStream(stm) => {
|
|
ready!(Pin::new(stm).poll_frame(cx))
|
|
}
|
|
};
|
|
// This is where we retry the NoData response
|
|
if matches!(res, ResponseStreamResult::NoData) {
|
|
continue;
|
|
}
|
|
break res;
|
|
};
|
|
|
|
if matches!(res, ResponseStreamResult::EndOfStream) {
|
|
if let Some(trailers) = record.self_mut().trailers.take() {
|
|
return Poll::Ready(Some(Ok(Frame::trailers(trailers))));
|
|
}
|
|
record.take_response_body();
|
|
}
|
|
Poll::Ready(res.into())
|
|
}
|
|
|
|
fn is_end_stream(&self) -> bool {
|
|
let inner = self.0.self_ref();
|
|
matches!(
|
|
inner.response_body,
|
|
ResponseBytesInner::Done | ResponseBytesInner::Empty
|
|
) && inner.trailers.is_none()
|
|
}
|
|
|
|
fn size_hint(&self) -> SizeHint {
|
|
// The size hint currently only used in the case where it is exact bounds in hyper, but we'll pass it through
|
|
// anyways just in case hyper needs it.
|
|
self.0.self_ref().response_body.size_hint()
|
|
}
|
|
}
|
|
|
|
impl Drop for HttpRecordResponse {
|
|
fn drop(&mut self) {
|
|
// SAFETY: this ManuallyDrop is not used again.
|
|
let record = unsafe { ManuallyDrop::take(&mut self.0) };
|
|
http_trace!(record, "HttpRecordResponse::drop");
|
|
record.finish();
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::response_body::Compression;
|
|
use crate::response_body::ResponseBytesInner;
|
|
use bytes::Buf;
|
|
use deno_net::raw::NetworkStreamType;
|
|
use hyper1::body::Body;
|
|
use hyper1::service::service_fn;
|
|
use hyper1::service::HttpService;
|
|
use hyper_util::rt::TokioIo;
|
|
use std::error::Error as StdError;
|
|
|
|
/// Execute client request on service and concurrently map the response.
|
|
async fn serve_request<B, S, T, F>(
|
|
req: http_1::Request<B>,
|
|
service: S,
|
|
map_response: impl FnOnce(hyper1::Response<Incoming>) -> F,
|
|
) -> hyper1::Result<T>
|
|
where
|
|
B: Body + Send + 'static, // Send bound due to DuplexStream
|
|
B::Data: Send,
|
|
B::Error: Into<Box<dyn StdError + Send + Sync>>,
|
|
S: HttpService<Incoming>,
|
|
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
|
S::ResBody: 'static,
|
|
<S::ResBody as Body>::Error: Into<Box<dyn StdError + Send + Sync>>,
|
|
F: std::future::Future<Output = hyper1::Result<T>>,
|
|
{
|
|
use hyper1::client::conn::http1::handshake;
|
|
use hyper1::server::conn::http1::Builder;
|
|
let (stream_client, stream_server) = tokio::io::duplex(16 * 1024);
|
|
let conn_server =
|
|
Builder::new().serve_connection(TokioIo::new(stream_server), service);
|
|
let (mut sender, conn_client) =
|
|
handshake(TokioIo::new(stream_client)).await?;
|
|
|
|
let (res, _, _) = tokio::try_join!(
|
|
async move {
|
|
let res = sender.send_request(req).await?;
|
|
map_response(res).await
|
|
},
|
|
conn_server,
|
|
conn_client,
|
|
)?;
|
|
Ok(res)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_handle_request() -> Result<(), AnyError> {
|
|
let (tx, mut rx) = tokio::sync::mpsc::channel(10);
|
|
let server_state = HttpServerState::new();
|
|
let server_state_check = server_state.clone();
|
|
let request_info = HttpConnectionProperties {
|
|
peer_address: "".into(),
|
|
peer_port: None,
|
|
local_port: None,
|
|
stream_type: NetworkStreamType::Tcp,
|
|
};
|
|
let svc = service_fn(move |req: hyper1::Request<Incoming>| {
|
|
handle_request(
|
|
req,
|
|
request_info.clone(),
|
|
server_state.clone(),
|
|
tx.clone(),
|
|
)
|
|
});
|
|
|
|
let client_req =
|
|
http_1::Request::builder().uri("/").body("".to_string())?;
|
|
|
|
// Response produced by concurrent tasks
|
|
tokio::try_join!(
|
|
async move {
|
|
// JavaScript handler produces response
|
|
let record = rx.recv().await.unwrap();
|
|
record.set_response_body(ResponseBytesInner::from_vec(
|
|
Compression::None,
|
|
b"hello world".to_vec(),
|
|
));
|
|
record.complete();
|
|
Ok(())
|
|
},
|
|
// Server connection executes service
|
|
async move {
|
|
serve_request(client_req, svc, |res| async {
|
|
// Client reads the response
|
|
use http_body_util::BodyExt;
|
|
assert_eq!(res.status(), 200);
|
|
let body = res.collect().await?.to_bytes();
|
|
assert_eq!(body.chunk(), b"hello world");
|
|
Ok(())
|
|
})
|
|
.await
|
|
},
|
|
)?;
|
|
assert_eq!(server_state_check.strong_count(), 1);
|
|
Ok(())
|
|
}
|
|
}
|