From 6fd999b337a81fc7215de581aa676a240410cdf9 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 16 Aug 2026 19:24:54 +0900 Subject: [PATCH 1/5] chore: upgrade typescript to 4.9 and typedoc Signed-off-by: kahirokunn --- package-lock.json | 89 +++++++++++++++++++++-------------------------- package.json | 4 +-- 2 files changed, 41 insertions(+), 52 deletions(-) diff --git a/package-lock.json b/package-lock.json index bbffeea7..dc25c00d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -49,8 +49,8 @@ "remark-preset-lint-recommended": "^5.0.0", "superagent": "^7.1.1", "ts-node": "^10.8.1", - "typedoc": "^0.22.11", - "typescript": "^4.3.5", + "typedoc": "^0.23.28", + "typescript": "^4.9.5", "webpack": "^5.76.0", "webpack-cli": "^4.10.0" }, @@ -1891,6 +1891,13 @@ "node": ">=6" } }, + "node_modules/ansi-sequence-parser": { + "version": "1.1.3", + "resolved": "https://registry.npmjs.org/ansi-sequence-parser/-/ansi-sequence-parser-1.1.3.tgz", + "integrity": "sha512-+fksAx9eG3Ab6LDnLs3ZqZa8KVJ/jYnX+D4Qe1azX+LFGFAXqynCQLOdLpNYN/l9e7l6hMWwZbrnctqr6eSQSw==", + "dev": true, + "license": "MIT" + }, "node_modules/ansi-styles": { "version": "4.3.0", "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-4.3.0.tgz", @@ -8863,15 +8870,16 @@ } }, "node_modules/shiki": { - "version": "0.10.1", - "resolved": "https://registry.npmjs.org/shiki/-/shiki-0.10.1.tgz", - "integrity": "sha512-VsY7QJVzU51j5o1+DguUd+6vmCmZ5v/6gYu4vyYAhzjuNQU6P/vmSy4uQaOhvje031qQMiW0d2BwgMH52vqMng==", + "version": "0.14.7", + "resolved": "https://registry.npmjs.org/shiki/-/shiki-0.14.7.tgz", + "integrity": "sha512-dNPAPrxSc87ua2sKJ3H5dQ/6ZaY8RNnaAqK+t0eG7p0Soi2ydiqbGOTaZCqaYvA/uZYfS1LJnemt3Q+mSfcPCg==", "dev": true, "license": "MIT", "dependencies": { - "jsonc-parser": "^3.0.0", - "vscode-oniguruma": "^1.6.1", - "vscode-textmate": "5.2.0" + "ansi-sequence-parser": "^1.1.0", + "jsonc-parser": "^3.2.0", + "vscode-oniguruma": "^1.7.0", + "vscode-textmate": "^8.0.0" } }, "node_modules/side-channel": { @@ -9725,76 +9733,57 @@ } }, "node_modules/typedoc": { - "version": "0.22.18", - "resolved": "https://registry.npmjs.org/typedoc/-/typedoc-0.22.18.tgz", - "integrity": "sha512-NK9RlLhRUGMvc6Rw5USEYgT4DVAUFk7IF7Q6MYfpJ88KnTZP7EneEa4RcP+tX1auAcz7QT1Iy0bUSZBYYHdoyA==", + "version": "0.23.28", + "resolved": "https://registry.npmjs.org/typedoc/-/typedoc-0.23.28.tgz", + "integrity": "sha512-9x1+hZWTHEQcGoP7qFmlo4unUoVJLB0H/8vfO/7wqTnZxg4kPuji9y3uRzEu0ZKez63OJAUmiGhUrtukC6Uj3w==", "dev": true, "license": "Apache-2.0", "dependencies": { - "glob": "^8.0.3", "lunr": "^2.3.9", - "marked": "^4.0.16", - "minimatch": "^5.1.0", - "shiki": "^0.10.1" + "marked": "^4.2.12", + "minimatch": "^7.1.3", + "shiki": "^0.14.1" }, "bin": { "typedoc": "bin/typedoc" }, "engines": { - "node": ">= 12.10.0" + "node": ">= 14.14" }, "peerDependencies": { - "typescript": "4.0.x || 4.1.x || 4.2.x || 4.3.x || 4.4.x || 4.5.x || 4.6.x || 4.7.x" + "typescript": "4.6.x || 4.7.x || 4.8.x || 4.9.x || 5.0.x" } }, "node_modules/typedoc/node_modules/brace-expansion": { - "version": "2.0.1", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.0.1.tgz", - "integrity": "sha512-XnAIvQ8eM+kC6aULx6wuQiwVsnzsi9d3WxzV3FpWTGA19F621kwdbsAcFKXgKUHZWsy+mY6iL1sHTxWEFCytDA==", + "version": "2.1.4", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.1.4.tgz", + "integrity": "sha512-hGfVzPxthbf3+2yjg/RBs60cB0FhqBS/zvdV/4wn4/BmN0bNMMHPc4V/BbFieqf1TKAGGAHnY4eSjajCl0f2Xg==", "dev": true, "license": "MIT", "dependencies": { "balanced-match": "^1.0.0" } }, - "node_modules/typedoc/node_modules/glob": { - "version": "8.1.0", - "resolved": "https://registry.npmjs.org/glob/-/glob-8.1.0.tgz", - "integrity": "sha512-r8hpEjiQEYlF2QU0df3dS+nxxSIreXQS1qRhMJM0Q5NDdR386C7jb7Hwwod8Fgiuex+k0GFjgft18yvxm5XoCQ==", - "deprecated": "Glob versions prior to v9 are no longer supported", + "node_modules/typedoc/node_modules/minimatch": { + "version": "7.4.9", + "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-7.4.9.tgz", + "integrity": "sha512-Brg/fp/iAVDOQoHxkuN5bEYhyQlZhxddI78yWsCbeEwTHXQjlNLtiJDUsp1GIptVqMI7/gkJMz4vVAc01mpoBw==", "dev": true, "license": "ISC", "dependencies": { - "fs.realpath": "^1.0.0", - "inflight": "^1.0.4", - "inherits": "2", - "minimatch": "^5.0.1", - "once": "^1.3.0" + "brace-expansion": "^2.0.2" }, "engines": { - "node": ">=12" + "node": ">=10" }, "funding": { "url": "https://github.com/sponsors/isaacs" } }, - "node_modules/typedoc/node_modules/minimatch": { - "version": "5.1.6", - "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-5.1.6.tgz", - "integrity": "sha512-lKwV/1brpG6mBUFHtb7NUmtABCb2WZZmm2wNiOA5hAb8VdCS4B3dtMWyvcoViccwAW/COERjXLt0zP1zXUN26g==", - "dev": true, - "license": "ISC", - "dependencies": { - "brace-expansion": "^2.0.1" - }, - "engines": { - "node": ">=10" - } - }, "node_modules/typescript": { - "version": "4.7.4", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-4.7.4.tgz", - "integrity": "sha512-C0WQT0gezHuw6AdY1M2jxUO83Rjf0HP7Sk1DtXj6j1EwkQNZrHAg2XPWlq62oqEhYvONq5pkC2Y9oPljWToLmQ==", + "version": "4.9.5", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-4.9.5.tgz", + "integrity": "sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==", "dev": true, "license": "Apache-2.0", "bin": { @@ -10402,9 +10391,9 @@ "license": "MIT" }, "node_modules/vscode-textmate": { - "version": "5.2.0", - "resolved": "https://registry.npmjs.org/vscode-textmate/-/vscode-textmate-5.2.0.tgz", - "integrity": "sha512-Uw5ooOQxRASHgu6C7GVvUxisKXfSgW4oFlO+aa+PAkgmH89O3CXxEEzNRNtHSqtXFTl0nAC1uYj0GMSH27uwtQ==", + "version": "8.0.0", + "resolved": "https://registry.npmjs.org/vscode-textmate/-/vscode-textmate-8.0.0.tgz", + "integrity": "sha512-AFbieoL7a5LMqcnOF04ji+rpXadgOXnZsxQr//r83kLPr7biP7am3g9zbaZIaBGwBRWeSvoMD4mgPdX3e4NWBg==", "dev": true, "license": "MIT" }, diff --git a/package.json b/package.json index 214d34a5..1729988e 100644 --- a/package.json +++ b/package.json @@ -150,8 +150,8 @@ "remark-preset-lint-recommended": "^5.0.0", "superagent": "^7.1.1", "ts-node": "^10.8.1", - "typedoc": "^0.22.11", - "typescript": "^4.3.5", + "typedoc": "^0.23.28", + "typescript": "^4.9.5", "webpack": "^5.76.0", "webpack-cli": "^4.10.0" }, From 282927466a60cd5d2321d18ffb07a1cee54f20bc Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 16 Aug 2026 23:18:43 +0900 Subject: [PATCH 2/5] chore: upgrade @types/node to match the supported Node.js versions Signed-off-by: kahirokunn --- package-lock.json | 20 +++++++++++++++----- package.json | 2 +- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/package-lock.json b/package-lock.json index dc25c00d..9fcdac6d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -23,7 +23,7 @@ "@types/got": "^9.6.11", "@types/json-bigint": "^1.0.1", "@types/mocha": "^7.0.2", - "@types/node": "^14.14.10", + "@types/node": "~22.0.0", "@types/superagent": "^4.1.10", "@types/uuid": "^8.3.4", "@typescript-eslint/eslint-plugin": "^4.29.0", @@ -1295,11 +1295,14 @@ "license": "MIT" }, "node_modules/@types/node": { - "version": "14.18.63", - "resolved": "https://registry.npmjs.org/@types/node/-/node-14.18.63.tgz", - "integrity": "sha512-fAtCfv4jJg+ExtXhvCkCqUKZ+4ok/JQk01qDKhL5BDDoS3AxKXhV5/MAVUZyQnSEd2GT92fkgZl0pz0Q0AzcIQ==", + "version": "22.0.3", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.0.3.tgz", + "integrity": "sha512-/e0NZtK2gs6Vk2DoyrXSZZ4AlamqTkx0CcKx1Aq8/P4ITlRgU9OtVf5fl+LXkWWJce1M89pkSlH6lJJEnK7bQA==", "dev": true, - "license": "MIT" + "license": "MIT", + "dependencies": { + "undici-types": "~6.11.1" + } }, "node_modules/@types/responselike": { "version": "1.0.3", @@ -9813,6 +9816,13 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/undici-types": { + "version": "6.11.1", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.11.1.tgz", + "integrity": "sha512-mIDEX2ek50x0OlRgxryxsenE5XaQD4on5U2inY7RApK3SOJpofyw7uW2AyfMKkhAxXIceo2DeWGVGwyvng1GNQ==", + "dev": true, + "license": "MIT" + }, "node_modules/unified": { "version": "10.1.2", "resolved": "https://registry.npmjs.org/unified/-/unified-10.1.2.tgz", diff --git a/package.json b/package.json index 1729988e..17b31a16 100644 --- a/package.json +++ b/package.json @@ -124,7 +124,7 @@ "@types/got": "^9.6.11", "@types/json-bigint": "^1.0.1", "@types/mocha": "^7.0.2", - "@types/node": "^14.14.10", + "@types/node": "~22.0.0", "@types/superagent": "^4.1.10", "@types/uuid": "^8.3.4", "@typescript-eslint/eslint-plugin": "^4.29.0", From c33b7b527fa7bf25f757c4970f7a7f52e484fc5c Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:19:18 +0900 Subject: [PATCH 3/5] feat!: use Fetch API for HTTP transport Use the platform HTTP client in Node.js and browsers without importing Node-only http and https modules into the SDK's public entry point. Preserve CloudEvent serialization, URL credentials and per-send headers, and expose transport-wide headers and Fetch request options. Keep the transport result inferred through emitterFor(). HTTP status codes still resolve at this point; delivery failure handling is a separate change. BREAKING CHANGE: httpTransport() resolves with { response, body }, where response is the native Fetch Response, instead of { body, headers }. Node.js global HTTP agents no longer configure these requests; a Fetch dispatcher configures proxies, client certificates and custom trust. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 65 ++++ README.md | 25 +- src/index.ts | 17 +- src/message/index.ts | 2 +- src/transport/emitter.ts | 27 +- src/transport/http/index.ts | 247 ++++++++++--- test/integration/emitter_factory_test.ts | 19 +- test/integration/http_transport_test.ts | 424 +++++++++++++++++++++++ webpack.config.js | 4 +- 9 files changed, 743 insertions(+), 87 deletions(-) create mode 100644 test/integration/http_transport_test.ts diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index 4370a732..d33bd8ae 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -100,3 +100,68 @@ new CloudEvent({ type, source, data }).emit(); // You can also have several listener to send the event to several endpoint ``` + +### Upgrading From 10.x to 11.0 + +#### HTTP Transport + +`httpTransport()` uses Fetch and resolves with `{ response, body }`, where +`response` is the native Fetch `Response` and `body` is its text, instead of +`{ body, headers }`. Access headers through `response.headers.get(name)` and +the status through `response.status`. + +```js +const { response, body } = await emit(event); +console.log(response.status, response.headers.get("x-request-id"), body); +``` + +#### Proxies, Custom CAs, mTLS and Connection Pooling + +The old transport went through Node.js `http.request()` or `https.request()`, so a proxy +library or connection options set on `http.globalAgent` or `https.globalAgent` reached +the corresponding requests. Fetch does not use those agents; on Node.js the equivalent +is a dispatcher from [undici](https://www.npmjs.com/package/undici), which `fetchOptions` +passes through: + +```js +import { emitterFor, httpTransport } from "cloudevents"; +import { ProxyAgent } from "undici"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { + fetchOptions: { dispatcher: new ProxyAgent("http://proxy.example.com:3128") }, +})); +``` + +For example, an order service that presents a client certificate to an events receiver +and trusts the receiver's private CA can use an `Agent`: + +```js +import { readFileSync } from "node:fs"; +import { emitterFor, httpTransport } from "cloudevents"; +import { Agent } from "undici"; + +const dispatcher = new Agent({ + connect: { + cert: readFileSync("./certs/order-service-client.pem"), + key: readFileSync("./certs/order-service-client-key.pem"), + ca: readFileSync("./certs/events-receiver-ca.pem"), + }, +}); + +const emit = emitterFor(httpTransport("https://events.example.com/orders", { + fetchOptions: { dispatcher }, +})); +``` + +The `cert` and `key` identify the client when the receiver requires mTLS. The `ca` +controls which server certificates the client trusts, so it is needed here because the +receiver uses a private CA; it is separate from client authentication. An `Agent` also +tunes connection pooling, and `undici.setGlobalDispatcher()` applies to every send rather +than one transport. In the browser the platform handles proxies, certificates and trust, +so `dispatcher` is a Node.js-only option. + +`fetchOptions` takes the Fetch options of whichever declarations type `fetch` in your +project, so TypeScript accepts `dispatcher` only where those are undici's. Against the +DOM `RequestInit` most projects compile with, pass it as +`fetchOptions: { dispatcher } as FetchRequestInit` or reach for +`setGlobalDispatcher()` instead. diff --git a/README.md b/README.md index b40b8de3..60a6874e 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,8 @@ app.post("/", (req, res) => { #### Emitting Events -The easiest way to send events is to use the built-in HTTP emitter. +The easiest way to send events is to use the built-in HTTP emitter, which sends +them with the Fetch API. ```js const { httpTransport, emitterFor, CloudEvent } = require("cloudevents"); @@ -57,10 +58,28 @@ const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); // Create a new CloudEvent const ce = new CloudEvent({ type, source, data }); -// Send it to the endpoint - encoded as HTTP binary by default -emit(ce); +async function main() { + // Send it to the endpoint - encoded as HTTP binary by default + const { response, body } = await emit(ce); + console.log(response.status, body); +} + +main().catch(console.error); ``` +The transport resolves with the native Fetch `Response` and its body as text +for every HTTP status. It sends each event once. Redirects are not followed +unless `fetchOptions: { redirect: "follow" }` enables them; Fetch preserves the +CloudEvent POST and its body only for `307` and `308`. + +`httpTransport()` accepts transport-wide `headers` and other Fetch options under +`fetchOptions`. Headers passed to `emit()` apply to that event only and take +precedence over transport and CloudEvent headers. URL credentials become an +authorization header unless the caller overrides it. + +See the [API transition guide](./API_TRANSITION_GUIDE.md) for response access and +Node.js proxy, certificate and connection settings. + If you prefer to use another transport mechanism for sending events over HTTP, you can use the `HTTP` binding to create a `Message` which has properties for `headers` and `body`, allowing greater flexibility diff --git a/src/index.ts b/src/index.ts index ab0148db..3c3d5bcd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,8 +7,13 @@ import { CloudEvent, V1, V03 } from "./event/cloudevent"; import { ValidationError } from "./event/validation"; import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces"; -import { Options, TransportFunction, EmitterFunction, emitterFor, Emitter } from "./transport/emitter"; -import { httpTransport } from "./transport/http"; +import { + Options, TransportFunction, EmitterFunction, emitterFor, Emitter, +} from "./transport/emitter"; +import { + FetchHeadersInit, FetchRequestInit, HTTPTransportFunction, HTTPTransportOptions, + HTTPTransportResponse, httpTransport, +} from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -26,6 +31,7 @@ export { Kafka, MQTT, MQTTMessageFactory, + // From transport emitterFor, httpTransport, Emitter, @@ -48,5 +54,10 @@ export type { // From transport TransportFunction, EmitterFunction, - Options + Options, + FetchHeadersInit, + FetchRequestInit, + HTTPTransportFunction, + HTTPTransportOptions, + HTTPTransportResponse }; diff --git a/src/message/index.ts b/src/message/index.ts index 4eda7320..bee6ab06 100644 --- a/src/message/index.ts +++ b/src/message/index.ts @@ -3,7 +3,7 @@ SPDX-License-Identifier: Apache-2.0 */ -import { IncomingHttpHeaders } from "http"; +import type { IncomingHttpHeaders } from "http"; import { CloudEventV1 } from ".."; // reexport the protocol bindings diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 2a8b604c..7ddaf7d3 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -20,20 +20,22 @@ export interface Options { * EmitterFunction is an invokable interface returned by {@linkcode emitterFor}. * Invoke an EmitterFunction with a CloudEvent and optional transport * options to send the event as a Message across supported transports. + * TResult is whatever the underlying {@linkcode TransportFunction} resolves with. * @interface */ -export interface EmitterFunction { - (event: CloudEvent, options?: Options): Promise; +export interface EmitterFunction { + (event: CloudEvent, options?: TOptions): Promise; } /** * TransportFunction is an invokable interface provided to the emitterFactory. * A TransportFunction's responsiblity is to send a JSON encoded event Message - * across the wire. + * across the wire. TResult is the value it resolves with, e.g. an HTTP client's + * response, or void for a transport that has nothing to hand back. * @interface */ -export interface TransportFunction { - (message: Message, options?: Options): Promise; +export interface TransportFunction { + (message: Message, options?: TOptions): Promise; } const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; @@ -51,19 +53,24 @@ const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; * @param {Mode} options.mode the encoding mode (Mode.BINARY or Mode.STRUCTURED) * @returns {EmitterFunction} an EmitterFunction to send events with */ -export function emitterFor(fn: TransportFunction, options = emitterDefaults): EmitterFunction { +export function emitterFor( + fn: TransportFunction, options?: Options, +): EmitterFunction; +export function emitterFor( + fn: TransportFunction, options: Options = emitterDefaults, +): EmitterFunction { if (!fn) { throw new TypeError("A TransportFunction is required"); } const { binding, mode }: any = { ...emitterDefaults, ...options }; - return function emit(event: CloudEvent, opts?: Options): Promise { - opts = opts || {}; + return function emit(event: CloudEvent, opts?: TOptions): Promise { + const transportOptions = opts ?? {} as TOptions; switch (mode) { case Mode.BINARY: - return fn(binding.binary(event), opts); + return fn(binding.binary(event), transportOptions); case Mode.STRUCTURED: - return fn(binding.structured(event), opts); + return fn(binding.structured(event), transportOptions); default: throw new TypeError(`Unexpected transport mode: ${mode}`); } diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 2ac2062c..1b5e9604 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -3,61 +3,204 @@ SPDX-License-Identifier: Apache-2.0 */ -import { Socket } from "net"; -import http, { OutgoingHttpHeaders } from "http"; -import https, { RequestOptions } from "https"; - -import { Message, Options } from "../.."; -import { TransportFunction } from "../emitter"; - -/** - * httpTransport provides a simple HTTP Transport function, which can send a CloudEvent, - * encoded as a Message to the endpoint. The returned function can be used with emitterFor() - * to provide an event emitter, for example: - * - * const emitter = emitterFor(httpTransport("http://example.com")); - * emitter.emit(myCloudEvent) - * .then(resp => console.log(resp)); - * +import { Headers as CloudEventHeaders, Message } from "../../message"; +import { Options, TransportFunction } from "../emitter"; + +/** The request options accepted by the Fetch implementation in the current environment */ +export type FetchRequestInit = NonNullable[1]>; + +/** The header forms accepted by the Fetch implementation in the current environment */ +export type FetchHeadersInit = NonNullable; + +/** Options applied to every request sent by an HTTP transport */ +export interface HTTPTransportOptions { + /** HTTP headers applied to every request, per-send headers take precedence */ + headers?: FetchHeadersInit; + /** Fetch options applied to every request, apart from the ones this transport controls */ + fetchOptions?: Omit; +} + +/** The native Fetch response and its body, read as text */ +export interface HTTPTransportResponse { + response: Response; + body: string; +} + +/** A transport that returns an HTTP response */ +export type HTTPTransportFunction = TransportFunction; + +/** + * Send an HTTP-bound CloudEvent with Fetch. The transport sends once and resolves with + * the native response and its body as text, for every HTTP status. Redirects are not + * followed unless fetchOptions.redirect enables them. Credentials in the sink URL become + * an authorization header, overridden by transport or per-send headers. + * + * @param {string|URL} sink the destination endpoint for the event + * @param {HTTPTransportOptions} options headers and Fetch request options + * @returns {HTTPTransportFunction} a function that sends CloudEvents to the sink + */ +export function httpTransport( + sink: string | URL, options: HTTPTransportOptions = {}, +): HTTPTransportFunction { + const url = validateHTTPURL(sink); + // fetch refuses a URL with credentials + const sinkHeaders = credentialsFrom(url); + const transportHeaders = headersFrom(options.headers); + const fetchOptions: FetchRequestInit = { + ...options.fetchOptions, + // fetch reads an explicitly undefined redirect as absent, so the default comes last + redirect: options.fetchOptions?.redirect ?? "manual", + method: "POST", + }; + + return async (message: Message, sendOptions?: Options): Promise => { + const headers = requestHeaders( + sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), + ); + const response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + }); + return { response, body: await response.text() }; + }; +} + +/** + * Check that the sink is a URL this transport knows how to POST to + * * @param {string|URL} sink the destination endpoint for the event - * @returns {TransportFunction} a function which can be used to send CloudEvents to _sink_ + * @returns {URL} the parsed sink */ -export function httpTransport(sink: string | URL): TransportFunction { +function validateHTTPURL(sink: string | URL): URL { const url = new URL(sink); - let base: any; - if (url.protocol === "https:") { - base = https; - } else if (url.protocol === "http:") { - base = http; - } else { + if (url.protocol !== "http:" && url.protocol !== "https:") { throw new TypeError(`unsupported protocol ${url.protocol}`); } - return function(message: Message, options?: Options): Promise { - return new Promise((resolve, reject) => { - options = { ...options }; - - // TODO: Callers should be able to set any Node.js RequestOptions - const opts: RequestOptions = { - method: "POST", - headers: {...message.headers, ...options.headers as OutgoingHttpHeaders}, - }; - try { - const response = { - body: "", - headers: {}, - }; - const req = base.request(url, opts, (res: Socket) => { - res.setEncoding("utf-8"); - response.headers = (res as any).headers; - res.on("data", (chunk) => response.body += chunk); - res.on("end", () => { resolve(response); }); - }); - req.on("error", reject); - req.write(message.body); - req.end(); - } catch (err) { - reject(err); - } - }); - }; + return url; +} + +/** + * Merge the headers for a single request + * + * @param {Headers} sinkHeaders the headers built from the sink URL, if it carried credentials + * @param {CloudEventHeaders} messageHeaders the headers of the event Message + * @param {Headers} transportHeaders the headers configured on this transport, if any + * @param {Headers} sendHeaders the headers supplied with this send, if any + * @returns {Headers} the headers to send + */ +function requestHeaders( + sinkHeaders: Headers | undefined, messageHeaders: CloudEventHeaders, + transportHeaders?: Headers, sendHeaders?: Headers, +): Headers { + const headers = new Headers(); + // lowest precedence first: sink credentials, binding, transport, this send + setHeaders(headers, sinkHeaders); + setHeaders(headers, messageHeaders); + setHeaders(headers, transportHeaders); + setHeaders(headers, sendHeaders); + return headers; +} + +/** + * Move the credentials of a sink URL into the header the old Node.js transport built, since + * Fetch refuses a URL that carries them + * + * @param {URL} url the sink URL, whose credentials are removed + * @returns {Headers|undefined} the authorization header, or undefined for a URL without credentials + */ +function credentialsFrom(url: URL): Headers | undefined { + if (url.username === "" && url.password === "") { + return undefined; + } + const userinfo = `${decodeCredential(url.username)}:${decodeCredential(url.password)}`; + url.username = ""; + url.password = ""; + return new Headers({ authorization: `Basic ${base64(userinfo)}` }); +} + +/** + * Read one percent-encoded credential of a sink URL, as urlToHttpOptions() did for the old + * transport + * + * @param {string} value the credential as the URL holds it + * @returns {string} the decoded credential + */ +function decodeCredential(value: string): string { + try { + return decodeURIComponent(value); + } catch { + throw new TypeError("sink credentials must be percent-encoded values"); + } +} + +/** + * Encode credentials as UTF-8 bytes, the way RFC 7617 and the old transport did, with APIs the + * browser bundle can use as well + * + * @param {string} value the credentials to encode + * @returns {string} the base64 form + */ +function base64(value: string): string { + const utf8 = new TextEncoder().encode(value); + return btoa(Array.from(utf8, (byte) => String.fromCharCode(byte)).join("")); +} + +/** + * Check the headers a caller supplied, either for this transport or for a single send + * + * Fetch does the conversion, so every form its `HeadersInit` accepts arrives intact, a + * `Headers` from another realm included. That leaves an untyped value to Fetch's own coercion, + * which is why a caller's array joins on "," while the array of a Message, normalized by + * {@linkcode headerValueFrom}, joins on ", " + * + * @param {unknown} source the headers supplied by the caller, if any + * @returns {Headers|undefined} the headers, or undefined when none were supplied + */ +function headersFrom(source: unknown): Headers | undefined { + if (source === undefined || source === null) { + return undefined; + } + return new Headers(source as FetchHeadersInit); +} + +/** + * Copy headers onto a target, replacing any that are already there + * + * @param {Headers} target the headers being built + * @param {CloudEventHeaders|Headers} source the headers to copy, undefined ones are ignored + * @returns {void} + */ +function setHeaders(target: Headers, source?: CloudEventHeaders | Headers): void { + if (source === undefined) { + return; + } + if (source instanceof Headers) { + source.forEach((value, name) => target.set(name, value)); + return; + } + for (const [name, value] of Object.entries(source as Record)) { + const headerValue = headerValueFrom(value); + if (headerValue !== undefined) { + target.set(name, headerValue); + } + } +} + +/** + * Read a header value, whatever form it arrived in + * + * @param {unknown} value the value to read + * @returns {string|undefined} the value to send, undefined for a header which is not to be sent + */ +function headerValueFrom(value: unknown): string | undefined { + if (value === undefined || value === null) { + return undefined; + } + // several values join; an absent one is skipped + if (Array.isArray(value)) { + const values = value.filter((item) => item !== undefined && item !== null); + return values.length > 0 ? values.join(", ") : undefined; + } + return String(value); } diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts index 568f6ccc..e73b10e9 100644 --- a/test/integration/emitter_factory_test.ts +++ b/test/integration/emitter_factory_test.ts @@ -11,7 +11,7 @@ import request from "superagent"; import got from "got"; import CONSTANTS from "../../src/constants"; -import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor, httpTransport } +import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } from "../../src"; const DEFAULT_CE_CONTENT_TYPE = CONSTANTS.DEFAULT_CE_CONTENT_TYPE; @@ -101,8 +101,8 @@ describe("emitterFor() defaults", () => { }); }); -function setupMock(uri: string) { - nock(uri) +function setupMock() { + nock(sink) .post("/") .reply(function (uri: string, body: nock.Body) { // return the request body and the headers so they can be @@ -116,18 +116,7 @@ function setupMock(uri: string) { } describe("HTTP Transport Binding for emitterFactory", () => { - beforeEach(() => { setupMock(sink); }); - - describe("HTTPS builtin", () => { - testEmitterBinary(httpTransport(sink), "body"); - }); - - describe("HTTP builtin", () => { - setupMock("http://cloudevents.io"); - testEmitterBinary(httpTransport("http://cloudevents.io"), "body"); - setupMock("http://cloudevents.io"); - testEmitterStructured(httpTransport("http://cloudevents.io"), "body"); - }); + beforeEach(() => { setupMock(); }); describe("Axios", () => { testEmitterBinary(axiosEmitter, "data"); diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts new file mode 100644 index 00000000..cf140d43 --- /dev/null +++ b/test/integration/http_transport_test.ts @@ -0,0 +1,424 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import "mocha"; +import { rejects } from "assert"; +import { expect } from "chai"; +import { createServer, IncomingMessage, ServerResponse } from "http"; +import { AddressInfo } from "net"; +import { json, text } from "stream/consumers"; +import { runInNewContext } from "vm"; + +import { + CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, + Message, Mode, emitterFor, httpTransport, +} from "../../src"; +import { assertStructured, fixture } from "./emitter_factory_test"; + +type Handler = (request: IncomingMessage, response: ServerResponse) => void | Promise; + +// nock cannot intercept fetch, so these tests use a local server + +// a reserved port nothing serves +const UNREACHABLE_URL = "http://127.0.0.1:1"; + +// a sink which accepts every event, for tests that assert on the request rather than the response +const acceptEvent: Handler = (_request, response) => { + response.writeHead(204); + response.end(); +}; + +describe("Built-in HTTP transport", () => { + it("Sends a binary event and resolves with the response for 2xx", async () => { + let receivedBody: Record | undefined; + let receivedHeaders: IncomingMessage["headers"] | undefined; + + await withServer(async (request, response) => { + receivedHeaders = request.headers; + receivedBody = await json(request) as Record; + response.writeHead(202, { "x-request-id": "receipt-7" }); + response.end("accepted"); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/events`)); + const { response, body } = await emit(fixture, { headers: { "x-request-id": "order-123" } }); + + expect(response.status).to.equal(202); + expect(response.headers.get("x-request-id")).to.equal("receipt-7"); + expect(response.bodyUsed).to.equal(true); + expect(body).to.equal("accepted"); + expect(receivedHeaders?.["x-request-id"]).to.equal("order-123"); + expect(receivedHeaders?.["content-type"]).to.equal(CONSTANTS.DEFAULT_CONTENT_TYPE); + expect(receivedHeaders?.["ce-id"]).to.equal(fixture.id); + expect(receivedHeaders?.["ce-specversion"]).to.equal(fixture.specversion); + expect(receivedHeaders?.["ce-type"]).to.equal(fixture.type); + expect(receivedHeaders?.["ce-source"]).to.equal(fixture.source); + // the extensions of the fixture become ce-* headers + expect(receivedHeaders?.["ce-lunch"]).to.equal(fixture.lunch); + expect(receivedHeaders?.["ce-supper"]).to.equal(fixture.supper); + // an extension holding an object has no binary mode form; the old http.request + // transport sent the same String() of it, and only nock kept the object itself + expect(receivedHeaders?.["ce-snack"]).to.equal("[object Object]"); + expect(receivedBody?.lunchBreak).to.equal("noon"); + }); + }); + + it("Resolves with an empty body when the sink sends none", async () => { + await withServer(acceptEvent, async (url) => { + const emit = emitterFor(httpTransport(url)); + const { response, body } = await emit(fixture); + + expect(response.status).to.equal(204); + expect(body).to.equal(""); + }); + }); + + it("Sends to a sink given as a URL", async () => { + let receivedPath: string | undefined; + + await withServer((request, response) => { + receivedPath = request.url; + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(new URL(`${url}/events`))); + const { response, body } = await emit(fixture); + + expect(response.status).to.equal(204); + expect(body).to.equal(""); + expect(receivedPath).to.equal("/events"); + }); + }); + + it("Sends a structured event", async () => { + let received: Record | undefined; + + await withServer(async (request, response) => { + received = { + ...await json(request) as Record, + ...request.headers, + }; + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(url), { mode: Mode.STRUCTURED }); + await emit(fixture); + assertStructured(received as Record>); + }); + }); + + it("Keeps every value of a repeated response header", async () => { + await withServer((_request, response) => { + response.writeHead(202, { + "set-cookie": ["session=abc; Path=/", "tenant=store-42; Path=/"], + }); + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const { response } = await emit(fixture); + const headers = response.headers as Headers & { getSetCookie(): string[] }; + + expect(headers.getSetCookie()).to.deep.equal([ + "session=abc; Path=/", + "tenant=store-42; Path=/", + ]); + }); + }); + + it("Applies transport headers and lets per-send headers override them", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url, { + headers: new Headers({ + authorization: "Bearer transport-token", + "x-tenant-id": "store-42", + }), + })); + + await emit(fixture); + await emit(fixture, { + headers: { "x-tenant-id": "store-99", "x-request-id": "order-123" }, + }); + + expect(receivedHeaders[0]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[0]?.["x-tenant-id"]).to.equal("store-42"); + expect(receivedHeaders[1]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[1]?.["x-tenant-id"]).to.equal("store-99"); + expect(receivedHeaders[1]?.["x-request-id"]).to.equal("order-123"); + // a header neither layer set keeps the value the binding produced + expect(receivedHeaders[1]?.["ce-type"]).to.equal(fixture.type); + }); + }); + + it("Sends credentials from the sink URL as an authorization header", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const { host } = new URL(url); + const sendAs = (credentials: string) => emitterFor(httpTransport(`http://${credentials}@${host}`))(fixture); + + await sendAs("us%40er:p%40ss"); + await sendAs("user:p%C3%A4ss"); + await sendAs("user"); + // anything the caller sets wins + await emitterFor(httpTransport(`http://user:pass@${host}`, { + headers: { authorization: "Bearer transport-token" }, + }))(fixture); + await emitterFor(httpTransport(`http://user:pass@${host}`))(fixture, { + headers: { authorization: "Bearer per-send-token" }, + }); + + // the credentials are percent-decoded, as the old transport decoded them + expect(receivedHeaders[0]?.authorization).to.equal("Basic dXNAZXI6cEBzcw=="); + // non-ASCII credentials are encoded as UTF-8 + expect(receivedHeaders[1]?.authorization).to.equal("Basic dXNlcjpww6Rzcw=="); + // a URL may carry a user without a password + expect(receivedHeaders[2]?.authorization).to.equal("Basic dXNlcjo="); + expect(receivedHeaders[3]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[4]?.authorization).to.equal("Bearer per-send-token"); + }); + }); + + it("Accepts Fetch header forms without losing their contents", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + const foreignHeaders = withForeignInterfacePrototype(new Headers({ "x-realm": "foreign" })); + expect(foreignHeaders).not.to.be.instanceOf(Headers); + const headerForms: Array<{ headers: FetchHeadersInit; name: string; value: string }> = [ + { headers: new Headers({ "x-headers": "native" }), name: "x-headers", value: "native" }, + { + headers: [["x-tuple", "store-42"], ["x-tuple", "store-99"]], + name: "x-tuple", + value: "store-42, store-99", + }, + { + headers: runInNewContext("({ 'x-record': 'another-realm' })") as FetchHeadersInit, + name: "x-record", + value: "another-realm", + }, + { headers: foreignHeaders, name: "x-realm", value: "foreign" }, + { + // Map is accepted by native Headers at runtime, although it is not in TypeScript's HeadersInit + headers: new Map([["x-map", "native"]]) as unknown as FetchHeadersInit, + name: "x-map", + value: "native", + }, + ]; + + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url)); + + for (const { headers } of headerForms) { + await emit(fixture, { headers }); + } + + headerForms.forEach(({ name, value }, index) => { + expect(receivedHeaders[index]?.[name]).to.equal(value); + }); + }); + }); + + it("Reads Message headers the same way as caller headers", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const send = httpTransport(url); + const message: Message = { + headers: { + "content-type": "application/json", + "x-tenant-id": null as unknown as string, + "x-request-id": ["order-123", "order-456"], + }, + body: `{"lunchBreak":"noon"}`, + }; + + await send(message); + await send({ + headers: new Headers({ "x-request-id": "native-headers" }) as unknown as CloudEventHeaders, + body: "event", + }); + + // a null value is skipped rather than sent as the string "null" + expect(receivedHeaders[0]?.["x-tenant-id"]).to.equal(undefined); + expect(receivedHeaders[0]?.["x-request-id"]).to.equal("order-123, order-456"); + expect(receivedHeaders[1]?.["x-request-id"]).to.equal("native-headers"); + }); + }); + + it("Passes standard Fetch options to fetch", async () => { + let receivedOptions: FetchRequestInit | undefined; + + await withFetch(async (_input, options) => { + receivedOptions = options; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com/orders", { + fetchOptions: { + cache: "no-store", + credentials: "include", + redirect: "error", + }, + })); + await emit(fixture); + }); + + expect(receivedOptions?.cache).to.equal("no-store"); + expect(receivedOptions?.credentials).to.equal("include"); + expect(receivedOptions?.redirect).to.equal("error"); + expect(receivedOptions?.method).to.equal("POST"); + }); + + it("Keeps the manual redirect when fetchOptions leaves redirect undefined", async () => { + let receivedOptions: FetchRequestInit | undefined; + + await withFetch(async (_input, options) => { + receivedOptions = options; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com/orders", { + // fetch reads an explicitly undefined redirect as absent, which would follow redirects + fetchOptions: { redirect: undefined }, + })); + await emit(fixture); + }); + + expect(receivedOptions?.redirect).to.equal("manual"); + }); + + // Fetch keeps POST only for 307 and 308, the others become a bodyless GET + for (const { statusCode, method, keepsBody } of [ + { statusCode: 301, method: "GET", keepsBody: false }, + { statusCode: 302, method: "GET", keepsBody: false }, + { statusCode: 303, method: "GET", keepsBody: false }, + { statusCode: 307, method: "POST", keepsBody: true }, + { statusCode: 308, method: "POST", keepsBody: true }, + ]) { + it(`Follows ${statusCode} with Fetch semantics when redirect is "follow"`, async () => { + let originalBody = ""; + let redirectedBody = "not received"; + let redirectedMethod: string | undefined; + + await withServer(async (request, response) => { + if (request.url === "/target") { + redirectedMethod = request.method; + redirectedBody = await text(request); + response.writeHead(204); + } else { + originalBody = await text(request); + response.writeHead(statusCode, { location: "/target" }); + } + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/start`, { + fetchOptions: { redirect: "follow" }, + })); + const { response, body } = await emit(fixture); + + expect(redirectedMethod).to.equal(method); + expect(redirectedBody).to.equal(keepsBody ? originalBody : ""); + expect(response.url).to.equal(`${url}/target`); + expect(body).to.equal(""); + }); + }); + } + + for (const statusCode of [400, 503]) { + it(`Resolves a ${statusCode} response without retrying it`, async () => { + let requestCount = 0; + await withServer((_request, response) => { + requestCount++; + response.writeHead(statusCode); + response.end("rejected"); + }, async (url) => { + const result = await emitterFor(httpTransport(url))(fixture); + expect(result.response.status).to.equal(statusCode); + expect(result.body).to.equal("rejected"); + expect(requestCount).to.equal(1); + }); + }); + } + + it("Rejects a per-send header that is not a valid header", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + // a TypeError, since nothing was sent + await rejects(emit(fixture, { headers: { "invalid header": "value" } }), TypeError); + }); + + it("Uses native Fetch coercion for untyped header values", async () => { + const legacyHeaders: CloudEventHeaders = { "x-tenant-id": ["store-42", "store-99"] }; + + await withServer((request, response) => { + expect(request.headers["x-tenant-id"]).to.equal("store-42,store-99"); + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + await emit(fixture, { + + headers: legacyHeaders, + }); + }); + }); + + it("Validates the sink and transport options when the transport is created", () => { + expect(() => httpTransport("not a URL")).to.throw(TypeError); + expect(() => httpTransport("ftp://events.example.com")).to.throw(TypeError, "unsupported protocol ftp:"); + expect(() => httpTransport("https://a%zz@events.example.com")) + .to.throw(TypeError, "sink credentials must be percent-encoded values"); + expect(() => httpTransport("https://events.example.com", { + headers: { "invalid header": "value" }, + })).to.throw(TypeError); + expect(() => httpTransport("https://events.example.com", { + headers: new Headers({ "x-tenant-id": "store-42" }), + })).not.to.throw(); + }); +}); + +// Model a platform object from another realm: same behavior, different instanceof chain. +function withForeignInterfacePrototype(value: T): T { + const interfacePrototype = Object.getPrototypeOf(value); + const foreignPrototype = Object.create(Object.getPrototypeOf(interfacePrototype)); + Object.defineProperties(foreignPrototype, Object.getOwnPropertyDescriptors(interfacePrototype)); + Object.setPrototypeOf(value, foreignPrototype); + return value; +} + +// A sink which records the headers of every event it accepts. +function headerCollector(): { received: IncomingMessage["headers"][]; handler: Handler } { + const received: IncomingMessage["headers"][] = []; + return { + received, + handler: (request, response) => { + received.push(request.headers); + acceptEvent(request, response); + }, + }; +} + +async function withFetch(stub: typeof globalThis.fetch, run: () => Promise): Promise { + const originalFetch = globalThis.fetch; + globalThis.fetch = stub; + + try { + await run(); + } finally { + globalThis.fetch = originalFetch; + } +} + +export async function withServer(handler: Handler, run: (url: string) => Promise): Promise { + const server = createServer((request, response) => { + Promise.resolve(handler(request, response)).catch((error) => { + response.destroy(error); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + + try { + await run(`http://127.0.0.1:${port}`); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + server.closeAllConnections(); + }); + } +} diff --git a/webpack.config.js b/webpack.config.js index 1894ee50..d35f024b 100644 --- a/webpack.config.js +++ b/webpack.config.js @@ -7,9 +7,7 @@ module.exports = { }, resolve: { fallback: { - util: require.resolve("util/"), - http: false, - https: false + util: require.resolve("util/") }, }, plugins: [ From a81d69668b4fcbdd5146e5eef0c210aff8dda0e8 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:21:26 +0900 Subject: [PATCH 4/5] feat!: report HTTP delivery failures and typed response bodies Reject responses that did not accept the event and distinguish HTTP status failures from requests that never produced a response. Keep an unreadable body after a successful status separate from a failed delivery, so a caller does not resend an event the receiver has already accepted. Allow transport-wide and per-send response handlers, preserving their body types through emitterFor(). Include text and discard handlers and clean up the body after a handler fails. BREAKING CHANGE: non-2xx HTTP responses reject with HTTPTransportError. Successful responses resolve with either body or bodyError. Options.headers is checked by TypeScript, and built-in HTTP transport annotations use HTTPTransportFunction and HTTPEmitterFunction. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 107 +++++++++- README.md | 109 ++++++++-- src/index.ts | 19 +- src/transport/emitter.ts | 34 +++- src/transport/http/index.ts | 240 ++++++++++++++++++++--- test/integration/emitter_factory_test.ts | 19 +- test/integration/http_transport_test.ts | 214 ++++++++++++++++++-- 7 files changed, 664 insertions(+), 78 deletions(-) diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index d33bd8ae..26c47a0e 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -103,16 +103,81 @@ new CloudEvent({ type, source, data }).emit(); ### Upgrading From 10.x to 11.0 +In the 11.0.0 release, the built-in HTTP transport was rewritten on top of the Fetch API. + #### HTTP Transport -`httpTransport()` uses Fetch and resolves with `{ response, body }`, where -`response` is the native Fetch `Response` and `body` is its text, instead of -`{ body, headers }`. Access headers through `response.headers.get(name)` and -the status through `response.status`. +`httpTransport()` used to resolve with the response for every status code, so a receiver +that rejected an event looked just like one that accepted it. + +```js +import { CloudEvent, emitterFor, httpTransport } from "cloudevents"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); + +// resolved with { body, headers } whether the receiver returned 202 or 503 +const response = await emit(new CloudEvent({ type, source, data })); +``` + +A 2xx response now resolves with the native Fetch `Response` and its body as text, and +anything else rejects with an `HTTPTransportError`. + +```js +import { + CloudEvent, + emitterFor, + HTTPTransportError, + httpTransport, +} from "cloudevents"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); + +try { + const { response, body } = await emit(new CloudEvent({ type, source, data })); + console.log(response.status, body); +} catch (error) { + if (error instanceof HTTPTransportError) { + console.error(error.kind, error.response?.status); + } + throw error; +} +``` + +The transport takes a second argument now, for headers, how the response body +is read and, under `fetchOptions`, the other Fetch options. Its headers are Fetch header +forms, so a `string[]` value becomes a `Headers` you `append()` to. A redirect is a +failed send unless `fetchOptions: { redirect: "follow" }` says otherwise, and a body +that cannot be read after a 2xx resolves with `bodyError` in place of `body`. +```js +const headers = new Headers(); +headers.append("accept", "application/cloudevents+json"); +headers.append("accept", "application/json"); +await emit(new CloudEvent({ type, source, data }), { headers }); +``` + +#### The Emitter Singleton + +A listener registered with `Emitter.on("cloudevent", emit)` now returns a promise which +rejects for a receiver that did not accept the event. `emitEvent()` awaits those promises +by default, so the rejection reaches whoever called `emit()` on the event: ```js -const { response, body } = await emit(event); -console.log(response.status, response.headers.get("x-request-id"), body); +try { + await new CloudEvent({ type, source, data }).emit(); +} catch (error) { + console.error(error.kind, error.response?.status); +} +``` + +With `ensureDelivery` turned off, the listeners run through `EventEmitter`, which discards +what they return. A rejection there becomes an unhandled promise rejection, which a `503` +could not produce before. Leave `ensureDelivery` alone, or catch inside the listener: + +```js +Emitter.on("cloudevent", (event) => emit(event).catch(reportFailedDelivery)); + +// listeners are not awaited, so nothing else can catch what they reject with +new CloudEvent({ type, source, data }).emit(false); ``` #### Proxies, Custom CAs, mTLS and Connection Pooling @@ -165,3 +230,33 @@ project, so TypeScript accepts `dispatcher` only where those are undici's. Again DOM `RequestInit` most projects compile with, pass it as `fetchOptions: { dispatcher } as FetchRequestInit` or reach for `setGlobalDispatcher()` instead. + +#### Emitter Options + +`Options` now declares `headers` instead of leaving it to an index +signature that resolved to `unknown`. Its default header type remains the +CloudEvent header record used by custom transports. The emitter returned from +`httpTransport()` specializes it to `FetchHeadersInit`. + +TypeScript now rejects spreading custom transport options into a client +configuration that types its own `headers`: + +```ts +// merge the headers of one send over the ones the binding produced +function sendWithAxios(message: Message, options?: Options) { + const { headers, ...rest } = options ?? {}; + return axios.post(url, message.body, { + headers: { ...message.headers, ...headers } as AxiosRequestHeaders, + ...rest, + }); +} +``` + +It also rejects the generic types as annotations for the built-in transport, whose +options are narrower. Use `HTTPTransportFunction` and `HTTPEmitterFunction`, or leave +the types to inference: + +```ts +const transport: HTTPTransportFunction = httpTransport(sink); +const emit: HTTPEmitterFunction = emitterFor(transport); +``` diff --git a/README.md b/README.md index 60a6874e..b3b46903 100644 --- a/README.md +++ b/README.md @@ -67,18 +67,74 @@ async function main() { main().catch(console.error); ``` -The transport resolves with the native Fetch `Response` and its body as text -for every HTTP status. It sends each event once. Redirects are not followed -unless `fetchOptions: { redirect: "follow" }` enables them; Fetch preserves the -CloudEvent POST and its body only for `307` and `308`. +A 2xx response resolves with the native Fetch `Response` and its body, read as +text by default. Anything else rejects with an `HTTPTransportError`, whose +`kind` separates a response that was not 2xx from a request that never reached +a response. -`httpTransport()` accepts transport-wide `headers` and other Fetch options under -`fetchOptions`. Headers passed to `emit()` apply to that event only and take -precedence over transport and CloudEvent headers. URL credentials become an -authorization header unless the caller overrides it. +The body of a response that was not 2xx is on the error as `error.body`. +`error.response` cannot be read a second time. If reading the body failed, +`error.body` is absent and the error carries a `cause`. -See the [API transition guide](./API_TRANSITION_GUIDE.md) for response access and -Node.js proxy, certificate and connection settings. +```js +import { HTTPTransportError } from "cloudevents"; + +async function main() { + try { + await emit(ce); + } catch (error) { + if (error instanceof HTTPTransportError) { + console.error(error.kind, error.response?.status, error.body); + } + throw error; + } +} + +main().catch(console.error); +``` + +`httpTransport()` also takes headers for every event, a `responseHandler` to read +the body another way (or `httpDiscardResponseHandler` to skip it), and the other +Fetch options under `fetchOptions`. Headers passed to `emit()` apply to that +event only. A response handler passed with an event overrides the transport's +handler, and its return value determines the response body type. + +```js +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { + headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` }, + responseHandler: (response) => response.json(), +})); +``` + +A 2xx response means the receiver accepted the event, so a body that cannot be +read resolves with `{ response, bodyError }` in place of `{ response, body }`. +Only one of the two is present. Check which property the result has before using +it: + +```js +async function main() { + const result = await emit(ce); + if ("body" in result) { + console.log(result.body); + } else { + console.error(result.bodyError); + } +} + +main().catch(console.error); +``` + +Redirects are not followed by default, so each one is reported as a failed send. +On Node.js the `HTTPTransportError` carries the 3xx response; a browser returns +an opaque redirect instead, so `error.response.status` is `0` and the `location` +header cannot be read. Set `fetchOptions: { redirect: "follow" }` to follow +them, keeping in mind that Fetch preserves the CloudEvent POST and its body only +for `307` and `308` - it turns a `301`, `302` or `303` into a bodyless `GET`, +which drops the event. + +The [API transition guide](./API_TRANSITION_GUIDE.md) covers the header forms +the transport accepts and where a proxy or a custom CA goes now that +`http.globalAgent` no longer applies. If you prefer to use another transport mechanism for sending events over HTTP, you can use the `HTTP` binding to create a `Message` which @@ -128,20 +184,31 @@ You may also use the `Emitter` singleton to send your `CloudEvents`. ```js const { emitterFor, httpTransport, Mode, CloudEvent, Emitter } = require("cloudevents"); -// Create a CloudEvent emitter function to send events to our receiver -const emit = emitterFor(httpTransport("https://example.com/receiver")); - -// Use the emit() function to send a CloudEvent to its endpoint when a "cloudevent" event is emitted -// (see: https://nodejs.org/api/events.html#class-eventemitter) -Emitter.on("cloudevent", emit); - -... -// In any part of the code, calling `emit()` on a `CloudEvent` instance will send the event -new CloudEvent({ type, source, data }).emit(); +async function main() { + // Create a CloudEvent emitter function to send events to our receiver + const emit = emitterFor(httpTransport("https://example.com/receiver")); + + // Use the emit() function to send a CloudEvent to its endpoint when a + // "cloudevent" event is emitted + // (see: https://nodejs.org/api/events.html#class-eventemitter) + Emitter.on("cloudevent", emit); + + // Calling emit() on a CloudEvent instance sends it through every listener + await new CloudEvent({ + type: "com.example.order.created", + source: "/stores/store-42", + data: { orderId: "order-123" }, + }).emit(); +} -// You can also have several listeners to send the event to several endpoints +main().catch(console.error); ``` +You can also have several listeners to send the event to several endpoints. +`emit()` waits for all of them, and an endpoint which refuses the event fails +the whole call. See the [API transition guide](./API_TRANSITION_GUIDE.md) for +the details. + ## CloudEvent Objects All created `CloudEvent` objects are read-only. If you need to update a property or add a new extension to an existing cloud event object, you can use the `cloneWith` method. This will return a new `CloudEvent` with any update or new properties. For example: diff --git a/src/index.ts b/src/index.ts index 3c3d5bcd..df586fc3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,12 +8,13 @@ import { ValidationError } from "./event/validation"; import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces"; import { - Options, TransportFunction, EmitterFunction, emitterFor, Emitter, + Options, TransportFunction, EmitterFunction, HTTPEmitterFunction, emitterFor, Emitter, } from "./transport/emitter"; import { - FetchHeadersInit, FetchRequestInit, HTTPTransportFunction, HTTPTransportOptions, - HTTPTransportResponse, httpTransport, -} from "./transport/http"; + FetchHeadersInit, FetchRequestInit, HTTPResponseHandler, HTTPTransportError, + HTTPTransportErrorDetails, HTTPTransportErrorKind, + HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions, + httpDiscardResponseHandler, httpTextResponseHandler, httpTransport } from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -34,6 +35,9 @@ export { // From transport emitterFor, httpTransport, + httpTextResponseHandler, + httpDiscardResponseHandler, + HTTPTransportError, Emitter, // From Constants CONSTANTS @@ -54,10 +58,15 @@ export type { // From transport TransportFunction, EmitterFunction, + HTTPEmitterFunction, Options, FetchHeadersInit, FetchRequestInit, + HTTPResponseHandler, + HTTPTransportErrorDetails, + HTTPTransportErrorKind, HTTPTransportFunction, HTTPTransportOptions, - HTTPTransportResponse + HTTPTransportResponse, + HTTPTransportSendOptions }; diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 7ddaf7d3..f2c854a0 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -4,15 +4,20 @@ */ import { CloudEvent } from "../event/cloudevent"; -import { HTTP, Message, Mode } from "../message"; +import { Headers as CloudEventHeaders, HTTP, Message, Mode } from "../message"; import { EventEmitter } from "events"; +import type { + FetchHeadersInit, HTTPTransportFunction, HTTPTransportResponse, HTTPTransportSendOptions, +} from "./http"; /** * Options is an additional, optional dictionary of options that may * be passed to an EmitterFunction and TransportFunction * @interface */ -export interface Options { +export interface Options { + /** Headers for this send, taking precedence over the ones the binding produced */ + headers?: THeaders; [key: string]: string | Record | unknown; } @@ -27,6 +32,17 @@ export interface EmitterFunction { (event: CloudEvent, options?: TOptions): Promise; } +/** + * An emitter backed by the built-in HTTP transport. A response handler passed with one event + * determines the body type for that call; otherwise the transport's default body type is used. + */ +export interface HTTPEmitterFunction + extends EmitterFunction, Options> { + ( + event: CloudEvent, options: HTTPTransportSendOptions, + ): Promise>; +} + /** * TransportFunction is an invokable interface provided to the emitterFactory. * A TransportFunction's responsiblity is to send a JSON encoded event Message @@ -39,6 +55,20 @@ export interface TransportFunction { } const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; +/** + * Creates and returns an {@linkcode HTTPEmitterFunction} using the built-in HTTP + * transport returned by `httpTransport()`. The returned function converts a + * {@linkcode CloudEvent} into a {@linkcode Message} the same way the generic overload + * below does, and resolves with what the sink sent back, or rejects with an + * `HTTPTransportError`. + * + * @param {HTTPTransportFunction} fn the built-in HTTP transport to send events with + * @param { {Binding, Mode} } options network binding and message serialization options + * @returns {HTTPEmitterFunction} an emitter which resolves with the sink's response + */ +export function emitterFor( + fn: HTTPTransportFunction, options?: Options, +): HTTPEmitterFunction; /** * Creates and returns an {@linkcode EmitterFunction} using the supplied * {@linkcode TransportFunction}. The returned {@linkcode EmitterFunction} diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 1b5e9604..404e9427 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -12,40 +12,152 @@ export type FetchRequestInit = NonNullable[1 /** The header forms accepted by the Fetch implementation in the current environment */ export type FetchHeadersInit = NonNullable; -/** Options applied to every request sent by an HTTP transport */ -export interface HTTPTransportOptions { +/** Turns a Fetch response into the body value handed back by the HTTP transport */ +export type HTTPResponseHandler = (response: Response) => Promise; + +/** + * Read a response body as text, the default behavior of the HTTP transport + * + * @param {Response} response the response whose body should be read + * @returns {Promise} the response body as text + */ +export function httpTextResponseHandler(response: Response): Promise { + return response.text(); +} + +/** + * Discard a response body when the caller only needs the response metadata + * + * @param {Response} response the response whose body should be discarded + * @returns {Promise} completion after the body has been cancelled + */ +export async function httpDiscardResponseHandler(response: Response): Promise { + await response.body?.cancel(); +} + +/** + * Options applied to every request sent by an HTTP transport + */ +export interface HTTPTransportOptions { /** HTTP headers applied to every request, per-send headers take precedence */ headers?: FetchHeadersInit; /** Fetch options applied to every request, apart from the ones this transport controls */ fetchOptions?: Omit; + /** Reads every response body; defaults to {@linkcode httpTextResponseHandler} */ + responseHandler?: HTTPResponseHandler; } -/** The native Fetch response and its body, read as text */ -export interface HTTPTransportResponse { - response: Response; - body: string; +/** + * Options which replace the response handler for one send + */ +export interface HTTPTransportSendOptions extends Options { + /** Reads this response body instead of the handler configured for the transport */ + responseHandler: HTTPResponseHandler; } -/** A transport that returns an HTTP response */ -export type HTTPTransportFunction = TransportFunction; +/** What the built-in HTTP transport hands back for a send the sink accepted */ +export type HTTPTransportResponse = + | { + /** The native Fetch response, whose body has been handled */ + response: Response; + /** What the configured response handler returned */ + body: TBody; + bodyError?: never; + } + | { + /** The native Fetch response, whose status is still a successful 2xx */ + response: Response; + body?: never; + /** What the response handler threw after the sink accepted the event */ + bodyError: unknown; + }; /** - * Send an HTTP-bound CloudEvent with Fetch. The transport sends once and resolves with - * the native response and its body as text, for every HTTP status. Redirects are not - * followed unless fetchOptions.redirect enables them. Credentials in the sink URL become - * an authorization header, overridden by transport or per-send headers. + * An HTTP transport whose default response body type can be replaced for one send + */ +export interface HTTPTransportFunction + extends TransportFunction, Options> { + (message: Message, options: HTTPTransportSendOptions): Promise>; +} + +/** The result of attempting to handle a response body */ +type HandledResponse = + | { handled: true; body: unknown } + | { handled: false; error: unknown }; + +/** + * The failure category reported by {@linkcode HTTPTransportError} + * + * - `http-status`: the sink returned a response that was not 2xx + * - `network`: the request never produced a response, e.g. DNS, connection, or TLS failure + */ +export type HTTPTransportErrorKind = "http-status" | "network"; + +/** + * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind} + */ +export type HTTPTransportErrorDetails = Pick; + +/** + * A request failure reported by the built-in HTTP transport + */ +export class HTTPTransportError extends Error { + /** The failure category */ + readonly kind: HTTPTransportErrorKind; + /** The response the sink sent, present only for `http-status` */ + readonly response?: Response; + /** What the response handler returned, present when it succeeded for `http-status` */ + readonly body?: unknown; + /** The underlying Fetch or response handler failure */ + readonly cause?: unknown; + + constructor(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails = {}) { + super(errorMessage(kind, details)); + this.name = "HTTPTransportError"; + this.kind = kind; + this.response = details.response; + this.body = details.body; + this.cause = details.cause; + } +} + +/** + * httpTransport provides a simple HTTP Transport function, which can send a CloudEvent, + * encoded as a Message to the endpoint. The returned function can be used with emitterFor() + * to provide an event emitter, for example: + * + * ```js + * const emit = emitterFor(httpTransport("http://example.com")); + * emit(myCloudEvent) + * .then(({ response, body }) => console.log(response.status, body)) + * .catch(err => console.error(err.kind, err.response?.status)); + * ``` + * + * The event is sent once, without retries. A response that is not 2xx rejects with an + * {@linkcode HTTPTransportError} holding the response the sink sent back. Redirects are + * reported as errors unless `fetchOptions.redirect` says otherwise, since Fetch keeps the + * CloudEvent POST only for 307 and 308 - see the README for the details. + * + * Every response body is passed to the configured response handler. It is read as text by + * default; a handler passed with one send takes precedence over the transport's handler. + * A handler may also return `response.body` to hand the stream to the caller, who then + * has to cancel it if it is not read to the end. + * + * Credentials in _sink_ are sent as an `authorization` header, which the `headers` option + * and the headers of one send override. * * @param {string|URL} sink the destination endpoint for the event - * @param {HTTPTransportOptions} options headers and Fetch request options - * @returns {HTTPTransportFunction} a function that sends CloudEvents to the sink + * @param {HTTPTransportOptions} options headers, Fetch options and response behavior + * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_ */ -export function httpTransport( - sink: string | URL, options: HTTPTransportOptions = {}, -): HTTPTransportFunction { +export function httpTransport( + sink: string | URL, options: HTTPTransportOptions = {}, +): HTTPTransportFunction { const url = validateHTTPURL(sink); // fetch refuses a URL with credentials const sinkHeaders = credentialsFrom(url); const transportHeaders = headersFrom(options.headers); + const transportResponseHandler = responseHandlerFrom(options.responseHandler, httpTextResponseHandler); const fetchOptions: FetchRequestInit = { ...options.fetchOptions, // fetch reads an explicitly undefined redirect as absent, so the default comes last @@ -53,17 +165,81 @@ export function httpTransport( method: "POST", }; - return async (message: Message, sendOptions?: Options): Promise => { + const send = async ( + message: Message, sendOptions?: Options, + ): Promise> => { const headers = requestHeaders( sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), ); - const response = await fetch(url, { - ...fetchOptions, - headers, - body: message.body as BodyInit, - }); - return { response, body: await response.text() }; + const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler); + let response: Response; + try { + response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + }); + } catch (cause) { + throw new HTTPTransportError("network", { cause }); + } + + const handled = await handleResponse(response, responseHandler); + if (!response.ok) { + // the status is the failure; the handler result or error goes alongside it + throw new HTTPTransportError("http-status", handled.handled + ? { response, body: handled.body } + : { response, cause: handled.error }); + } + return handled.handled + ? { response, body: handled.body } + : { response, bodyError: handled.error }; }; + + // the overloads type the per-send handler this implementation cannot + return send as HTTPTransportFunction; +} + +/** + * Check a response handler and fall back to the handler from the enclosing transport + * + * @param {unknown} value the handler supplied at this layer, if any + * @param {HTTPResponseHandler} fallback the handler inherited from the transport or default + * @returns {HTTPResponseHandler} the handler to run + */ +function responseHandlerFrom( + value: unknown, fallback: HTTPResponseHandler, +): HTTPResponseHandler { + if (value === undefined || value === null) { + return fallback; + } + if (typeof value !== "function") { + throw new TypeError("options.responseHandler must be a function"); + } + return value as HTTPResponseHandler; +} + +/** + * Run the caller's response handler. A handler that resolves may leave its body unread on + * purpose; one that throws has the rest of its body cancelled, so a failed send does not + * hold a connection open. + * + * @param {Response} response the response to handle + * @param {HTTPResponseHandler} handler the selected handler + * @returns {HandledResponse} the handler's body value or thrown error + */ +async function handleResponse( + response: Response, handler: HTTPResponseHandler, +): Promise { + try { + return { handled: true, body: await handler(response) }; + } catch (error) { + try { + await httpDiscardResponseHandler(response); + } catch { + // best effort cleanup + } + return { handled: false, error }; + } } /** @@ -204,3 +380,19 @@ function headerValueFrom(value: unknown): string | undefined { } return String(value); } + +/** + * Build the message for an {@linkcode HTTPTransportError} + * + * @param {HTTPTransportErrorKind} kind the failure category + * @param {HTTPTransportErrorDetails} details what is known about the failure + * @returns {string} the error message + */ +function errorMessage(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails): string { + switch (kind) { + case "http-status": + return `HTTP transport received a non-2xx response: ${details.response?.status}`; + default: + return "HTTP transport request failed"; + } +} diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts index e73b10e9..52313346 100644 --- a/test/integration/emitter_factory_test.ts +++ b/test/integration/emitter_factory_test.ts @@ -11,8 +11,7 @@ import request from "superagent"; import got from "got"; import CONSTANTS from "../../src/constants"; -import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } - from "../../src"; +import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } from "../../src"; const DEFAULT_CE_CONTENT_TYPE = CONSTANTS.DEFAULT_CE_CONTENT_TYPE; const sink = "https://cloudevents.io/"; @@ -39,7 +38,12 @@ export const fixture = new CloudEvent({ }); function axiosEmitter(message: Message, options?: Options): Promise { - return axios.post(sink, message.body, { headers: message.headers as AxiosRequestHeaders, ...options }); + // per-send headers are merged over the ones the binding produced + const { headers, ...rest } = options ?? {}; + return axios.post(sink, message.body, { + headers: { ...message.headers, ...headers } as AxiosRequestHeaders, + ...rest, + }); } function superagentEmitter(message: Message, options?: Options): Promise { @@ -66,6 +70,15 @@ function gotEmitter(message: Message, options?: Options): Promise { } describe("emitterFor() defaults", () => { + it("Keeps CloudEvent header arrays in custom transport options", async () => { + const transport: TransportFunction = async (_message, options) => { + expect(options?.headers?.["x-tenant-id"]).to.deep.equal(["store-42", "store-99"]); + }; + const emit = emitterFor(transport); + + await emit(fixture, { headers: { "x-tenant-id": ["store-42", "store-99"] } }); + }); + it("Defaults to HTTP binding, binary mode", () => { function transport(message: Message): Promise { // A binary message will have the source attribute as a header diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index cf140d43..a0c83d17 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -12,8 +12,8 @@ import { json, text } from "stream/consumers"; import { runInNewContext } from "vm"; import { - CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, - Message, Mode, emitterFor, httpTransport, + CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, HTTPTransportError, + Message, Mode, emitterFor, httpDiscardResponseHandler, httpTextResponseHandler, httpTransport, } from "../../src"; import { assertStructured, fixture } from "./emitter_factory_test"; @@ -124,6 +124,79 @@ describe("Built-in HTTP transport", () => { }); }); + it("Reports an unreadable body without failing an accepted event", async () => { + // the sink accepts the event, then closes the connection before finishing its body + await withServer((_request, response) => { + response.writeHead(202); + response.write("partial", () => response.socket?.end()); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture); + + expect(result.response.status).to.equal(202); + expect("body" in result).to.equal(false); + expect("bodyError" in result).to.equal(true); + expect(result.bodyError).not.to.equal(undefined); + }); + }); + + it("Infers bodies from transport and per-send response handlers", async () => { + await withServer((_request, response) => { + response.writeHead(202); + response.end("42"); + }, async (url) => { + const emit = emitterFor(httpTransport(url, { + responseHandler: async (response) => ({ receipt: Number(await response.text()) }), + })); + + const configured = await emit(fixture); + expectType<{ receipt: number } | undefined>(configured.body); + // @ts-expect-error the transport handler determines the default body type + expectType(configured.body); + expect(configured.body).to.deep.equal({ receipt: 42 }); + + const overridden = await emit(fixture, { responseHandler: httpTextResponseHandler }); + expectType(overridden.body); + // @ts-expect-error the per-send handler replaces the transport body type for this call + expectType<{ receipt: number } | undefined>(overridden.body); + expect(overridden.body).to.equal("42"); + }); + }); + + it("Discards a body with the exported response handler", async () => { + await withServer((_request, response) => { + response.writeHead(202); + response.end("not needed"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { responseHandler: httpDiscardResponseHandler }); + + expectType(result.body); + expect(result.body).to.equal(undefined); + expect(result.response.bodyUsed).to.equal(true); + expect("bodyError" in result).to.equal(false); + }); + }); + + it("Returns what a custom response handler throws as bodyError for 2xx", async () => { + const handlerError = new Error("receipt was malformed"); + + await withServer((_request, response) => { + response.writeHead(202); + response.end("not a receipt"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { + responseHandler: async () => { throw handlerError; }, + }); + + expect(result.response.status).to.equal(202); + expect(result.response.bodyUsed).to.equal(true); + expect("body" in result).to.equal(false); + expect(result.bodyError).to.equal(handlerError); + }); + }); + it("Applies transport headers and lets per-send headers override them", async () => { const { received: receivedHeaders, handler } = headerCollector(); @@ -284,6 +357,72 @@ describe("Built-in HTTP transport", () => { expect(receivedOptions?.redirect).to.equal("manual"); }); + for (const statusCode of [400, 503]) { + it(`Reports a ${statusCode} response without retrying it`, async () => { + let requestCount = 0; + + await withServer((_request, response) => { + requestCount++; + response.writeHead(statusCode, { "x-request-id": `request-${statusCode}` }); + response.end(`status ${statusCode}`); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(statusCode); + expect(error.response?.headers.get("x-request-id")).to.equal(`request-${statusCode}`); + expect(error.body).to.equal(`status ${statusCode}`); + expect(error.response?.bodyUsed).to.equal(true); + expect(error.cause).to.equal(undefined); + expect(requestCount).to.equal(1); + }); + }); + } + + it("Applies a custom response handler to a non-2xx body", async () => { + await withServer((_request, response) => { + response.writeHead(422, { "content-type": "application/json" }); + response.end(`{"message":"invalid event"}`); + }, async (url) => { + const emit = emitterFor(httpTransport(url, { + responseHandler: async (response) => JSON.parse(await response.text()) as { message: string }, + })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(422); + expect(error.body).to.deep.equal({ message: "invalid event" }); + expect(error.cause).to.equal(undefined); + }); + }); + + for (const statusCode of [301, 302, 303]) { + it(`Returns ${statusCode} as an error by default instead of following it`, async () => { + let requestCount = 0; + let targetRequestCount = 0; + + await withServer((request, response) => { + requestCount++; + if (request.url === "/target") { + targetRequestCount++; + response.writeHead(204); + } else { + response.writeHead(statusCode, { location: "/target" }); + } + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/start`)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(statusCode); + expect(requestCount).to.equal(1); + expect(targetRequestCount).to.equal(0); + }); + }); + } + // Fetch keeps POST only for 307 and 308, the others become a bodyless GET for (const { statusCode, method, keepsBody } of [ { statusCode: 301, method: "GET", keepsBody: false }, @@ -321,21 +460,26 @@ describe("Built-in HTTP transport", () => { }); } - for (const statusCode of [400, 503]) { - it(`Resolves a ${statusCode} response without retrying it`, async () => { - let requestCount = 0; - await withServer((_request, response) => { - requestCount++; - response.writeHead(statusCode); - response.end("rejected"); - }, async (url) => { - const result = await emitterFor(httpTransport(url))(fixture); - expect(result.response.status).to.equal(statusCode); - expect(result.body).to.equal("rejected"); - expect(requestCount).to.equal(1); - }); + it("Reports a non-2xx status whose body cannot be read", async () => { + // the sink starts an error body, then closes the connection before finishing it + await withServer((_request, response) => { + response.writeHead(503); + response.write("partial", () => response.socket?.end()); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(503); + expect(error.body).to.equal(undefined); + expect(error.cause).not.to.equal(undefined); }); - } + }); + + it("Rejects a per-send response handler that is not a function", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + await rejects(emit(fixture, { responseHandler: "text" }), TypeError); + }); it("Rejects a per-send header that is not a valid header", async () => { const emit = emitterFor(httpTransport(UNREACHABLE_URL)); @@ -352,12 +496,34 @@ describe("Built-in HTTP transport", () => { }, async (url) => { const emit = emitterFor(httpTransport(url)); await emit(fixture, { - + // @ts-expect-error built-in HTTP accepts FetchHeadersInit, not CloudEventHeaders arrays headers: legacyHeaders, }); }); }); + it("Reports a network failure with its cause", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("network"); + expect(error.response).to.equal(undefined); + expect(error.cause).not.to.equal(undefined); + }); + + it("Reports a synchronous Fetch failure with its cause", async () => { + const cause = new Error("Fetch failed before returning a promise"); + + await withFetch(() => { throw cause; }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com")); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("network"); + expect(error.response).to.equal(undefined); + expect(error.cause).to.equal(cause); + }); + }); + it("Validates the sink and transport options when the transport is created", () => { expect(() => httpTransport("not a URL")).to.throw(TypeError); expect(() => httpTransport("ftp://events.example.com")).to.throw(TypeError, "unsupported protocol ftp:"); @@ -369,9 +535,23 @@ describe("Built-in HTTP transport", () => { expect(() => httpTransport("https://events.example.com", { headers: new Headers({ "x-tenant-id": "store-42" }), })).not.to.throw(); + expect(() => httpTransport("https://events.example.com", { + responseHandler: "text" as never, + })).to.throw(TypeError, "options.responseHandler must be a function"); }); }); +async function transportErrorFrom(emitted: Promise): Promise { + const error = await emitted.then(() => undefined, (reason: unknown) => reason); + expect(error).to.be.instanceOf(HTTPTransportError); + return error as HTTPTransportError; +} + +// Compile-time assertion. +function expectType(value: T): void { + void value; +} + // Model a platform object from another realm: same behavior, different instanceof chain. function withForeignInterfacePrototype(value: T): T { const interfacePrototype = Object.getPrototypeOf(value); From da558bb114c6c3393ff54df250a7bf321be5ada1 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:22:25 +0900 Subject: [PATCH 5/5] feat!: add abort signals to HTTP transport Combine transport-wide and per-send abort signals while fetching and handling a response. Preserve the first abort reason, support signals from another realm and detach subscriptions when the operation finishes. Keep the shared signal implementation in its final module so emitter wrappers can reuse it without moving transport code in a later feature. BREAKING CHANGE: Options.signal is checked as an AbortSignal. A cancelled HTTP request rejects with HTTPTransportError of kind aborted; a body failure after a successful status remains a successful delivery result. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 9 +- README.md | 20 +- src/transport/emitter.ts | 2 + src/transport/http/index.ts | 62 +++--- src/transport/signal.ts | 161 ++++++++++++++++ test/integration/http_transport_test.ts | 240 ++++++++++++++++++++++++ 6 files changed, 463 insertions(+), 31 deletions(-) create mode 100644 src/transport/signal.ts diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index 26c47a0e..16738f85 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -143,11 +143,14 @@ try { } ``` -The transport takes a second argument now, for headers, how the response body +The transport takes a second argument now, for headers, a signal, how the response body is read and, under `fetchOptions`, the other Fetch options. Its headers are Fetch header forms, so a `string[]` value becomes a `Headers` you `append()` to. A redirect is a failed send unless `fetchOptions: { redirect: "follow" }` says otherwise, and a body -that cannot be read after a 2xx resolves with `bodyError` in place of `body`. +that cannot be read after a 2xx resolves with `bodyError` in place of `body`. The +signal given to `httpTransport()` lasts as long as the transport, so a per-request +deadline belongs on the signal passed with the event. + ```js const headers = new Headers(); headers.append("accept", "application/cloudevents+json"); @@ -233,7 +236,7 @@ DOM `RequestInit` most projects compile with, pass it as #### Emitter Options -`Options` now declares `headers` instead of leaving it to an index +`Options` now declares `headers` and `signal` instead of leaving them to an index signature that resolved to `unknown`. Its default header type remains the CloudEvent header record used by custom transports. The emitter returned from `httpTransport()` specializes it to `FetchHeadersInit`. diff --git a/README.md b/README.md index b3b46903..94b2a4da 100644 --- a/README.md +++ b/README.md @@ -69,8 +69,8 @@ main().catch(console.error); A 2xx response resolves with the native Fetch `Response` and its body, read as text by default. Anything else rejects with an `HTTPTransportError`, whose -`kind` separates a response that was not 2xx from a request that never reached -a response. +`kind` separates a response that was not 2xx from an abort and from a request +that never reached a response. The body of a response that was not 2xx is on the error as `error.body`. `error.response` cannot be read a second time. If reading the body failed, @@ -93,17 +93,23 @@ async function main() { main().catch(console.error); ``` -`httpTransport()` also takes headers for every event, a `responseHandler` to read -the body another way (or `httpDiscardResponseHandler` to skip it), and the other -Fetch options under `fetchOptions`. Headers passed to `emit()` apply to that -event only. A response handler passed with an event overrides the transport's -handler, and its return value determines the response body type. +`httpTransport()` also takes the headers for every event, an `AbortSignal`, a +`responseHandler` to read the body another way (or `httpDiscardResponseHandler` +to skip it), and the remaining Fetch options under `fetchOptions`. Headers +passed to `emit()` apply to that event only, and there is no timeout option: a +deadline is an `AbortSignal`. ```js const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` }, responseHandler: (response) => response.json(), })); + +async function main() { + await emit(ce, { signal: AbortSignal.timeout(10000) }); +} + +main().catch(console.error); ``` A 2xx response means the receiver accepted the event, so a body that cannot be diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index f2c854a0..601e7880 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -16,6 +16,8 @@ import type { * @interface */ export interface Options { + /** Aborts the send, for transports that support cancellation */ + signal?: AbortSignal; /** Headers for this send, taking precedence over the ones the binding produced */ headers?: THeaders; [key: string]: string | Record | unknown; diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 404e9427..0d490646 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -5,6 +5,7 @@ import { Headers as CloudEventHeaders, Message } from "../../message"; import { Options, TransportFunction } from "../emitter"; +import { combineSignals, signalFrom } from "../signal"; /** The request options accepted by the Fetch implementation in the current environment */ export type FetchRequestInit = NonNullable[1]>; @@ -41,6 +42,11 @@ export async function httpDiscardResponseHandler(response: Response): Promise { /** HTTP headers applied to every request, per-send headers take precedence */ headers?: FetchHeadersInit; + /** + * Aborts every request sent by this transport, combined with the per-send signal. It lasts + * as long as the transport does, so set a per-request deadline on the per-send signal + */ + signal?: AbortSignal; /** Fetch options applied to every request, apart from the ones this transport controls */ fetchOptions?: Omit; /** Reads every response body; defaults to {@linkcode httpTextResponseHandler} */ @@ -89,9 +95,10 @@ type HandledResponse = * The failure category reported by {@linkcode HTTPTransportError} * * - `http-status`: the sink returned a response that was not 2xx + * - `aborted`: a signal supplied by the caller aborted the request * - `network`: the request never produced a response, e.g. DNS, connection, or TLS failure */ -export type HTTPTransportErrorKind = "http-status" | "network"; +export type HTTPTransportErrorKind = "http-status" | "aborted" | "network"; /** * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind} @@ -108,7 +115,7 @@ export class HTTPTransportError extends Error { readonly response?: Response; /** What the response handler returned, present when it succeeded for `http-status` */ readonly body?: unknown; - /** The underlying Fetch or response handler failure */ + /** The underlying failure, e.g. what fetch threw or the reason carried by a signal */ readonly cause?: unknown; constructor(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails = {}) { @@ -141,13 +148,14 @@ export class HTTPTransportError extends Error { * Every response body is passed to the configured response handler. It is read as text by * default; a handler passed with one send takes precedence over the transport's handler. * A handler may also return `response.body` to hand the stream to the caller, who then - * has to cancel it if it is not read to the end. + * has to cancel it if it is not read to the end, since the abort signals no longer apply + * once the handler has resolved. * * Credentials in _sink_ are sent as an `authorization` header, which the `headers` option * and the headers of one send override. * * @param {string|URL} sink the destination endpoint for the event - * @param {HTTPTransportOptions} options headers, Fetch options and response behavior + * @param {HTTPTransportOptions} options headers, Fetch options, abort and response behavior * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_ */ export function httpTransport( @@ -156,6 +164,7 @@ export function httpTransport( const url = validateHTTPURL(sink); // fetch refuses a URL with credentials const sinkHeaders = credentialsFrom(url); + const transportSignal = signalFrom(options.signal); const transportHeaders = headersFrom(options.headers); const transportResponseHandler = responseHandlerFrom(options.responseHandler, httpTextResponseHandler); const fetchOptions: FetchRequestInit = { @@ -168,31 +177,40 @@ export function httpTransport( const send = async ( message: Message, sendOptions?: Options, ): Promise> => { + const sendSignal = signalFrom(sendOptions?.signal); const headers = requestHeaders( sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), ); const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler); - let response: Response; + const combinedSignal = combineSignals(transportSignal, sendSignal); try { - response = await fetch(url, { - ...fetchOptions, - headers, - body: message.body as BodyInit, - }); - } catch (cause) { - throw new HTTPTransportError("network", { cause }); - } + let response: Response; + try { + response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + signal: combinedSignal.signal, + }); + } catch (cause) { + throw combinedSignal.signal?.aborted + ? new HTTPTransportError("aborted", { cause: combinedSignal.signal.reason ?? cause }) + : new HTTPTransportError("network", { cause }); + } - const handled = await handleResponse(response, responseHandler); - if (!response.ok) { - // the status is the failure; the handler result or error goes alongside it - throw new HTTPTransportError("http-status", handled.handled + const handled = await handleResponse(response, responseHandler); + if (!response.ok) { + // the status is the failure; the handler result or error goes alongside it + throw new HTTPTransportError("http-status", handled.handled + ? { response, body: handled.body } + : { response, cause: handled.error }); + } + return handled.handled ? { response, body: handled.body } - : { response, cause: handled.error }); + : { response, bodyError: handled.error }; + } finally { + combinedSignal.dispose(); } - return handled.handled - ? { response, body: handled.body } - : { response, bodyError: handled.error }; }; // the overloads type the per-send handler this implementation cannot @@ -392,6 +410,8 @@ function errorMessage(kind: HTTPTransportErrorKind, details: HTTPTransportErrorD switch (kind) { case "http-status": return `HTTP transport received a non-2xx response: ${details.response?.status}`; + case "aborted": + return "HTTP transport request was aborted"; default: return "HTTP transport request failed"; } diff --git a/src/transport/signal.ts b/src/transport/signal.ts new file mode 100644 index 00000000..14dd50a2 --- /dev/null +++ b/src/transport/signal.ts @@ -0,0 +1,161 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +/** + * Check a signal supplied by a caller, whether it was given to a transport, to a single send, + * or to an emitter wrapper, by its platform brand rather than the interface prototype of this + * realm + * + * @param {unknown} value the signal supplied by the caller, if any + * @returns {AbortSignal|undefined} the signal, or undefined when none was supplied + */ +export function signalFrom(value: unknown): AbortSignal | undefined { + if (value === undefined || value === null) { + return undefined; + } + try { + // the getter checks the AbortSignal platform brand and accepts signals from another realm + Reflect.get(AbortSignal.prototype, "aborted", value); + } catch { + throw new TypeError("options.signal must be an AbortSignal"); + } + return value as AbortSignal; +} + +/** + * Read the conventional signal from otherwise transport-specific send options + * + * @param {unknown} options the options passed to one emitter invocation + * @returns {AbortSignal|undefined} the validated signal, when one was supplied + */ +export function signalFromOptions(options: unknown): AbortSignal | undefined { + if (typeof options !== "object" || options === null) { + return undefined; + } + return signalFrom((options as { signal?: unknown }).signal); +} + +/** An abort signal scoped to one operation, and the cleanup which ends that scope */ +export interface CombinedSignal { + signal?: AbortSignal; + dispose: () => void; +} + +/** Subscribers sharing the one platform listener attached to an abort signal */ +interface AbortFanout { + subscribers: Set<() => void>; + listener: () => void; + listening: boolean; +} + +/** Active fan-outs, weakly keyed so an otherwise unused signal can still be collected */ +const abortFanouts = new WeakMap(); + +/** The relay for an operation which has no signal to relay */ +const NO_ABORT: CombinedSignal = { signal: undefined, dispose: () => undefined }; + +/** + * Create and attach the shared platform listener for one signal + * + * @param {AbortSignal} source the signal which owns the listener + * @returns {AbortFanout} the shared subscriber collection + */ +function createAbortFanout(source: AbortSignal): AbortFanout { + const subscribers = new Set<() => void>(); + const fanout: AbortFanout = { + subscribers, + listening: true, + listener: () => { + const current = Array.from(subscribers); + subscribers.clear(); + fanout.listening = false; + abortFanouts.delete(source); + current.forEach((callback) => callback()); + }, + }; + + abortFanouts.set(source, fanout); + source.addEventListener("abort", fanout.listener, { once: true }); + return fanout; +} + +/** + * Subscribe to a signal through one shared platform listener, however many operations use it + * + * @param {AbortSignal} source the signal whose abort should be relayed + * @param {Function} subscriber one operation's abort callback + * @returns {Function} an idempotent function which removes this subscription + */ +function subscribeAbort(source: AbortSignal, subscriber: () => void): () => void { + const fanout = abortFanouts.get(source) ?? createAbortFanout(source); + + fanout.subscribers.add(subscriber); + let disposed = false; + + return (): void => { + if (disposed) { + return; + } + disposed = true; + fanout.subscribers.delete(subscriber); + + if (fanout.listening && fanout.subscribers.size === 0) { + fanout.listening = false; + source.removeEventListener("abort", fanout.listener); + abortFanouts.delete(source); + } + }; +} + +/** + * Combine abort signals behind a disposable relay, so they apply only while an operation is + * running and do not retain a long-lived source after that work ends. + * + * `AbortSignal.any()` is not used here. Node.js 24 validates its arguments with `instanceof`, + * so it rejects the signals from another realm that {@linkcode signalFrom} accepts, with + * `The "signals[0]" argument must be an instance of AbortSignal`. `addEventListener()` works + * across realms on every supported version, so the sources are relayed through + * {@linkcode subscribeAbort}, which also keeps one listener per source however many operations + * share it. + * + * @param {Array} candidates the signals which can abort the operation + * @returns {CombinedSignal} the combined signal and an idempotent cleanup function + */ +export function combineSignals(...candidates: Array): CombinedSignal { + const sources = Array.from(new Set(candidates)) + .filter((signal): signal is AbortSignal => signal !== undefined); + if (sources.length === 0) { + return NO_ABORT; + } + + const controller = new AbortController(); + const unsubscribe: Array<() => void> = []; + let disposed = false; + + const dispose = (): void => { + if (disposed) { + return; + } + disposed = true; + unsubscribe.forEach((remove) => remove()); + unsubscribe.length = 0; + }; + + const abortFrom = (source: AbortSignal): void => { + controller.abort(source.reason); + dispose(); + }; + + const alreadyAborted = sources.find((source) => source.aborted); + if (alreadyAborted) { + abortFrom(alreadyAborted); + return { signal: controller.signal, dispose }; + } + + sources.forEach((source) => { + unsubscribe.push(subscribeAbort(source, () => abortFrom(source))); + }); + return { signal: controller.signal, dispose }; +} diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index a0c83d17..650770f1 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -6,6 +6,7 @@ import "mocha"; import { rejects } from "assert"; import { expect } from "chai"; +import { getEventListeners } from "events"; import { createServer, IncomingMessage, ServerResponse } from "http"; import { AddressInfo } from "net"; import { json, text } from "stream/consumers"; @@ -30,6 +31,13 @@ const acceptEvent: Handler = (_request, response) => { response.end(); }; +// a sink which accepts the event, then finishes its body a moment later +const streamEvent: Handler = (_request, response) => { + response.writeHead(202); + response.write("stream"); + setTimeout(() => response.end("ed"), 50); +}; + describe("Built-in HTTP transport", () => { it("Sends a binary event and resolves with the response for 2xx", async () => { let receivedBody: Record | undefined; @@ -178,6 +186,23 @@ describe("Built-in HTTP transport", () => { }); }); + it("Lets a custom response handler hand its stream to the caller", async () => { + await withServer(streamEvent, async (url) => { + const controller = new AbortController(); + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { + signal: controller.signal, + responseHandler: async (response) => response.body, + }); + + expectType | null | undefined>(result.body); + expect(result.response.bodyUsed).to.equal(false); + // after the handler returns the stream, cancelling it is up to the caller + controller.abort(); + expect(await new Response(result.body).text()).to.equal("streamed"); + }); + }); + it("Returns what a custom response handler throws as bodyError for 2xx", async () => { const handlerError = new Error("receipt was malformed"); @@ -460,6 +485,47 @@ describe("Built-in HTTP transport", () => { }); } + it("Reports a caller timeout as an abort, with the reason it carries", async () => { + await withServer(() => undefined, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: AbortSignal.timeout(50) })); + + expect(error.kind).to.equal("aborted"); + expect(error.response).to.equal(undefined); + expect((error.cause as Error)?.name).to.equal("TimeoutError"); + }); + }); + + it("Returns an abort during a 2xx body as bodyError", async () => { + const controller = new AbortController(); + let markHandlerStarted = (): void => undefined; + const handlerStarted = new Promise((resolve) => { + markHandlerStarted = resolve; + }); + + await withServer((_request, response) => { + response.writeHead(202); + response.write("partial"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const emitted = emit(fixture, { + signal: controller.signal, + responseHandler: async (response) => { + markHandlerStarted(); + return response.text(); + }, + }); + + await handlerStarted; + controller.abort(new DOMException("deadline exceeded", "TimeoutError")); + const result = await emitted; + + expect(result.response.status).to.equal(202); + expect("body" in result).to.equal(false); + expect((result.bodyError as Error).name).to.equal("TimeoutError"); + }); + }); + it("Reports a non-2xx status whose body cannot be read", async () => { // the sink starts an error body, then closes the connection before finishing it await withServer((_request, response) => { @@ -476,6 +542,177 @@ describe("Built-in HTTP transport", () => { }); }); + it("Reports an abort from the transport signal", async () => { + const controller = new AbortController(); + + await withServer(() => controller.abort(new Error("shutting down")), async (url) => { + const emit = emitterFor(httpTransport(url, { signal: controller.signal })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("aborted"); + expect(error.response).to.equal(undefined); + expect((error.cause as Error).message).to.equal("shutting down"); + }); + }); + + it("Reports an abort from a per-send signal", async () => { + const controller = new AbortController(); + + await withServer(() => controller.abort(), async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: controller.signal })); + + expect(error.kind).to.equal("aborted"); + expect(error.cause).not.to.equal(undefined); + }); + }); + + it("Accepts an AbortSignal whose interface prototype comes from another realm", async () => { + const controller = new AbortController(); + const signal = withForeignInterfacePrototype(controller.signal); + expect(signal).not.to.be.instanceOf(AbortSignal); + + await withServer(() => controller.abort(new Error("foreign signal gave up")), async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal("foreign signal gave up"); + }); + }); + + it("Does not send when the signal is already aborted", async () => { + let requestCount = 0; + const controller = new AbortController(); + controller.abort(); + + await withServer(() => { requestCount++; }, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: controller.signal })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("aborted"); + expect(requestCount).to.equal(0); + }); + }); + + it("Uses the first reason when both signals are already aborted", async () => { + const transport = new AbortController(); + const send = new AbortController(); + transport.abort(new Error("transport stopped first")); + send.abort(new Error("send stopped second")); + + const emit = emitterFor(httpTransport(UNREACHABLE_URL, { signal: transport.signal })); + const error = await transportErrorFrom(emit(fixture, { signal: send.signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal("transport stopped first"); + }); + + it("Sends with a transport signal and a per-send signal set at once", async () => { + const transport = new AbortController(); + + await withServer(acceptEvent, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const { response } = await emit(fixture, { signal: new AbortController().signal }); + + expect(response.status).to.equal(204); + expect(transport.signal.aborted).to.equal(false); + }); + }); + + it("Reports an abort from either signal when both are set", async () => { + for (const abortedSignal of ["transport", "per-send"]) { + const transport = new AbortController(); + const send = new AbortController(); + const aborting = abortedSignal === "transport" ? transport : send; + + await withServer(() => aborting.abort(new Error(`${abortedSignal} gave up`)), async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const error = await transportErrorFrom(emit(fixture, { signal: send.signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal(`${abortedSignal} gave up`); + }); + } + }); + + it("Shares one source listener between concurrent sends and removes it afterward", async () => { + const controller = new AbortController(); + let releaseRequests = (): void => undefined; + const requestsMayComplete = new Promise((resolve) => { + releaseRequests = resolve; + }); + + await withFetch(async () => { + await requestsMayComplete; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com", { signal: controller.signal })); + const emitted = Array.from({ length: 20 }, () => emit(fixture)); + + try { + expect(getEventListeners(controller.signal, "abort")).to.have.length(1); + releaseRequests(); + await Promise.all(emitted); + expect(getEventListeners(controller.signal, "abort")).to.have.length(0); + } finally { + releaseRequests(); + await Promise.allSettled(emitted); + } + }); + }); + + it("Fans a shared transport abort out to every concurrent send", async () => { + const controller = new AbortController(); + const reason = new Error("transport shutting down"); + + await withFetch(async (_input, options) => new Promise((_resolve, reject) => { + const signal = options?.signal; + if (!signal) { + reject(new Error("expected a combined abort signal")); + return; + } + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }), async () => { + const emit = emitterFor(httpTransport("https://events.example.com", { signal: controller.signal })); + const emitted = Array.from({ length: 20 }, () => transportErrorFrom(emit(fixture))); + + try { + expect(getEventListeners(controller.signal, "abort")).to.have.length(1); + controller.abort(reason); + const errors = await Promise.all(emitted); + errors.forEach((error) => { + expect(error.kind).to.equal("aborted"); + expect(error.cause).to.equal(reason); + }); + expect(getEventListeners(controller.signal, "abort")).to.have.length(0); + } finally { + controller.abort(reason); + await Promise.allSettled(emitted); + } + }); + }); + + it("Stops applying a transport signal once the send has resolved", async () => { + const transport = new AbortController(); + + await withServer(streamEvent, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const result = await emit(fixture, { + responseHandler: async (response) => response.body, + }); + + // the transport signal no longer reaches a body the handler handed to the caller + transport.abort(); + expect(await new Response(result.body).text()).to.equal("streamed"); + }); + }); + + it("Rejects a per-send signal that is not an AbortSignal", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + await rejects(emit(fixture, { signal: "not a signal" as unknown as AbortSignal }), TypeError); + }); + it("Rejects a per-send response handler that is not a function", async () => { const emit = emitterFor(httpTransport(UNREACHABLE_URL)); await rejects(emit(fixture, { responseHandler: "text" }), TypeError); @@ -535,6 +772,9 @@ describe("Built-in HTTP transport", () => { expect(() => httpTransport("https://events.example.com", { headers: new Headers({ "x-tenant-id": "store-42" }), })).not.to.throw(); + expect(() => httpTransport("https://events.example.com", { + signal: "not a signal" as unknown as AbortSignal, + })).to.throw(TypeError, "options.signal must be an AbortSignal"); expect(() => httpTransport("https://events.example.com", { responseHandler: "text" as never, })).to.throw(TypeError, "options.responseHandler must be a function");