diff --git a/js/README.md b/js/README.md index caeec65c0..d358aa819 100644 --- a/js/README.md +++ b/js/README.md @@ -156,6 +156,29 @@ such as the Rust implementation, otherwise waits forever. byte streams with a half-close, such as WebTransport bidirectional streams or TCP sockets (with half-open connections allowed). +`fromFetch(input, options?)` returns an HTTP client transport for one +invocation, sending a `POST` request to `input` with `options.fetch` +(`globalThis.fetch` by default) and failing the invocation unless the response +status is `2xx`. The other `options`, such as `headers`, are passed to `fetch`. +The request body is streamed, which needs a runtime that supports streaming +request bodies: Firefox and Safari do not, and Chromium only does over HTTP/2 +and only exposes the response once the request body is complete. With +`buffer: true`, the whole request body is kept in memory and only sent once the +client ends transmit, so the server receives nothing until every asynchronous +parameter is complete. + +`fromRequest(request)` serves an invocation over a WHATWG `Request`, such as +in Deno, Bun or a service worker, returning the `transport` to `accept` it on +and the `response` to reply with without waiting for the request body: + +```js +Deno.serve((request) => { + const { transport, response } = fromRequest(request); + serve(transport); // `accept(transport)`, then receive the parameters and send the results + return response; +}); +``` + ## Value mapping Component-model values map to JavaScript following the `ToJSValue` and @@ -230,6 +253,8 @@ It deviates from the draft in that: - `accept(transport)` — accept an invocation; resolves to `{ instance, func, receiveParams, sendResults, finished, mux }`. - `fromWebStreams({ readable, writable })` — adapt a WHATWG stream duplex to a transport (see [Transports](#transports)). +- `fromFetch(input, options?)` / `fromRequest(request)` — the HTTP client and server transports + (see [Transports](#transports)). - `encodeValue(writer, ty, value, sink?, path?)` / `decodeValue(reader, ty, sink?, path?)` — the low-level value codec. - `decodeSteps(reader, ty, sink?, path?)` / `atomic(reader, f)` — the resumable form of the decoder, driven by `AsyncReader.decode` across chunk boundaries. - `Mux` / `Outgoing` — the framing multiplexer, for custom session control; `DISCARDED` is the error the reader of a discarded sub-stream fails with. diff --git a/js/src/http.js b/js/src/http.js new file mode 100644 index 000000000..e7a7c53f5 --- /dev/null +++ b/js/src/http.js @@ -0,0 +1,91 @@ +// The HTTP transport: one `POST` per invocation, whose request body carries +// the client's data and whose response body carries the server's. + +import { fromWebStreams } from "./mux.js"; + +/** + * @typedef {import("./mux.js").Transport} Transport + * @typedef {Omit & { fetch?: typeof fetch, buffer?: boolean }} FetchOptions + */ + +/** + * A client {@link Transport} sending a `POST` request to `input` with `fetch`, + * whose response must have a `2xx` status. The other `options` are passed to + * `fetch`. The request body is streamed, unless `buffer` is set: then it is + * sent once the client has ended transmit, for runtimes which cannot stream + * request bodies. + * @param {Parameters[0]} input + * @param {FetchOptions} [options] + * @returns {Transport} + */ +export function fromFetch(input, { fetch = globalThis.fetch, buffer = false, ...init } = {}) { + const ctrl = new AbortController(); + /** @type {TransformStreamDefaultController | undefined} */ + let controller; + const stream = buffer ? undefined : new TransformStream({ start: (c) => void (controller = c) }); + /** @type {(body: RequestInit["body"]) => void} */ + let send = () => {}; + /** @type {(reason: unknown) => void} */ + let fail = () => {}; + /** @type {Promise>} */ + const reader = new Promise((resolve, reject) => { + send = resolve; + fail = reject; + }) + .then((body) => { + const req = { ...init, method: "POST", body, signal: ctrl.signal }; + return fetch(input, buffer ? req : /** @type {RequestInit} */ ({ ...req, duplex: "half" })); + }) + .then(async (res) => { + if (!res.ok) { + await res.body?.cancel().catch(() => {}); + throw new Error(`HTTP request failed with status ${res.status}`); + } + return (res.body ?? new Blob([]).stream()).getReader(); + }); + /** @param {unknown} reason */ + const abort = (reason) => { + fail(reason); + ctrl.abort(reason); + controller?.error(reason); + reader.then((r) => r.cancel(reason)).catch(() => {}); + }; + reader.catch(abort); + /** @type {Pick} */ + let body; + if (stream === undefined) { + /** @type {Uint8Array[]} */ + const chunks = []; + body = { + write: (bytes) => void chunks.push(bytes), + closeWrite: () => send(new Blob(/** @type {Uint8Array[]} */ (chunks))), + }; + } else { + const writer = stream.writable.getWriter(); + send(stream.readable); + body = { write: (bytes) => writer.write(bytes), closeWrite: () => writer.close() }; + } + return { + ...body, + read: async () => { + const { value, done } = await (await reader).read(); + return done ? null : value; + }, + abort, + }; +} + +/** + * A server {@link Transport} over the body of `request`, returned with the + * `response` carrying the server's data, which must be sent without waiting + * for the request body to complete. + * @param {Request} request + * @returns {{ transport: Transport, response: Response }} + */ +export function fromRequest(request) { + const { readable, writable } = new TransformStream(); + return { + transport: fromWebStreams({ readable: request.body ?? new Blob([]).stream(), writable }), + response: new Response(readable), + }; +} diff --git a/js/src/index.d.ts b/js/src/index.d.ts index 3766227d8..806664b1f 100644 --- a/js/src/index.d.ts +++ b/js/src/index.d.ts @@ -332,6 +332,22 @@ export class Mux { closeWrite(): Promise; } +// ----- http.js -------------------------------------------------------------- + +/** Options of {@link fromFetch}: the `fetch` request options, except for the method, body and signal. */ +export interface FetchOptions extends Omit { + /** The `fetch` implementation to use, `globalThis.fetch` by default. */ + fetch?: typeof fetch; + /** Buffer the request body and send it once the client ends transmit. */ + buffer?: boolean; +} + +/** A client transport sending a `POST` request to `input` per invocation. */ +export function fromFetch(input: RequestInfo | URL, options?: FetchOptions): Transport; + +/** A server transport over `request`, with the response carrying the server's data. */ +export function fromRequest(request: Request): { transport: Transport; response: Response }; + // ----- session.js ----------------------------------------------------------- /** diff --git a/js/src/index.js b/js/src/index.js index edc949fae..ed0a7d099 100644 --- a/js/src/index.js +++ b/js/src/index.js @@ -12,4 +12,5 @@ export * as types from "./types.js"; export { Reader, Writer, AsyncReader, Chan, atomic } from "./io.js"; export { encodeValue, decodeValue, decodeSteps } from "./value.js"; export { DISCARDED, Mux, Outgoing, PROTOCOL, fromWebStreams } from "./mux.js"; +export { fromFetch, fromRequest } from "./http.js"; export { invoke, accept } from "./session.js"; diff --git a/js/test/http.test.js b/js/test/http.test.js new file mode 100644 index 000000000..fe35f3603 --- /dev/null +++ b/js/test/http.test.js @@ -0,0 +1,49 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import * as t from "../src/types.js"; +import { fromFetch, fromRequest } from "../src/http.js"; +import { accept, invoke } from "../src/session.js"; + +const ADDR = "http://localhost/rpc"; + +test("values echo over HTTP", async () => { + const ty = t.tuple(t.string, t.stream(t.u8)); + for (const buffer of [false, true]) { + /** @type {Promise[]} */ + const serving = []; + /** @type {typeof globalThis.fetch} */ + const fetch = async (input, init) => { + const { transport, response } = fromRequest(new Request(input, init)); + serving.push( + (async () => { + const inv = await accept(transport); + const { params, done } = await inv.receiveParams([ty]); + await Promise.all([done, inv.sendResults([ty], params), inv.finished]); + })(), + ); + return response; + }; + const { results, done, finished } = await invoke( + fromFetch(ADDR, { fetch, buffer }), + "i", + "f", + [ty], + [["hi", Uint8Array.of(1, 2)]], + [ty], + ); + const [s, bytes] = /** @type {[string, AsyncIterable]} */ (results[0]); + const chunks = []; + for await (const chunk of bytes) chunks.push(...chunk); + assert.deepEqual([s, chunks], ["hi", [1, 2]]); + await Promise.all([done, finished, ...serving]); + } +}); + +test("a response status other than 2xx fails the invocation", async () => { + /** @type {typeof globalThis.fetch} */ + const fetch = async () => new Response("nope", { status: 404 }); + await assert.rejects(invoke(fromFetch(ADDR, { fetch }), "i", "f", [], [], [t.u8]), { + message: "HTTP request failed with status 404", + }); +}); diff --git a/tests/js.rs b/tests/js.rs index cb9db4703..426fb4b5a 100644 --- a/tests/js.rs +++ b/tests/js.rs @@ -2,29 +2,27 @@ use core::net::{Ipv6Addr, SocketAddr}; use core::pin::pin; use std::process::Stdio; +use std::sync::Arc; use anyhow::{Context as _, bail, ensure}; use futures::{StreamExt as _, stream}; +use hyper::client::conn::http2; +use hyper::server::conn::http1; +use hyper_util::rt::{TokioExecutor, TokioIo}; use tokio::io::{AsyncBufReadExt as _, BufReader}; -use tokio::net::TcpListener; -use tokio::process::Command; +use tokio::net::{TcpListener, TcpStream}; +use tokio::process::{Child, Command}; use tokio::{select, spawn}; use tracing::{info, instrument}; pub mod common; -#[instrument(ret)] -#[test_log::test(tokio::test(flavor = "multi_thread"))] -async fn js_server() -> anyhow::Result<()> { - let port = wrpc_test::free_port() - .await - .context("failed to find free port")?; - let addr = SocketAddr::from((Ipv6Addr::LOCALHOST, port)); - +/// Start `tests/js/server.js` listening on `addr` and wait until it is ready. +async fn start_server(addr: &str) -> anyhow::Result { info!("starting `tests/js/server.js`"); let mut server = Command::new("node") .arg("tests/js/server.js") - .arg(addr.to_string()) + .arg(addr) .stdout(Stdio::piped()) .kill_on_drop(true) .spawn() @@ -37,19 +35,22 @@ async fn js_server() -> anyhow::Result<()> { line == "listening", "unexpected `tests/js/server.js` output: {line}" ); - - let client = wrpc_transport::frame::tcp::Client::from(addr); - common::assert_sync((), &client).await?; - common::assert_async((), &client).await?; - - server.start_kill().context("failed to kill `node`")?; - server.wait().await.context("failed to wait for `node`")?; - Ok(()) + Ok(server) } -#[instrument(ret)] -#[test_log::test(tokio::test(flavor = "multi_thread"))] -async fn js_client() -> anyhow::Result<()> { +/// Serve the `echo-server` world of `tests/js/wit` on `srv`, passing each +/// connection accepted on `lis` to `accept`, until `tests/js/client.js`, run +/// against `addr`, exits. +async fn serve_echo( + srv: &S, + lis: TcpListener, + addr: &str, + mut accept: impl AsyncFnMut(TcpStream, SocketAddr), +) -> anyhow::Result<()> +where + S: wrpc_transport::Serve, + S::Context: 'static, +{ mod bindings { wit_bindgen_wrpc::generate!({ world: "echo-server", @@ -72,12 +73,7 @@ async fn js_client() -> anyhow::Result<()> { } } - let lis = TcpListener::bind((Ipv6Addr::LOCALHOST, 0)) - .await - .context("failed to start TCP listener")?; - let addr = lis.local_addr().context("failed to get server address")?; - let srv = wrpc_transport::frame::Server::default(); - let invocations = bindings::serve(&srv, Echo) + let invocations = bindings::serve(srv, Echo) .await .context("failed to serve `wrpc-test:js/echo`")?; let mut invocations = stream::select_all( @@ -89,7 +85,7 @@ async fn js_client() -> anyhow::Result<()> { info!("running `tests/js/client.js`"); let mut client = pin!( Command::new("node") - .args(["--test-force-exit", "tests/js/client.js", &addr.to_string()]) + .args(["--test-force-exit", "tests/js/client.js", addr]) .kill_on_drop(true) .status() ); @@ -102,8 +98,7 @@ async fn js_client() -> anyhow::Result<()> { } res = lis.accept() => { let (stream, addr) = res.expect("failed to accept connection"); - let (rx, tx) = wrpc_transport::tcp::split(stream); - srv.accept(addr, tx, BufReader::new(rx)).await.expect("failed to accept invocation"); + accept(stream, addr).await; } Some((instance, name, invocation)) = invocations.next() => { let invocation = invocation @@ -117,3 +112,90 @@ async fn js_client() -> anyhow::Result<()> { } } } + +#[instrument(ret)] +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn js_server() -> anyhow::Result<()> { + let port = wrpc_test::free_port() + .await + .context("failed to find free port")?; + let addr = SocketAddr::from((Ipv6Addr::LOCALHOST, port)); + let mut server = start_server(&addr.to_string()).await?; + + let client = wrpc_transport::frame::tcp::Client::from(addr); + common::assert_sync((), &client).await?; + common::assert_async((), &client).await?; + + server.start_kill().context("failed to kill `node`")?; + server.wait().await.context("failed to wait for `node`")?; + Ok(()) +} + +#[instrument(ret)] +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn js_http_server() -> anyhow::Result<()> { + let port = wrpc_test::free_port() + .await + .context("failed to find free port")?; + let addr = SocketAddr::from((Ipv6Addr::LOCALHOST, port)); + let mut server = start_server(&format!("http://{addr}")).await?; + + let conn = wrpc_test::tcp_connect(addr).await?; + let (sender, conn) = http2::handshake(TokioExecutor::new(), TokioIo::new(conn)) + .await + .context("failed to establish HTTP/2 connection")?; + spawn(conn); + let client = wrpc_http::Client::from(sender); + let (mut cx, ()) = http::Request::new(()).into_parts(); + cx.uri = format!("http://{addr}/") + .parse() + .context("failed to parse request URI")?; + common::assert_sync(cx.clone(), &client).await?; + common::assert_async(cx, &client).await?; + + server.start_kill().context("failed to kill `node`")?; + server.wait().await.context("failed to wait for `node`")?; + Ok(()) +} + +#[instrument(ret)] +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn js_client() -> anyhow::Result<()> { + let lis = TcpListener::bind((Ipv6Addr::LOCALHOST, 0)) + .await + .context("failed to start TCP listener")?; + let addr = lis.local_addr().context("failed to get server address")?; + let srv = wrpc_transport::frame::Server::default(); + serve_echo(&srv, lis, &addr.to_string(), async |stream, addr| { + let (rx, tx) = wrpc_transport::tcp::split(stream); + srv.accept(addr, tx, BufReader::new(rx)) + .await + .expect("failed to accept invocation"); + }) + .await +} + +#[instrument(ret)] +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn js_http_client() -> anyhow::Result<()> { + let lis = TcpListener::bind((Ipv6Addr::LOCALHOST, 0)) + .await + .context("failed to start TCP listener")?; + let addr = lis.local_addr().context("failed to get server address")?; + let srv = wrpc_http::Server::default(); + serve_echo( + srv.0.as_ref(), + lis, + &format!("http://{addr}"), + async |stream, _| { + let svc = wrpc_http::Server::from(Arc::clone(&srv.0)); + spawn(async move { + http1::Builder::new() + .serve_connection(TokioIo::new(stream), svc) + .await + .expect("failed to serve HTTP connection"); + }); + }, + ) + .await +} diff --git a/tests/js/client.js b/tests/js/client.js index 1ef751681..fc54bd780 100644 --- a/tests/js/client.js +++ b/tests/js/client.js @@ -5,7 +5,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { invoke, t } from "../../js/src/index.js"; -import { connect } from "./tcp.js"; +import { connect } from "./transport.js"; /** * @typedef {import("../../js/src/index.js").Chan} Chan diff --git a/tests/js/server.js b/tests/js/server.js index 8dffd6e44..71366c4ad 100644 --- a/tests/js/server.js +++ b/tests/js/server.js @@ -2,7 +2,7 @@ // address given as the first argument, for the Rust clients in `tests/js.rs`. import { accept, t } from "../../js/src/index.js"; -import { listen } from "./tcp.js"; +import { listen } from "./transport.js"; /** * @typedef {import("../../js/src/index.js").InputValue} InputValue diff --git a/tests/js/tcp.js b/tests/js/tcp.js deleted file mode 100644 index 98224844a..000000000 --- a/tests/js/tcp.js +++ /dev/null @@ -1,49 +0,0 @@ -// TCP plumbing shared by the scripts `tests/js.rs` drives. - -import net from "node:net"; -import { Duplex } from "node:stream"; - -import { fromWebStreams } from "../../js/src/index.js"; - -/** - * Split a `host:port` address, such as `[::1]:7761`. - * @param {string} addr - */ -function parse(addr) { - const i = addr.lastIndexOf(":"); - return { host: addr.slice(0, i).replace(/^\[(.*)\]$/, "$1"), port: Number(addr.slice(i + 1)) }; -} - -/** - * A transport over a socket. wRPC signals the end of each direction by - * shutting down its write side, so sockets must allow half-open connections. - * @param {net.Socket} sock - */ -export const transport = (sock) => fromWebStreams(Duplex.toWeb(sock)); - -/** - * Connect to `addr`. - * @param {string} addr - * @returns {Promise} - */ -export function connect(addr) { - return new Promise((resolve, reject) => { - const sock = net.connect({ ...parse(addr), allowHalfOpen: true }, () => - resolve(transport(sock)), - ); - sock.once("error", reject); - }); -} - -/** - * Listen on `addr`, calling `f` with a transport for each connection, and - * print `listening` once ready. - * @param {string} addr - * @param {(transport: import("../../js/src/index.js").Transport) => void} f - */ -export function listen(addr, f) { - const { host, port } = parse(addr); - net - .createServer({ allowHalfOpen: true }, (sock) => f(transport(sock))) - .listen(port, host, () => console.log("listening")); -} diff --git a/tests/js/transport.js b/tests/js/transport.js new file mode 100644 index 000000000..dcbd79057 --- /dev/null +++ b/tests/js/transport.js @@ -0,0 +1,71 @@ +// TCP and HTTP plumbing shared by the scripts `tests/js.rs` drives. An +// address of the form `http://host:port` selects HTTP, any other TCP. + +import http2 from "node:http2"; +import net from "node:net"; +import { Duplex, Readable, pipeline } from "node:stream"; + +import { fromFetch, fromRequest, fromWebStreams } from "../../js/src/index.js"; + +/** + * @typedef {import("../../js/src/index.js").Transport} Transport + */ + +/** + * Split a `host:port` address, such as `[::1]:7761`. + * @param {string} addr + */ +function parse(addr) { + const i = addr.lastIndexOf(":"); + return { host: addr.slice(0, i).replace(/^\[(.*)\]$/, "$1"), port: Number(addr.slice(i + 1)) }; +} + +/** + * A transport over a socket. wRPC signals the end of each direction by + * shutting down its write side, so sockets must allow half-open connections. + * @param {net.Socket} sock + */ +export const transport = (sock) => fromWebStreams(Duplex.toWeb(sock)); + +/** + * Connect to `addr`. + * @param {string} addr + * @returns {Promise} + */ +export function connect(addr) { + if (addr.startsWith("http:")) return Promise.resolve(fromFetch(addr)); + return new Promise((resolve, reject) => { + const sock = net.connect({ ...parse(addr), allowHalfOpen: true }, () => + resolve(transport(sock)), + ); + sock.once("error", reject); + }); +} + +/** + * Listen on `addr`, calling `f` with a transport for each connection, or + * each HTTP/2 request without TLS, and print `listening` once ready. + * @param {string} addr + * @param {(transport: Transport) => void} f + */ +export function listen(addr, f) { + if (addr.startsWith("http:")) { + const { hostname, port } = new URL(addr); + http2 + .createServer((req, res) => { + const init = { method: req.method, body: Readable.toWeb(req), duplex: "half" }; + const { transport, response } = fromRequest( + new Request(new URL(req.url, addr), /** @type {RequestInit} */ (init)), + ); + f(transport); + res.writeHead(response.status); + pipeline(Readable.fromWeb(/** @type {any} */ (response.body)), res, () => {}); + }) + .listen(Number(port), hostname.replace(/^\[(.*)\]$/, "$1"), () => console.log("listening")); + return; + } + const { host, port } = parse(addr); + net + .createServer({ allowHalfOpen: true }, (sock) => f(transport(sock))) + .listen(port, host, () => console.log("listening")); +}