// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
use crate::ops::io::TcpStreamResource;
use crate::ops::io::TlsServerStreamResource;
use deno_core::error::bad_resource_id;
use deno_core::error::null_opbuf;
use deno_core::error::type_error;
use deno_core::error::AnyError;
use deno_core::futures::future::poll_fn;
use deno_core::futures::FutureExt;
use deno_core::futures::Stream;
use deno_core::futures::StreamExt;
use deno_core::op_async;
use deno_core::op_sync;
use deno_core::AsyncRefCell;
use deno_core::CancelFuture;
use deno_core::CancelHandle;
use deno_core::CancelTryFuture;
use deno_core::Extension;
use deno_core::OpState;
use deno_core::RcRef;
use deno_core::Resource;
use deno_core::ResourceId;
use deno_core::ZeroCopyBuf;
use hyper::body::HttpBody;
use hyper::http;
use hyper::server::conn::Connection;
use hyper::server::conn::Http;
use hyper::service::Service as HyperService;
use hyper::Body;
use hyper::Request;
use hyper::Response;
use serde::Deserialize;
use serde::Serialize;
use std::borrow::Cow;
use std::cell::RefCell;
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::rc::Rc;
use std::task::Context;
use std::task::Poll;
use tokio::io::AsyncReadExt;
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tokio_rustls::server::TlsStream;
use tokio_util::io::StreamReader;
pub fn init() -> Extension {
Extension::builder()
.ops(vec![
("op_http_start", op_sync(op_http_start)),
("op_http_request_next", op_async(op_http_request_next)),
("op_http_request_read", op_async(op_http_request_read)),
("op_http_response", op_async(op_http_response)),
("op_http_response_write", op_async(op_http_response_write)),
("op_http_response_close", op_async(op_http_response_close)),
])
.build()
}
struct ServiceInner {
request: Request
,
response_tx: oneshot::Sender>,
}
#[derive(Clone, Default)]
struct Service {
inner: Rc>>,
}
impl HyperService> for Service {
type Response = Response;
type Error = http::Error;
#[allow(clippy::type_complexity)]
type Future =
Pin>>>;
fn poll_ready(
&mut self,
_cx: &mut Context<'_>,
) -> Poll> {
if self.inner.borrow().is_some() {
Poll::Pending
} else {
Poll::Ready(Ok(()))
}
}
fn call(&mut self, req: Request) -> Self::Future {
let (resp_tx, resp_rx) = oneshot::channel();
self.inner.borrow_mut().replace(ServiceInner {
request: req,
response_tx: resp_tx,
});
async move { Ok(resp_rx.await.unwrap()) }.boxed_local()
}
}
enum ConnType {
Tcp(Rc>>),
Tls(Rc, Service, LocalExecutor>>>),
}
struct ConnResource {
hyper_connection: ConnType,
deno_service: Service,
addr: SocketAddr,
}
impl ConnResource {
// TODO(ry) impl Future for ConnResource?
fn poll(&self, cx: &mut Context<'_>) -> Poll> {
match &self.hyper_connection {
ConnType::Tcp(c) => c.borrow_mut().poll_unpin(cx),
ConnType::Tls(c) => c.borrow_mut().poll_unpin(cx),
}
.map_err(AnyError::from)
}
}
impl Resource for ConnResource {
fn name(&self) -> Cow {
"httpConnection".into()
}
}
// We use a tuple instead of struct to avoid serialization overhead of the keys.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct NextRequestResponse(
// request_body_rid:
Option,
// response_sender_rid:
ResourceId,
// method:
String,
// headers:
Vec<(String, String)>,
// url:
String,
);
async fn op_http_request_next(
state: Rc>,
conn_rid: ResourceId,
_: (),
) -> Result