feat(js): add HTTP transport

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.

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<RequestInit, "method" | "body" | "signal"> & { 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<typeof fetch>[0]} input
+ * @param {FetchOptions} [options]
+ * @returns {Transport}
+ */
+export function fromFetch(input, { fetch = globalThis.fetch, buffer = false, ...init } = {}) {
+  const ctrl = new AbortController();
+  /** @type {TransformStreamDefaultController<Uint8Array> | 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<ReadableStreamDefaultReader<Uint8Array>>} */
+  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<Transport, "write" | "closeWrite">} */
+  let body;
+  if (stream === undefined) {
+    /** @type {Uint8Array[]} */
+    const chunks = [];
+    body = {
+      write: (bytes) => void chunks.push(bytes),
+      closeWrite: () => send(new Blob(/** @type {Uint8Array<ArrayBuffer>[]} */ (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),
+  };
+}

js/src/index.d.ts

@@ -332,6 +332,22 @@ export class Mux {
   closeWrite(): Promise<void>;
 }
 
+// ----- http.js --------------------------------------------------------------
+
+/** Options of {@link fromFetch}: the `fetch` request options, except for the method, body and signal. */
+export interface FetchOptions extends Omit<RequestInit, "method" | "body" | "signal"> {
+  /** 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 -----------------------------------------------------------
 
 /**

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";

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<unknown>[]} */
+    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<Uint8Array>]} */ (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",
+  });
+});