refactor: deno_http op crate (#11335)

Cargo.lock

@@ -551,6 +551,7 @@ dependencies = [
  "deno_crypto",
  "deno_doc",
  "deno_fetch",
+ "deno_http",
  "deno_lint",
  "deno_net",
  "deno_runtime",
@@ -707,6 +708,21 @@ dependencies = [
  "tokio-util",
 ]
 
+[[package]]
+name = "deno_http"
+version = "0.1.0"
+dependencies = [
+ "base64 0.13.0",
+ "bytes",
+ "deno_core",
+ "deno_websocket",
+ "hyper",
+ "ring",
+ "serde",
+ "tokio",
+ "tokio-util",
+]
+
 [[package]]
 name = "deno_lint"
 version = "0.9.0"
@@ -731,19 +747,12 @@ dependencies = [
 name = "deno_net"
 version = "0.1.0"
 dependencies = [
- "base64 0.13.0",
- "bytes",
  "deno_core",
- "deno_websocket",
- "http",
- "hyper",
  "lazy_static",
  "log",
- "ring",
  "rustls",
  "serde",
  "tokio",
- "tokio-util",
  "trust-dns-proto",
  "trust-dns-resolver",
  "webpki",
@@ -760,6 +769,7 @@ dependencies = [
  "deno_core",
  "deno_crypto",
  "deno_fetch",
+ "deno_http",
  "deno_net",
  "deno_timers",
  "deno_url",

Cargo.toml

@@ -12,6 +12,7 @@ members = [
   "extensions/console",
   "extensions/crypto",
   "extensions/fetch",
+  "extensions/http",
   "extensions/net",
   "extensions/timers",
   "extensions/url",
@@ -55,6 +56,8 @@ opt-level = 3
 opt-level = 3
 [profile.bench.package.deno_runtime]
 opt-level = 3
+[profile.bench.package.deno_http]
+opt-level = 3
 [profile.bench.package.deno_net]
 opt-level = 3
 [profile.bench.package.rusty_v8]
@@ -81,6 +84,8 @@ opt-level = 3
 opt-level = 3
 [profile.release.package.deno_runtime]
 opt-level = 3
+[profile.release.package.deno_http]
+opt-level = 3
 [profile.release.package.deno_net]
 opt-level = 3
 [profile.release.package.rusty_v8]

cli/Cargo.toml

@@ -25,6 +25,7 @@ deno_console = { version = "0.10.0", path = "../extensions/console" }
 deno_core = { version = "0.92.0", path = "../core" }
 deno_crypto = { version = "0.24.0", path = "../extensions/crypto" }
 deno_fetch = { version = "0.32.0", path = "../extensions/fetch" }
+deno_http = { version = "0.1.0", path = "../extensions/http" }
 deno_net = { version = "0.1.0", path = "../extensions/net" }
 deno_timers = { version = "0.8.0", path = "../extensions/timers" }
 deno_url = { version = "0.10.0", path = "../extensions/url" }

cli/build.rs

@@ -71,6 +71,8 @@ fn create_compiler_snapshot(
   op_crate_libs.insert("deno.net", deno_net::get_declaration());
   op_crate_libs
     .insert("deno.net_unstable", deno_net::get_unstable_declaration());
+  op_crate_libs
+    .insert("deno.http_unstable", deno_http::get_unstable_declaration());
 
   // ensure we invalidate the build properly.
   for (_, path) in op_crate_libs.iter() {
@@ -313,6 +315,10 @@ fn main() {
     "cargo:rustc-env=DENO_NET_UNSTABLE_LIB_PATH={}",
     deno_net::get_unstable_declaration().display()
   );
+  println!(
+    "cargo:rustc-env=DENO_HTTP_UNSTABLE_LIB_PATH={}",
+    deno_http::get_unstable_declaration().display()
+  );
 
   println!("cargo:rustc-env=TARGET={}", env::var("TARGET").unwrap());
   println!("cargo:rustc-env=PROFILE={}", env::var("PROFILE").unwrap());

cli/dts/lib.deno.unstable.d.ts

@@ -3,6 +3,7 @@
 /// <reference no-default-lib="true" />
 /// <reference lib="deno.ns" />
 /// <reference lib="deno.net_unstable" />
+/// <reference lib="deno.http_unstable" />
 
 declare namespace Deno {
   /**
@@ -1098,6 +1099,24 @@ declare namespace Deno {
       write?: "inherit" | boolean | Array<string | URL>;
     };
   }
+
+  /** **UNSTABLE**: new API, yet to be vetted.
+   *
+   * Services HTTP requests given a TCP or TLS socket.
+   *
+   * ```ts
+   * const conn = await Deno.connect({ port: 80, hostname: "127.0.0.1" });
+   * const httpConn = Deno.serveHttp(conn);
+   * const e = await httpConn.nextRequest();
+   * if (e) {
+   *   e.respondWith(new Response("Hello World"));
+   * }
+   * ```
+   *
+   * If `httpConn.nextRequest()` encounters an error or returns `null`
+   * then the underlying HttpConn resource is closed automatically.
+   */
+  export function serveHttp(conn: Conn): HttpConn;
 }
 
 declare function fetch(

cli/main.rs

@@ -348,6 +348,7 @@ pub fn get_types(unstable: bool) -> String {
   if unstable {
     types.push(crate::tsc::UNSTABLE_NS_LIB);
     types.push(crate::tsc::DENO_NET_UNSTABLE_LIB);
+    types.push(crate::tsc::DENO_HTTP_UNSTABLE_LIB);
   }
 
   types.join("\n")

cli/tsc.rs

@@ -47,6 +47,8 @@ pub static DENO_BROADCAST_CHANNEL_LIB: &str =
 pub static DENO_NET_LIB: &str = include_str!(env!("DENO_NET_LIB_PATH"));
 pub static DENO_NET_UNSTABLE_LIB: &str =
   include_str!(env!("DENO_NET_UNSTABLE_LIB_PATH"));
+pub static DENO_HTTP_UNSTABLE_LIB: &str =
+  include_str!(env!("DENO_HTTP_UNSTABLE_LIB_PATH"));
 pub static SHARED_GLOBALS_LIB: &str =
   include_str!("dts/lib.deno.shared_globals.d.ts");
 pub static WINDOW_LIB: &str = include_str!("dts/lib.deno.window.d.ts");

extensions/http/01_http.js

@@ -32,11 +32,6 @@
     Uint8Array,
   } = window.__bootstrap.primordials;
 
-  function serveHttp(conn) {
-    const rid = core.opSync("op_http_start", conn.rid);
-    return new HttpConn(rid);
-  }
-
   const connErrorSymbol = Symbol("connError");
 
   class HttpConn {
@@ -373,7 +368,7 @@
   }
 
   window.__bootstrap.http = {
-    serveHttp,
+    HttpConn,
     upgradeWebSocket,
   };
 })(this);

extensions/http/Cargo.toml

@@ -0,0 +1,25 @@
+# Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+[package]
+name = "deno_http"
+version = "0.1.0"
+edition = "2018"
+description = "HTTP server implementation for Deno"
+authors = ["the Deno authors"]
+license = "MIT"
+readme = "README.md"
+repository = "https://github.com/denoland/deno"
+
+[lib]
+path = "lib.rs"
+
+[dependencies]
+base64 = "0.13.0"
+bytes = "1"
+deno_core = { version = "0.92.0", path = "../../core" }
+deno_websocket = { version = "0.15.1", path = "../websocket" }
+hyper = { version = "0.14.9", features = ["server", "stream", "http1", "http2", "runtime"] }
+ring = "0.16.20"
+serde = { version = "1.0.125", features = ["derive"] }
+tokio = { version = "1.8.0", features = ["full"] }
+tokio-util = "0.6.7"

extensions/http/README.md

@@ -0,0 +1,4 @@
+# deno_http
+
+This crate implements server-side HTTP based on primitives from the
+[Fetch API](https://fetch.spec.whatwg.org/).

extensions/http/lib.deno_http.unstable.d.ts

@@ -0,0 +1,65 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+/// <reference no-default-lib="true" />
+/// <reference lib="esnext" />
+
+declare namespace Deno {
+  export interface RequestEvent {
+    readonly request: Request;
+    respondWith(r: Response | Promise<Response>): Promise<void>;
+  }
+
+  export interface HttpConn extends AsyncIterable<RequestEvent> {
+    readonly rid: number;
+
+    nextRequest(): Promise<RequestEvent | null>;
+    close(): void;
+  }
+
+  export interface WebSocketUpgrade {
+    response: Response;
+    websocket: WebSocket;
+  }
+
+  export interface UpgradeWebSocketOptions {
+    protocol?: string;
+  }
+
+  /** **UNSTABLE**: new API, yet to be vetted.
+   *
+   * Used to upgrade an incoming HTTP request to a WebSocket.
+   *
+   * Given a request, returns a pair of WebSocket and Response. The original
+   * request must be responded to with the returned response for the websocket
+   * upgrade to be successful.
+   *
+   * ```ts
+   * const conn = await Deno.connect({ port: 80, hostname: "127.0.0.1" });
+   * const httpConn = Deno.serveHttp(conn);
+   * const e = await httpConn.nextRequest();
+   * if (e) {
+   *   const { websocket, response } = Deno.upgradeWebSocket(e.request);
+   *   websocket.onopen = () => {
+   *     websocket.send("Hello World!");
+   *   };
+   *   websocket.onmessage = (e) => {
+   *     console.log(e.data);
+   *     websocket.close();
+   *   };
+   *   websocket.onclose = () => console.log("WebSocket has been closed.");
+   *   websocket.onerror = (e) => console.error("WebSocket error:", e.message);
+   *   e.respondWith(response);
+   * }
+   * ```
+   *
+   * If the request body is disturbed (read from) before the upgrade is
+   * completed, upgrading fails.
+   *
+   * This operation does not yet consume the request or open the websocket. This
+   * only happens once the returned response has been passed to `respondWith`.
+   */
+  export function upgradeWebSocket(
+    request: Request,
+    options?: UpgradeWebSocketOptions,
+  ): WebSocketUpgrade;
+}

extensions/http/lib.rs

@@ -1,7 +1,5 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 
-use crate::io::TcpStreamResource;
-use crate::io::TlsStreamResource;
 use deno_core::error::bad_resource_id;
 use deno_core::error::null_opbuf;
 use deno_core::error::type_error;
@@ -10,13 +8,14 @@ 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::include_js_files;
 use deno_core::op_async;
 use deno_core::op_sync;
 use deno_core::AsyncRefCell;
 use deno_core::ByteString;
 use deno_core::CancelHandle;
 use deno_core::CancelTryFuture;
-use deno_core::OpPair;
+use deno_core::Extension;
 use deno_core::OpState;
 use deno_core::RcRef;
 use deno_core::Resource;
@@ -35,17 +34,28 @@ use std::borrow::Cow;
 use std::cell::RefCell;
 use std::future::Future;
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::pin::Pin;
 use std::rc::Rc;
 use std::task::Context;
 use std::task::Poll;
+use tokio::io::AsyncRead;
 use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWrite;
 use tokio::sync::oneshot;
 use tokio_util::io::StreamReader;
 
-pub fn init() -> Vec<OpPair> {
-  vec![
-    ("op_http_start", op_sync(op_http_start)),
+pub fn get_unstable_declaration() -> PathBuf {
+  PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("lib.deno_http.unstable.d.ts")
+}
+
+pub fn init() -> Extension {
+  Extension::builder()
+    .js(include_js_files!(
+      prefix "deno:extensions/http",
+      "01_http.js",
+    ))
+    .ops(vec![
       ("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)),
@@ -59,7 +69,8 @@ pub fn init() -> Vec<OpPair> {
         "op_http_upgrade_websocket",
         op_async(op_http_upgrade_websocket),
       ),
-  ]
+    ])
+    .build()
 }
 
 struct ServiceInner {
@@ -106,13 +117,13 @@ type ConnFuture = Pin<Box<dyn Future<Output = hyper::Result<()>>>>;
 
 struct Conn {
   scheme: &'static str,
+  addr: SocketAddr,
   conn: Rc<RefCell<ConnFuture>>,
 }
 
 struct ConnResource {
   hyper_connection: Conn,
   deno_service: Service,
-  addr: SocketAddr,
   cancel: CancelHandle,
 }
 
@@ -221,7 +232,7 @@ async fn op_http_request_next(
         } else if let Some(host) = req.headers().get("HOST") {
           Cow::Borrowed(host.to_str()?)
         } else {
-          Cow::Owned(conn_resource.addr.to_string())
+          Cow::Owned(conn_resource.hyper_connection.addr.to_string())
         };
         let path = req.uri().path_and_query().map_or("/", |p| p.as_str());
         format!("{}://{}{}", scheme, host, path)
@@ -299,69 +310,30 @@ fn should_ignore_error(e: &AnyError) -> bool {
   false
 }
 
-fn op_http_start(
+pub fn start_http<IO: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
   state: &mut OpState,
-  tcp_stream_rid: ResourceId,
-  _: (),
+  io: IO,
+  addr: SocketAddr,
+  scheme: &'static str,
 ) -> Result<ResourceId, AnyError> {
   let deno_service = Service::default();
 
-  if let Some(resource_rc) = state
-    .resource_table
-    .take::<TcpStreamResource>(tcp_stream_rid)
-  {
-    let resource = Rc::try_unwrap(resource_rc)
-      .expect("Only a single use of this resource should happen");
-    let (read_half, write_half) = resource.into_inner();
-    let tcp_stream = read_half.reunite(write_half)?;
-    let addr = tcp_stream.local_addr()?;
   let hyper_connection = Http::new()
     .with_executor(LocalExecutor)
-      .serve_connection(tcp_stream, deno_service.clone())
+    .serve_connection(io, deno_service.clone())
     .with_upgrades();
   let conn = Pin::new(Box::new(hyper_connection));
   let conn_resource = ConnResource {
     hyper_connection: Conn {
+      scheme,
+      addr,
       conn: Rc::new(RefCell::new(conn)),
-        scheme: "http",
     },
     deno_service,
-      addr,
     cancel: CancelHandle::default(),
   };
   let rid = state.resource_table.add(conn_resource);
-    return Ok(rid);
-  }
-
-  if let Some(resource_rc) = state
-    .resource_table
-    .take::<TlsStreamResource>(tcp_stream_rid)
-  {
-    let resource = Rc::try_unwrap(resource_rc)
-      .expect("Only a single use of this resource should happen");
-    let (read_half, write_half) = resource.into_inner();
-    let tls_stream = read_half.reunite(write_half);
-    let addr = tls_stream.get_ref().0.local_addr()?;
-
-    let hyper_connection = Http::new()
-      .with_executor(LocalExecutor)
-      .serve_connection(tls_stream, deno_service.clone())
-      .with_upgrades();
-    let conn = Pin::new(Box::new(hyper_connection));
-    let conn_resource = ConnResource {
-      hyper_connection: Conn {
-        conn: Rc::new(RefCell::new(conn)),
-        scheme: "https",
-      },
-      deno_service,
-      addr,
-      cancel: CancelHandle::default(),
-    };
-    let rid = state.resource_table.add(conn_resource);
-    return Ok(rid);
-  }
-
-  Err(bad_resource_id())
+  Ok(rid)
 }
 
 // We use a tuple instead of struct to avoid serialization overhead of the keys.

extensions/net/Cargo.toml

@@ -15,19 +15,12 @@ path = "lib.rs"
 
 [dependencies]
 deno_core = { version = "0.92.0", path = "../../core" }
-deno_websocket = { version = "0.15.0", path = "../websocket" }
 
-base64 = "0.13.0"
-bytes = "1"
 log = "0.4.14"
 lazy_static = "1.4.0"
-http = "0.2.4"
-hyper = { version = "0.14.10", features = ["server", "stream", "http1", "http2", "runtime"] }
-ring = "0.16.20"
-rustls = "0.19.1"
+rustls = "0.19.0"
 serde = { version = "1.0.126", features = ["derive"] }
 tokio = { version = "1.8.1", features = ["full"] }
-tokio-util = { version = "0.6", features = ["io"] }
 webpki = "0.21.4"
 webpki-roots = "0.21.1"
 trust-dns-proto = "0.20.3"

extensions/net/lib.deno_net.unstable.d.ts

@@ -229,48 +229,4 @@ declare namespace Deno {
   */
     alpnProtocols?: string[];
   }
-
-  export interface RequestEvent {
-    readonly request: Request;
-    respondWith(r: Response | Promise<Response>): Promise<void>;
-  }
-
-  export interface HttpConn extends AsyncIterable<RequestEvent> {
-    readonly rid: number;
-
-    nextRequest(): Promise<RequestEvent | null>;
-    close(): void;
-  }
-
-  /** **UNSTABLE**: new API, yet to be vetted.
-   *
-   * Services HTTP requests given a TCP or TLS socket.
-   *
-   * ```ts
-   * const conn = await Deno.connect({ port: 80, hostname: "127.0.0.1" });
-   * const httpConn = Deno.serveHttp(conn);
-   * const e = await httpConn.nextRequest();
-   * if (e) {
-   *   e.respondWith(new Response("Hello World"));
-   * }
-   * ```
-   *
-   * If `httpConn.nextRequest()` encounters an error or returns `null`
-   * then the underlying HttpConn resource is closed automatically.
-   */
-  export function serveHttp(conn: Conn): HttpConn;
-
-  export interface WebSocketUpgrade {
-    response: Response;
-    websocket: WebSocket;
-  }
-
-  export interface UpgradeWebSocketOptions {
-    protocol?: string;
-  }
-
-  export function upgradeWebSocket(
-    request: Request,
-    options?: UpgradeWebSocketOptions,
-  ): WebSocketUpgrade;
 }

extensions/net/lib.rs

@@ -2,7 +2,6 @@
 
 pub mod io;
 pub mod ops;
-pub mod ops_http;
 pub mod ops_tls;
 #[cfg(unix)]
 pub mod ops_unix;
@@ -94,14 +93,12 @@ pub fn init<P: NetPermissions + 'static>(unstable: bool) -> Extension {
   ops_to_register.extend(io::init());
   ops_to_register.extend(ops::init::<P>());
   ops_to_register.extend(ops_tls::init::<P>());
-  ops_to_register.extend(ops_http::init());
 
   Extension::builder()
     .js(include_js_files!(
       prefix "deno:extensions/net",
       "01_net.js",
       "02_tls.js",
-      "03_http.js",
       "04_net_unstable.js",
     ))
     .ops(ops_to_register)

runtime/Cargo.toml

@@ -23,6 +23,7 @@ deno_console = { version = "0.10.1", path = "../extensions/console" }
 deno_core = { version = "0.92.0", path = "../core" }
 deno_crypto = { version = "0.24.1", path = "../extensions/crypto" }
 deno_fetch = { version = "0.32.1", path = "../extensions/fetch" }
+deno_http = { version = "0.1.0", path = "../extensions/http" }
 deno_net = { version = "0.1.0", path = "../extensions/net" }
 deno_timers = { version = "0.8.1", path = "../extensions/timers" }
 deno_url = { version = "0.10.1", path = "../extensions/url" }
@@ -42,6 +43,7 @@ deno_console = { version = "0.10.1", path = "../extensions/console" }
 deno_core = { version = "0.92.0", path = "../core" }
 deno_crypto = { version = "0.24.1", path = "../extensions/crypto" }
 deno_fetch = { version = "0.32.1", path = "../extensions/fetch" }
+deno_http = { version = "0.1.0", path = "../extensions/http" }
 deno_net = { version = "0.1.0", path = "../extensions/net" }
 deno_timers = { version = "0.8.1", path = "../extensions/timers" }
 deno_url = { version = "0.10.1", path = "../extensions/url" }

runtime/build.rs

@@ -60,6 +60,7 @@ fn create_runtime_snapshot(snapshot_path: &Path, files: Vec<PathBuf>) {
       false, // No --unstable.
     ),
     deno_net::init::<deno_net::NoNetPermissions>(false), // No --unstable.
+    deno_http::init(),
   ];
 
   let js_runtime = JsRuntime::new(RuntimeOptions {

runtime/js/40_http.js

@@ -0,0 +1,14 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+"use strict";
+
+((window) => {
+  const core = window.__bootstrap.core;
+  const { HttpConn } = window.__bootstrap.http;
+
+  function serveHttp(conn) {
+    const rid = core.opSync("op_http_start", conn.rid);
+    return new HttpConn(rid);
+  }
+
+  window.__bootstrap.http.serveHttp = serveHttp;
+})(globalThis);

runtime/lib.rs

@@ -4,6 +4,7 @@ pub use deno_broadcast_channel;
 pub use deno_console;
 pub use deno_crypto;
 pub use deno_fetch;
+pub use deno_http;
 pub use deno_net;
 pub use deno_timers;
 pub use deno_url;

runtime/ops/http.rs

@@ -0,0 +1,48 @@
+use std::rc::Rc;
+
+use deno_core::error::bad_resource_id;
+use deno_core::error::AnyError;
+use deno_core::op_sync;
+use deno_core::Extension;
+use deno_core::OpState;
+use deno_core::ResourceId;
+use deno_net::io::TcpStreamResource;
+use deno_net::io::TlsStreamResource;
+
+pub fn init() -> Extension {
+  Extension::builder()
+    .ops(vec![("op_http_start", op_sync(op_http_start))])
+    .build()
+}
+
+fn op_http_start(
+  state: &mut OpState,
+  tcp_stream_rid: ResourceId,
+  _: (),
+) -> Result<ResourceId, AnyError> {
+  if let Some(resource_rc) = state
+    .resource_table
+    .take::<TcpStreamResource>(tcp_stream_rid)
+  {
+    let resource = Rc::try_unwrap(resource_rc)
+      .expect("Only a single use of this resource should happen");
+    let (read_half, write_half) = resource.into_inner();
+    let tcp_stream = read_half.reunite(write_half)?;
+    let addr = tcp_stream.local_addr()?;
+    return deno_http::start_http(state, tcp_stream, addr, "http");
+  }
+
+  if let Some(resource_rc) = state
+    .resource_table
+    .take::<TlsStreamResource>(tcp_stream_rid)
+  {
+    let resource = Rc::try_unwrap(resource_rc)
+      .expect("Only a single use of this resource should happen");
+    let (read_half, write_half) = resource.into_inner();
+    let tls_stream = read_half.reunite(write_half);
+    let addr = tls_stream.get_ref().0.local_addr()?;
+    return deno_http::start_http(state, tls_stream, addr, "https");
+  }
+
+  Err(bad_resource_id())
+}

runtime/ops/mod.rs

@@ -2,6 +2,7 @@
 
 pub mod fs;
 pub mod fs_events;
+pub mod http;
 pub mod io;
 pub mod os;
 pub mod permissions;

runtime/web_worker.rs

@@ -339,6 +339,8 @@ impl WebWorker {
         ops::process::init(),
         ops::signal::init(),
         ops::tty::init(),
+        deno_http::init(),
+        ops::http::init(),
         ops::io::init_stdio(),
       ]
     } else {

runtime/worker.rs

@@ -130,6 +130,8 @@ impl MainWorker {
       ops::process::init(),
       ops::signal::init(),
       ops::tty::init(),
+      deno_http::init(),
+      ops::http::init(),
       // Permissions ext (worker specific state)
       perm_ext,
     ];