mirror of
https://github.com/denoland/deno.git
synced 2024-12-11 01:58:05 -05:00
e2761df3fe
Fix internal "upgradeHttpRaw" API restoring capability to upgrade HTTP connection in polyfilles "node:http" API.
960 lines
26 KiB
Rust
960 lines
26 KiB
Rust
// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
|
|
use crate::extract_network_stream;
|
|
use crate::network_buffered_stream::NetworkStreamPrefixCheck;
|
|
use crate::request_body::HttpRequestBody;
|
|
use crate::request_properties::DefaultHttpRequestProperties;
|
|
use crate::request_properties::HttpConnectionProperties;
|
|
use crate::request_properties::HttpListenProperties;
|
|
use crate::request_properties::HttpPropertyExtractor;
|
|
use crate::response_body::CompletionHandle;
|
|
use crate::response_body::ResponseBytes;
|
|
use crate::response_body::ResponseBytesInner;
|
|
use crate::response_body::V8StreamHttpResponseBody;
|
|
use crate::websocket_upgrade::WebSocketUpgrade;
|
|
use crate::LocalExecutor;
|
|
use deno_core::error::AnyError;
|
|
use deno_core::futures::TryFutureExt;
|
|
use deno_core::op;
|
|
use deno_core::AsyncRefCell;
|
|
use deno_core::AsyncResult;
|
|
use deno_core::BufView;
|
|
use deno_core::ByteString;
|
|
use deno_core::CancelFuture;
|
|
use deno_core::CancelHandle;
|
|
use deno_core::CancelTryFuture;
|
|
use deno_core::OpState;
|
|
use deno_core::RcRef;
|
|
use deno_core::Resource;
|
|
use deno_core::ResourceId;
|
|
use deno_core::ZeroCopyBuf;
|
|
use deno_net::ops_tls::TlsStream;
|
|
use deno_net::raw::put_network_stream_resource;
|
|
use deno_net::raw::NetworkStream;
|
|
use deno_net::raw::NetworkStreamAddress;
|
|
use http::request::Parts;
|
|
use hyper1::body::Incoming;
|
|
use hyper1::header::COOKIE;
|
|
use hyper1::http::HeaderName;
|
|
use hyper1::http::HeaderValue;
|
|
use hyper1::server::conn::http1;
|
|
use hyper1::server::conn::http2;
|
|
use hyper1::service::service_fn;
|
|
use hyper1::service::HttpService;
|
|
use hyper1::upgrade::OnUpgrade;
|
|
|
|
use hyper1::StatusCode;
|
|
use pin_project::pin_project;
|
|
use pin_project::pinned_drop;
|
|
use slab::Slab;
|
|
use std::borrow::Cow;
|
|
use std::cell::RefCell;
|
|
use std::future::Future;
|
|
use std::io;
|
|
use std::net::Ipv4Addr;
|
|
use std::net::SocketAddr;
|
|
use std::net::SocketAddrV4;
|
|
use std::pin::Pin;
|
|
use std::rc::Rc;
|
|
|
|
use tokio::io::AsyncReadExt;
|
|
use tokio::io::AsyncWriteExt;
|
|
|
|
use tokio::task::spawn_local;
|
|
use tokio::task::JoinHandle;
|
|
|
|
type Request = hyper1::Request<Incoming>;
|
|
type Response = hyper1::Response<ResponseBytes>;
|
|
|
|
/// All HTTP/2 connections start with this byte string.
|
|
///
|
|
/// In HTTP/2, each endpoint is required to send a connection preface as a final confirmation
|
|
/// of the protocol in use and to establish the initial settings for the HTTP/2 connection. The
|
|
/// client and server each send a different connection preface.
|
|
///
|
|
/// The client connection preface starts with a sequence of 24 octets, which in hex notation is:
|
|
///
|
|
/// 0x505249202a20485454502f322e300d0a0d0a534d0d0a0d0a
|
|
///
|
|
/// That is, the connection preface starts with the string PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n). This sequence
|
|
/// MUST be followed by a SETTINGS frame (Section 6.5), which MAY be empty.
|
|
const HTTP2_PREFIX: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
|
|
|
|
/// ALPN negotation for "h2"
|
|
const TLS_ALPN_HTTP_2: &[u8] = b"h2";
|
|
|
|
/// ALPN negotation for "http/1.1"
|
|
const TLS_ALPN_HTTP_11: &[u8] = b"http/1.1";
|
|
|
|
/// Name a trait for streams we can serve HTTP over.
|
|
trait HttpServeStream:
|
|
tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static
|
|
{
|
|
}
|
|
impl<
|
|
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
|
|
> HttpServeStream for S
|
|
{
|
|
}
|
|
|
|
pub struct HttpSlabRecord {
|
|
request_info: HttpConnectionProperties,
|
|
request_parts: Parts,
|
|
request_body: Option<Incoming>,
|
|
// The response may get taken before we tear this down
|
|
response: Option<Response>,
|
|
body: Option<Rc<HttpRequestBody>>,
|
|
promise: CompletionHandle,
|
|
#[cfg(__zombie_http_tracking)]
|
|
alive: bool,
|
|
}
|
|
|
|
thread_local! {
|
|
pub static SLAB: RefCell<Slab<HttpSlabRecord>> = RefCell::new(Slab::with_capacity(1024));
|
|
}
|
|
|
|
/// Generates getters and setters for the [`SLAB`]. For example,
|
|
/// `with!(with_req, with_req_mut, Parts, http, http.request_parts);` expands to:
|
|
///
|
|
/// ```ignore
|
|
/// #[inline(always)]
|
|
/// #[allow(dead_code)]
|
|
/// pub(crate) fn with_req_mut<T>(key: usize, f: impl FnOnce(&mut Parts) -> T) -> T {
|
|
/// SLAB.with(|slab| {
|
|
/// let mut borrow = slab.borrow_mut();
|
|
/// let mut http = borrow.get_mut(key).unwrap();
|
|
/// #[cfg(__zombie_http_tracking)]
|
|
/// if !http.alive {
|
|
/// panic!("Attempted to access a dead HTTP object")
|
|
/// }
|
|
/// f(&mut http.expr)
|
|
/// })
|
|
/// }
|
|
|
|
/// #[inline(always)]
|
|
/// #[allow(dead_code)]
|
|
/// pub(crate) fn with_req<T>(key: usize, f: impl FnOnce(&Parts) -> T) -> T {
|
|
/// SLAB.with(|slab| {
|
|
/// let mut borrow = slab.borrow();
|
|
/// let mut http = borrow.get(key).unwrap();
|
|
/// #[cfg(__zombie_http_tracking)]
|
|
/// if !http.alive {
|
|
/// panic!("Attempted to access a dead HTTP object")
|
|
/// }
|
|
/// f(&http.expr)
|
|
/// })
|
|
/// }
|
|
/// ```
|
|
macro_rules! with {
|
|
($ref:ident, $mut:ident, $type:ty, $http:ident, $expr:expr) => {
|
|
#[inline(always)]
|
|
#[allow(dead_code)]
|
|
pub(crate) fn $mut<T>(key: u32, f: impl FnOnce(&mut $type) -> T) -> T {
|
|
SLAB.with(|slab| {
|
|
let mut borrow = slab.borrow_mut();
|
|
#[allow(unused_mut)] // TODO(mmastrac): compiler issue?
|
|
let mut $http = match borrow.get_mut(key as usize) {
|
|
Some(http) => http,
|
|
None => panic!(
|
|
"Attemped to access invalid request {} ({} in total available)",
|
|
key,
|
|
borrow.len()
|
|
),
|
|
};
|
|
#[cfg(__zombie_http_tracking)]
|
|
if !$http.alive {
|
|
panic!("Attempted to access a dead HTTP object")
|
|
}
|
|
f(&mut $expr)
|
|
})
|
|
}
|
|
|
|
#[inline(always)]
|
|
#[allow(dead_code)]
|
|
pub(crate) fn $ref<T>(key: u32, f: impl FnOnce(&$type) -> T) -> T {
|
|
SLAB.with(|slab| {
|
|
let borrow = slab.borrow();
|
|
let $http = borrow.get(key as usize).unwrap();
|
|
#[cfg(__zombie_http_tracking)]
|
|
if !$http.alive {
|
|
panic!("Attempted to access a dead HTTP object")
|
|
}
|
|
f(&$expr)
|
|
})
|
|
}
|
|
};
|
|
}
|
|
|
|
with!(with_req, with_req_mut, Parts, http, http.request_parts);
|
|
with!(
|
|
with_req_body,
|
|
with_req_body_mut,
|
|
Option<Incoming>,
|
|
http,
|
|
http.request_body
|
|
);
|
|
with!(
|
|
with_resp,
|
|
with_resp_mut,
|
|
Option<Response>,
|
|
http,
|
|
http.response
|
|
);
|
|
with!(
|
|
with_body,
|
|
with_body_mut,
|
|
Option<Rc<HttpRequestBody>>,
|
|
http,
|
|
http.body
|
|
);
|
|
with!(
|
|
with_promise,
|
|
with_promise_mut,
|
|
CompletionHandle,
|
|
http,
|
|
http.promise
|
|
);
|
|
with!(with_http, with_http_mut, HttpSlabRecord, http, http);
|
|
|
|
fn slab_insert(
|
|
request: Request,
|
|
request_info: HttpConnectionProperties,
|
|
) -> u32 {
|
|
SLAB.with(|slab| {
|
|
let (request_parts, request_body) = request.into_parts();
|
|
slab.borrow_mut().insert(HttpSlabRecord {
|
|
request_info,
|
|
request_parts,
|
|
request_body: Some(request_body),
|
|
response: Some(Response::new(ResponseBytes::default())),
|
|
body: None,
|
|
promise: CompletionHandle::default(),
|
|
#[cfg(__zombie_http_tracking)]
|
|
alive: true,
|
|
})
|
|
}) as u32
|
|
}
|
|
|
|
#[op]
|
|
pub fn op_upgrade_raw(
|
|
state: &mut OpState,
|
|
index: u32,
|
|
) -> Result<ResourceId, AnyError> {
|
|
// Stage 1: extract the upgrade future
|
|
let upgrade = with_http_mut(index, |http| {
|
|
// Manually perform the upgrade. We're peeking into hyper's underlying machinery here a bit
|
|
http
|
|
.request_parts
|
|
.extensions
|
|
.remove::<OnUpgrade>()
|
|
.ok_or_else(|| AnyError::msg("upgrade unavailable"))
|
|
})?;
|
|
|
|
let (read, write) = tokio::io::duplex(1024);
|
|
let (read_rx, write_tx) = tokio::io::split(read);
|
|
let (mut write_rx, mut read_tx) = tokio::io::split(write);
|
|
|
|
spawn_local(async move {
|
|
let mut upgrade_stream = WebSocketUpgrade::<ResponseBytes>::default();
|
|
|
|
// Stage 2: Extract the Upgraded connection
|
|
let mut buf = [0; 1024];
|
|
let upgraded = loop {
|
|
let read = Pin::new(&mut write_rx).read(&mut buf).await?;
|
|
match upgrade_stream.write(&buf[..read]) {
|
|
Ok(None) => continue,
|
|
Ok(Some((response, bytes))) => {
|
|
with_resp_mut(index, |resp| *resp = Some(response));
|
|
with_promise_mut(index, |promise| promise.complete(true));
|
|
let mut upgraded = upgrade.await?;
|
|
upgraded.write_all(&bytes).await?;
|
|
break upgraded;
|
|
}
|
|
Err(err) => return Err(err),
|
|
}
|
|
};
|
|
|
|
// Stage 3: Pump the data
|
|
let (mut upgraded_rx, mut upgraded_tx) = tokio::io::split(upgraded);
|
|
|
|
spawn_local(async move {
|
|
let mut buf = [0; 1024];
|
|
loop {
|
|
let read = upgraded_rx.read(&mut buf).await?;
|
|
if read == 0 {
|
|
break;
|
|
}
|
|
read_tx.write_all(&buf[..read]).await?;
|
|
}
|
|
Ok::<_, AnyError>(())
|
|
});
|
|
spawn_local(async move {
|
|
let mut buf = [0; 1024];
|
|
loop {
|
|
let read = write_rx.read(&mut buf).await?;
|
|
if read == 0 {
|
|
break;
|
|
}
|
|
upgraded_tx.write_all(&buf[..read]).await?;
|
|
}
|
|
Ok::<_, AnyError>(())
|
|
});
|
|
|
|
Ok(())
|
|
});
|
|
|
|
Ok(
|
|
state
|
|
.resource_table
|
|
.add(UpgradeStream::new(read_rx, write_tx)),
|
|
)
|
|
}
|
|
|
|
#[op]
|
|
pub async fn op_upgrade(
|
|
state: Rc<RefCell<OpState>>,
|
|
index: u32,
|
|
headers: Vec<(ByteString, ByteString)>,
|
|
) -> Result<(ResourceId, ZeroCopyBuf), AnyError> {
|
|
// Stage 1: set the respnse to 101 Switching Protocols and send it
|
|
let upgrade = with_http_mut(index, |http| {
|
|
// Manually perform the upgrade. We're peeking into hyper's underlying machinery here a bit
|
|
let upgrade = http
|
|
.request_parts
|
|
.extensions
|
|
.remove::<OnUpgrade>()
|
|
.ok_or_else(|| AnyError::msg("upgrade unavailable"))?;
|
|
|
|
let response = http.response.as_mut().unwrap();
|
|
*response.status_mut() = StatusCode::SWITCHING_PROTOCOLS;
|
|
for (name, value) in headers {
|
|
response.headers_mut().append(
|
|
HeaderName::from_bytes(&name).unwrap(),
|
|
HeaderValue::from_bytes(&value).unwrap(),
|
|
);
|
|
}
|
|
http.promise.complete(true);
|
|
Ok::<_, AnyError>(upgrade)
|
|
})?;
|
|
|
|
// Stage 2: wait for the request to finish upgrading
|
|
let upgraded = upgrade.await?;
|
|
|
|
// Stage 3: return the extracted raw network stream
|
|
let (stream, bytes) = extract_network_stream(upgraded);
|
|
|
|
// We're allocating for those extra bytes, but they are probably going to be empty most of the time
|
|
Ok((
|
|
put_network_stream_resource(
|
|
&mut state.borrow_mut().resource_table,
|
|
stream,
|
|
)?,
|
|
ZeroCopyBuf::from(bytes.to_vec()),
|
|
))
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_promise_complete(index: u32, status: u16) {
|
|
with_resp_mut(index, |resp| {
|
|
// The Javascript code will never provide a status that is invalid here (see 23_response.js)
|
|
*resp.as_mut().unwrap().status_mut() =
|
|
StatusCode::from_u16(status).unwrap();
|
|
});
|
|
with_promise_mut(index, |promise| {
|
|
promise.complete(true);
|
|
});
|
|
}
|
|
|
|
#[op]
|
|
pub fn op_get_request_method_and_url(
|
|
index: u32,
|
|
) -> (String, Option<String>, String, String, Option<u16>) {
|
|
// TODO(mmastrac): Passing method can be optimized
|
|
with_http(index, |http| {
|
|
let request_properties = DefaultHttpRequestProperties::request_properties(
|
|
&http.request_info,
|
|
&http.request_parts.uri,
|
|
&http.request_parts.headers,
|
|
);
|
|
|
|
// Only extract the path part - we handle authority elsewhere
|
|
let path = match &http.request_parts.uri.path_and_query() {
|
|
Some(path_and_query) => path_and_query.to_string(),
|
|
None => "".to_owned(),
|
|
};
|
|
|
|
(
|
|
http.request_parts.method.as_str().to_owned(),
|
|
request_properties.authority,
|
|
path,
|
|
String::from(http.request_info.peer_address.as_ref()),
|
|
http.request_info.peer_port,
|
|
)
|
|
})
|
|
}
|
|
|
|
#[op]
|
|
pub fn op_get_request_header(index: u32, name: String) -> Option<ByteString> {
|
|
with_req(index, |req| {
|
|
let value = req.headers.get(name);
|
|
value.map(|value| value.as_bytes().into())
|
|
})
|
|
}
|
|
|
|
#[op]
|
|
pub fn op_get_request_headers(index: u32) -> Vec<(ByteString, ByteString)> {
|
|
with_req(index, |req| {
|
|
let headers = &req.headers;
|
|
let mut vec = Vec::with_capacity(headers.len());
|
|
let mut cookies: Option<Vec<&[u8]>> = None;
|
|
for (name, value) in headers {
|
|
if name == COOKIE {
|
|
if let Some(ref mut cookies) = cookies {
|
|
cookies.push(value.as_bytes());
|
|
} else {
|
|
cookies = Some(vec![value.as_bytes()]);
|
|
}
|
|
} else {
|
|
let name: &[u8] = name.as_ref();
|
|
vec.push((name.into(), value.as_bytes().into()))
|
|
}
|
|
}
|
|
|
|
// We treat cookies specially, because we don't want them to get them
|
|
// mangled by the `Headers` object in JS. What we do is take all cookie
|
|
// headers and concat them into a single cookie header, separated by
|
|
// semicolons.
|
|
// TODO(mmastrac): This should probably happen on the JS side on-demand
|
|
if let Some(cookies) = cookies {
|
|
let cookie_sep = "; ".as_bytes();
|
|
vec.push((
|
|
ByteString::from(COOKIE.as_str()),
|
|
ByteString::from(cookies.join(cookie_sep)),
|
|
));
|
|
}
|
|
vec
|
|
})
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_read_request_body(state: &mut OpState, index: u32) -> ResourceId {
|
|
let incoming = with_req_body_mut(index, |body| body.take().unwrap());
|
|
let body_resource = Rc::new(HttpRequestBody::new(incoming));
|
|
let res = state.resource_table.add_rc(body_resource.clone());
|
|
with_body_mut(index, |body| {
|
|
*body = Some(body_resource);
|
|
});
|
|
res
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_response_header(index: u32, name: &str, value: &str) {
|
|
with_resp_mut(index, |resp| {
|
|
let resp_headers = resp.as_mut().unwrap().headers_mut();
|
|
// These are valid latin-1 strings
|
|
let name = HeaderName::from_bytes(name.as_bytes()).unwrap();
|
|
let value = HeaderValue::from_bytes(value.as_bytes()).unwrap();
|
|
resp_headers.append(name, value);
|
|
});
|
|
}
|
|
|
|
#[op]
|
|
pub fn op_set_response_headers(
|
|
index: u32,
|
|
headers: Vec<(ByteString, ByteString)>,
|
|
) {
|
|
// TODO(mmastrac): Invalid headers should be handled?
|
|
with_resp_mut(index, |resp| {
|
|
let resp_headers = resp.as_mut().unwrap().headers_mut();
|
|
resp_headers.reserve(headers.len());
|
|
for (name, value) in headers {
|
|
// These are valid latin-1 strings
|
|
let name = HeaderName::from_bytes(&name).unwrap();
|
|
let value = HeaderValue::from_bytes(&value).unwrap();
|
|
resp_headers.append(name, value);
|
|
}
|
|
})
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_response_body_resource(
|
|
state: &mut OpState,
|
|
index: u32,
|
|
stream_rid: ResourceId,
|
|
auto_close: bool,
|
|
) -> Result<(), AnyError> {
|
|
// If the stream is auto_close, we will hold the last ref to it until the response is complete.
|
|
let resource = if auto_close {
|
|
state.resource_table.take_any(stream_rid)?
|
|
} else {
|
|
state.resource_table.get_any(stream_rid)?
|
|
};
|
|
|
|
with_resp_mut(index, move |response| {
|
|
let future = resource.clone().read(64 * 1024);
|
|
response
|
|
.as_mut()
|
|
.unwrap()
|
|
.body_mut()
|
|
.initialize(ResponseBytesInner::Resource(auto_close, resource, future));
|
|
});
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_response_body_stream(
|
|
state: &mut OpState,
|
|
index: u32,
|
|
) -> Result<ResourceId, AnyError> {
|
|
// TODO(mmastrac): what should this channel size be?
|
|
let (tx, rx) = tokio::sync::mpsc::channel(1);
|
|
let (tx, rx) = (
|
|
V8StreamHttpResponseBody::new(tx),
|
|
ResponseBytesInner::V8Stream(rx),
|
|
);
|
|
|
|
with_resp_mut(index, move |response| {
|
|
response.as_mut().unwrap().body_mut().initialize(rx);
|
|
});
|
|
|
|
Ok(state.resource_table.add(tx))
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_response_body_text(index: u32, text: String) {
|
|
if !text.is_empty() {
|
|
with_resp_mut(index, move |response| {
|
|
response
|
|
.as_mut()
|
|
.unwrap()
|
|
.body_mut()
|
|
.initialize(ResponseBytesInner::Bytes(BufView::from(text.into_bytes())))
|
|
});
|
|
}
|
|
}
|
|
|
|
#[op(fast)]
|
|
pub fn op_set_response_body_bytes(index: u32, buffer: &[u8]) {
|
|
if !buffer.is_empty() {
|
|
with_resp_mut(index, |response| {
|
|
response
|
|
.as_mut()
|
|
.unwrap()
|
|
.body_mut()
|
|
.initialize(ResponseBytesInner::Bytes(BufView::from(buffer.to_vec())))
|
|
});
|
|
};
|
|
}
|
|
|
|
#[op]
|
|
pub async fn op_http_track(
|
|
state: Rc<RefCell<OpState>>,
|
|
index: u32,
|
|
server_rid: ResourceId,
|
|
) -> Result<(), AnyError> {
|
|
let handle = with_resp(index, |resp| {
|
|
resp.as_ref().unwrap().body().completion_handle()
|
|
});
|
|
|
|
let join_handle = state
|
|
.borrow_mut()
|
|
.resource_table
|
|
.get::<HttpJoinHandle>(server_rid)?;
|
|
|
|
match handle.or_cancel(join_handle.cancel_handle()).await {
|
|
Ok(true) => Ok(()),
|
|
Ok(false) => {
|
|
Err(AnyError::msg("connection closed before message completed"))
|
|
}
|
|
Err(_e) => Ok(()),
|
|
}
|
|
}
|
|
|
|
#[pin_project(PinnedDrop)]
|
|
pub struct SlabFuture<F: Future<Output = ()>>(u32, #[pin] F);
|
|
|
|
pub fn new_slab_future(
|
|
request: Request,
|
|
request_info: HttpConnectionProperties,
|
|
tx: tokio::sync::mpsc::Sender<u32>,
|
|
) -> SlabFuture<impl Future<Output = ()>> {
|
|
let index = slab_insert(request, request_info);
|
|
let rx = with_promise(index, |promise| promise.clone());
|
|
SlabFuture(index, async move {
|
|
if tx.send(index).await.is_ok() {
|
|
// We only need to wait for completion if we aren't closed
|
|
rx.await;
|
|
}
|
|
})
|
|
}
|
|
|
|
impl<F: Future<Output = ()>> SlabFuture<F> {}
|
|
|
|
#[pinned_drop]
|
|
impl<F: Future<Output = ()>> PinnedDrop for SlabFuture<F> {
|
|
fn drop(self: Pin<&mut Self>) {
|
|
SLAB.with(|slab| {
|
|
#[cfg(__zombie_http_tracking)]
|
|
{
|
|
slab.borrow_mut().get_mut(self.0 as usize).unwrap().alive = false;
|
|
}
|
|
#[cfg(not(__zombie_http_tracking))]
|
|
{
|
|
slab.borrow_mut().remove(self.0 as usize);
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
impl<F: Future<Output = ()>> Future for SlabFuture<F> {
|
|
type Output = Result<Response, hyper::Error>;
|
|
|
|
fn poll(
|
|
self: Pin<&mut Self>,
|
|
cx: &mut std::task::Context<'_>,
|
|
) -> std::task::Poll<Self::Output> {
|
|
let index = self.0;
|
|
self
|
|
.project()
|
|
.1
|
|
.poll(cx)
|
|
.map(|_| Ok(with_resp_mut(index, |resp| resp.take().unwrap())))
|
|
}
|
|
}
|
|
|
|
fn serve_http11_unconditional(
|
|
io: impl HttpServeStream,
|
|
svc: impl HttpService<Incoming, ResBody = ResponseBytes> + 'static,
|
|
cancel: RcRef<CancelHandle>,
|
|
) -> impl Future<Output = Result<(), AnyError>> + 'static {
|
|
let conn = http1::Builder::new()
|
|
.keep_alive(true)
|
|
.serve_connection(io, svc);
|
|
|
|
conn
|
|
.with_upgrades()
|
|
.map_err(AnyError::from)
|
|
.try_or_cancel(cancel)
|
|
}
|
|
|
|
fn serve_http2_unconditional(
|
|
io: impl HttpServeStream,
|
|
svc: impl HttpService<Incoming, ResBody = ResponseBytes> + 'static,
|
|
cancel: RcRef<CancelHandle>,
|
|
) -> impl Future<Output = Result<(), AnyError>> + 'static {
|
|
let conn = http2::Builder::new(LocalExecutor).serve_connection(io, svc);
|
|
conn.map_err(AnyError::from).try_or_cancel(cancel)
|
|
}
|
|
|
|
async fn serve_http2_autodetect(
|
|
io: impl HttpServeStream,
|
|
svc: impl HttpService<Incoming, ResBody = ResponseBytes> + 'static,
|
|
cancel: RcRef<CancelHandle>,
|
|
) -> Result<(), AnyError> {
|
|
let prefix = NetworkStreamPrefixCheck::new(io, HTTP2_PREFIX);
|
|
let (matches, io) = prefix.match_prefix().await?;
|
|
if matches {
|
|
serve_http2_unconditional(io, svc, cancel).await
|
|
} else {
|
|
serve_http11_unconditional(io, svc, cancel).await
|
|
}
|
|
}
|
|
|
|
fn serve_https(
|
|
mut io: TlsStream,
|
|
request_info: HttpConnectionProperties,
|
|
cancel: RcRef<CancelHandle>,
|
|
tx: tokio::sync::mpsc::Sender<u32>,
|
|
) -> JoinHandle<Result<(), AnyError>> {
|
|
// TODO(mmastrac): This is faster if we can use tokio::spawn but then the send bounds get us
|
|
let svc = service_fn(move |req: Request| {
|
|
new_slab_future(req, request_info.clone(), tx.clone())
|
|
});
|
|
spawn_local(async {
|
|
io.handshake().await?;
|
|
// If the client specifically negotiates a protocol, we will use it. If not, we'll auto-detect
|
|
// based on the prefix bytes
|
|
let handshake = io.get_ref().1.alpn_protocol();
|
|
if handshake == Some(TLS_ALPN_HTTP_2) {
|
|
serve_http2_unconditional(io, svc, cancel).await
|
|
} else if handshake == Some(TLS_ALPN_HTTP_11) {
|
|
serve_http11_unconditional(io, svc, cancel).await
|
|
} else {
|
|
serve_http2_autodetect(io, svc, cancel).await
|
|
}
|
|
})
|
|
}
|
|
|
|
fn serve_http(
|
|
io: impl HttpServeStream,
|
|
request_info: HttpConnectionProperties,
|
|
cancel: RcRef<CancelHandle>,
|
|
tx: tokio::sync::mpsc::Sender<u32>,
|
|
) -> JoinHandle<Result<(), AnyError>> {
|
|
// TODO(mmastrac): This is faster if we can use tokio::spawn but then the send bounds get us
|
|
let svc = service_fn(move |req: Request| {
|
|
new_slab_future(req, request_info.clone(), tx.clone())
|
|
});
|
|
spawn_local(serve_http2_autodetect(io, svc, cancel))
|
|
}
|
|
|
|
fn serve_http_on(
|
|
network_stream: NetworkStream,
|
|
listen_properties: &HttpListenProperties,
|
|
cancel: RcRef<CancelHandle>,
|
|
tx: tokio::sync::mpsc::Sender<u32>,
|
|
) -> JoinHandle<Result<(), AnyError>> {
|
|
// We always want some sort of peer address. If we can't get one, just make up one.
|
|
let peer_address = network_stream.peer_address().unwrap_or_else(|_| {
|
|
NetworkStreamAddress::Ip(SocketAddr::V4(SocketAddrV4::new(
|
|
Ipv4Addr::new(0, 0, 0, 0),
|
|
0,
|
|
)))
|
|
});
|
|
let connection_properties: HttpConnectionProperties =
|
|
DefaultHttpRequestProperties::connection_properties(
|
|
listen_properties,
|
|
&peer_address,
|
|
);
|
|
|
|
match network_stream {
|
|
NetworkStream::Tcp(conn) => {
|
|
serve_http(conn, connection_properties, cancel, tx)
|
|
}
|
|
NetworkStream::Tls(conn) => {
|
|
serve_https(conn, connection_properties, cancel, tx)
|
|
}
|
|
#[cfg(unix)]
|
|
NetworkStream::Unix(conn) => {
|
|
serve_http(conn, connection_properties, cancel, tx)
|
|
}
|
|
}
|
|
}
|
|
|
|
struct HttpJoinHandle(
|
|
AsyncRefCell<Option<JoinHandle<Result<(), AnyError>>>>,
|
|
CancelHandle,
|
|
AsyncRefCell<tokio::sync::mpsc::Receiver<u32>>,
|
|
);
|
|
|
|
impl HttpJoinHandle {
|
|
fn cancel_handle(self: &Rc<Self>) -> RcRef<CancelHandle> {
|
|
RcRef::map(self, |this| &this.1)
|
|
}
|
|
}
|
|
|
|
impl Resource for HttpJoinHandle {
|
|
fn name(&self) -> Cow<str> {
|
|
"http".into()
|
|
}
|
|
|
|
fn close(self: Rc<Self>) {
|
|
self.1.cancel()
|
|
}
|
|
}
|
|
|
|
#[op(v8)]
|
|
pub fn op_serve_http(
|
|
state: Rc<RefCell<OpState>>,
|
|
listener_rid: ResourceId,
|
|
) -> Result<(ResourceId, &'static str, String), AnyError> {
|
|
let listener =
|
|
DefaultHttpRequestProperties::get_network_stream_listener_for_rid(
|
|
&mut state.borrow_mut(),
|
|
listener_rid,
|
|
)?;
|
|
|
|
let local_address = listener.listen_address()?;
|
|
let listen_properties = DefaultHttpRequestProperties::listen_properties(
|
|
listener.stream(),
|
|
&local_address,
|
|
);
|
|
|
|
let (tx, rx) = tokio::sync::mpsc::channel(10);
|
|
let resource: Rc<HttpJoinHandle> = Rc::new(HttpJoinHandle(
|
|
AsyncRefCell::new(None),
|
|
CancelHandle::new(),
|
|
AsyncRefCell::new(rx),
|
|
));
|
|
let cancel_clone = resource.cancel_handle();
|
|
|
|
let listen_properties_clone = listen_properties.clone();
|
|
let handle = spawn_local(async move {
|
|
loop {
|
|
let conn = listener
|
|
.accept()
|
|
.try_or_cancel(cancel_clone.clone())
|
|
.await?;
|
|
serve_http_on(
|
|
conn,
|
|
&listen_properties_clone,
|
|
cancel_clone.clone(),
|
|
tx.clone(),
|
|
);
|
|
}
|
|
#[allow(unreachable_code)]
|
|
Ok::<_, AnyError>(())
|
|
});
|
|
|
|
// Set the handle after we start the future
|
|
*RcRef::map(&resource, |this| &this.0)
|
|
.try_borrow_mut()
|
|
.unwrap() = Some(handle);
|
|
|
|
Ok((
|
|
state.borrow_mut().resource_table.add_rc(resource),
|
|
listen_properties.scheme,
|
|
listen_properties.fallback_host,
|
|
))
|
|
}
|
|
|
|
#[op(v8)]
|
|
pub fn op_serve_http_on(
|
|
state: Rc<RefCell<OpState>>,
|
|
conn: ResourceId,
|
|
) -> Result<(ResourceId, &'static str, String), AnyError> {
|
|
let network_stream =
|
|
DefaultHttpRequestProperties::get_network_stream_for_rid(
|
|
&mut state.borrow_mut(),
|
|
conn,
|
|
)?;
|
|
|
|
let local_address = network_stream.local_address()?;
|
|
let listen_properties = DefaultHttpRequestProperties::listen_properties(
|
|
network_stream.stream(),
|
|
&local_address,
|
|
);
|
|
|
|
let (tx, rx) = tokio::sync::mpsc::channel(10);
|
|
let resource: Rc<HttpJoinHandle> = Rc::new(HttpJoinHandle(
|
|
AsyncRefCell::new(None),
|
|
CancelHandle::new(),
|
|
AsyncRefCell::new(rx),
|
|
));
|
|
|
|
let handle: JoinHandle<Result<(), deno_core::anyhow::Error>> = serve_http_on(
|
|
network_stream,
|
|
&listen_properties,
|
|
resource.cancel_handle(),
|
|
tx,
|
|
);
|
|
|
|
// Set the handle after we start the future
|
|
*RcRef::map(&resource, |this| &this.0)
|
|
.try_borrow_mut()
|
|
.unwrap() = Some(handle);
|
|
|
|
Ok((
|
|
state.borrow_mut().resource_table.add_rc(resource),
|
|
listen_properties.scheme,
|
|
listen_properties.fallback_host,
|
|
))
|
|
}
|
|
|
|
#[op]
|
|
pub async fn op_http_wait(
|
|
state: Rc<RefCell<OpState>>,
|
|
rid: ResourceId,
|
|
) -> Result<u32, AnyError> {
|
|
// We will get the join handle initially, as we might be consuming requests still
|
|
let join_handle = state
|
|
.borrow_mut()
|
|
.resource_table
|
|
.get::<HttpJoinHandle>(rid)?;
|
|
|
|
let cancel = join_handle.clone().cancel_handle();
|
|
let next = async {
|
|
let mut recv = RcRef::map(&join_handle, |this| &this.2).borrow_mut().await;
|
|
recv.recv().await
|
|
}
|
|
.or_cancel(cancel)
|
|
.unwrap_or_else(|_| None)
|
|
.await;
|
|
|
|
// Do we have a request?
|
|
if let Some(req) = next {
|
|
return Ok(req);
|
|
}
|
|
|
|
// No - we're shutting down
|
|
let res = RcRef::map(join_handle, |this| &this.0)
|
|
.borrow_mut()
|
|
.await
|
|
.take()
|
|
.unwrap()
|
|
.await?;
|
|
|
|
// Drop the cancel and join handles
|
|
state
|
|
.borrow_mut()
|
|
.resource_table
|
|
.take::<HttpJoinHandle>(rid)?;
|
|
|
|
// Filter out shutdown (ENOTCONN) errors
|
|
if let Err(err) = res {
|
|
if let Some(err) = err.source() {
|
|
if let Some(err) = err.downcast_ref::<io::Error>() {
|
|
if err.kind() == io::ErrorKind::NotConnected {
|
|
return Ok(u32::MAX);
|
|
}
|
|
}
|
|
}
|
|
return Err(err);
|
|
}
|
|
|
|
Ok(u32::MAX)
|
|
}
|
|
|
|
struct UpgradeStream {
|
|
read: AsyncRefCell<tokio::io::ReadHalf<tokio::io::DuplexStream>>,
|
|
write: AsyncRefCell<tokio::io::WriteHalf<tokio::io::DuplexStream>>,
|
|
cancel_handle: CancelHandle,
|
|
}
|
|
|
|
impl UpgradeStream {
|
|
pub fn new(
|
|
read: tokio::io::ReadHalf<tokio::io::DuplexStream>,
|
|
write: tokio::io::WriteHalf<tokio::io::DuplexStream>,
|
|
) -> Self {
|
|
Self {
|
|
read: AsyncRefCell::new(read),
|
|
write: AsyncRefCell::new(write),
|
|
cancel_handle: CancelHandle::new(),
|
|
}
|
|
}
|
|
|
|
async fn read(self: Rc<Self>, buf: &mut [u8]) -> Result<usize, AnyError> {
|
|
let cancel_handle = RcRef::map(self.clone(), |this| &this.cancel_handle);
|
|
async {
|
|
let read = RcRef::map(self, |this| &this.read);
|
|
let mut read = read.borrow_mut().await;
|
|
Ok(Pin::new(&mut *read).read(buf).await?)
|
|
}
|
|
.try_or_cancel(cancel_handle)
|
|
.await
|
|
}
|
|
|
|
async fn write(self: Rc<Self>, buf: &[u8]) -> Result<usize, AnyError> {
|
|
let cancel_handle = RcRef::map(self.clone(), |this| &this.cancel_handle);
|
|
async {
|
|
let write = RcRef::map(self, |this| &this.write);
|
|
let mut write = write.borrow_mut().await;
|
|
Ok(Pin::new(&mut *write).write(buf).await?)
|
|
}
|
|
.try_or_cancel(cancel_handle)
|
|
.await
|
|
}
|
|
}
|
|
|
|
impl Resource for UpgradeStream {
|
|
fn name(&self) -> Cow<str> {
|
|
"httpRawUpgradeStream".into()
|
|
}
|
|
|
|
deno_core::impl_readable_byob!();
|
|
deno_core::impl_writable!();
|
|
|
|
fn close(self: Rc<Self>) {
|
|
self.cancel_handle.cancel();
|
|
}
|
|
}
|