From 3c49fd819b54111d18469ed6eec707fe92bfd1aa Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:19:18 +0900 Subject: [PATCH 1/8] feat!: use Fetch API for HTTP transport 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 f968e902..7e98e83e 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -102,3 +102,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 abac1560..7a297c09 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 95d95be6411bb01143539aa849c11e008e22d6f8 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:21:26 +0900 Subject: [PATCH 2/8] feat!: report HTTP delivery failures and typed response bodies 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 7e98e83e..78ab76b5 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -105,16 +105,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 @@ -167,3 +232,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 7a297c09..74ba3041 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 @@ -129,20 +185,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 1298a748ba0caad54094a0f941b4bef5274df8ee Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:22:25 +0900 Subject: [PATCH 3/8] feat!: add abort signals to HTTP transport 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 78ab76b5..80222fea 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -145,11 +145,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"); @@ -235,7 +238,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 74ba3041..b0fcb48f 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"); From 63569939d94ff3e128535b0b204925703ea5d986 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 18 Oct 2026 12:20:11 +0900 Subject: [PATCH 4/8] refactor: keep HTTP response helpers private Signed-off-by: kahirokunn --- README.md | 8 ++++---- src/index.ts | 4 +--- src/transport/http/index.ts | 26 +++---------------------- test/integration/http_transport_test.ts | 10 ++++++---- 4 files changed, 14 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index b0fcb48f..e8244996 100644 --- a/README.md +++ b/README.md @@ -94,10 +94,10 @@ main().catch(console.error); ``` `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`. +`responseHandler` to read the body another way, 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`. To discard a response +body, use `responseHandler: async (response) => { await response.body?.cancel(); }`. ```js const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { diff --git a/src/index.ts b/src/index.ts index df586fc3..c7a6af45 100644 --- a/src/index.ts +++ b/src/index.ts @@ -14,7 +14,7 @@ import { FetchHeadersInit, FetchRequestInit, HTTPResponseHandler, HTTPTransportError, HTTPTransportErrorDetails, HTTPTransportErrorKind, HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions, - httpDiscardResponseHandler, httpTextResponseHandler, httpTransport } from "./transport/http"; + httpTransport } from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -35,8 +35,6 @@ export { // From transport emitterFor, httpTransport, - httpTextResponseHandler, - httpDiscardResponseHandler, HTTPTransportError, Emitter, // From Constants diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 0d490646..84fbdffe 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -16,26 +16,6 @@ export type FetchHeadersInit = NonNullable; /** 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 */ @@ -49,7 +29,7 @@ export interface HTTPTransportOptions { 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} */ + /** Reads every response body; defaults to reading it as text */ responseHandler?: HTTPResponseHandler; } @@ -166,7 +146,7 @@ export function httpTransport( const sinkHeaders = credentialsFrom(url); const transportSignal = signalFrom(options.signal); const transportHeaders = headersFrom(options.headers); - const transportResponseHandler = responseHandlerFrom(options.responseHandler, httpTextResponseHandler); + const transportResponseHandler = responseHandlerFrom(options.responseHandler, (response) => response.text()); const fetchOptions: FetchRequestInit = { ...options.fetchOptions, // fetch reads an explicitly undefined redirect as absent, so the default comes last @@ -252,7 +232,7 @@ async function handleResponse( return { handled: true, body: await handler(response) }; } catch (error) { try { - await httpDiscardResponseHandler(response); + await response.body?.cancel(); } catch { // best effort cleanup } diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index 650770f1..bdc4f707 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -14,7 +14,7 @@ import { runInNewContext } from "vm"; import { CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, HTTPTransportError, - Message, Mode, emitterFor, httpDiscardResponseHandler, httpTextResponseHandler, httpTransport, + Message, Mode, emitterFor, httpTransport, } from "../../src"; import { assertStructured, fixture } from "./emitter_factory_test"; @@ -163,7 +163,7 @@ describe("Built-in HTTP transport", () => { expectType(configured.body); expect(configured.body).to.deep.equal({ receipt: 42 }); - const overridden = await emit(fixture, { responseHandler: httpTextResponseHandler }); + const overridden = await emit(fixture, { responseHandler: (response) => response.text() }); expectType(overridden.body); // @ts-expect-error the per-send handler replaces the transport body type for this call expectType<{ receipt: number } | undefined>(overridden.body); @@ -171,13 +171,15 @@ describe("Built-in HTTP transport", () => { }); }); - it("Discards a body with the exported response handler", async () => { + it("Discards a body with a caller's 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 }); + const result = await emit(fixture, { + responseHandler: async (response) => { await response.body?.cancel(); }, + }); expectType(result.body); expect(result.body).to.equal(undefined); From f58fb59c618608502c6703f69d93c635040480ee Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 18 Oct 2026 17:39:05 +0900 Subject: [PATCH 5/8] refactor!: narrow HTTP transport options Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 20 +-- README.md | 12 +- src/index.ts | 14 +- src/transport/emitter.ts | 31 ---- src/transport/http/index.ts | 91 +++++----- src/transport/signal.ts | 5 +- test/integration/emitter_factory_test.ts | 9 + test/integration/http_transport_test.ts | 211 ++++++++++++----------- 8 files changed, 180 insertions(+), 213 deletions(-) diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index 80222fea..337f2b77 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -145,13 +145,12 @@ try { } ``` -The transport takes a second argument now, for headers, a signal, how the response body +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`. The -signal given to `httpTransport()` lasts as long as the transport, so a per-request -deadline belongs on the signal passed with the event. +that cannot be read after a 2xx resolves with `bodyError` in place of `body`. Pass +`headers` and `signal` with the event for settings that apply to one message. ```js const headers = new Headers(); @@ -233,7 +232,7 @@ 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 +`fetchOptions: { dispatcher } as RequestInit` or reach for `setGlobalDispatcher()` instead. #### Emitter Options @@ -241,7 +240,7 @@ DOM `RequestInit` most projects compile with, pass it as `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`. +`httpTransport()` accepts Fetch header forms instead. TypeScript now rejects spreading custom transport options into a client configuration that types its own `headers`: @@ -257,11 +256,10 @@ function sendWithAxios(message: Message, options?: Options) { } ``` -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: +Leave the built-in transport and emitter types to inference so their message +options and response body types are preserved: ```ts -const transport: HTTPTransportFunction = httpTransport(sink); -const emit: HTTPEmitterFunction = emitterFor(transport); +const transport = httpTransport(sink); +const emit = emitterFor(transport); ``` diff --git a/README.md b/README.md index e8244996..9b73c03b 100644 --- a/README.md +++ b/README.md @@ -93,11 +93,15 @@ async function main() { main().catch(console.error); ``` -`httpTransport()` also takes the headers for every event, an `AbortSignal`, a +`httpTransport()` also takes the headers for every event, a `responseHandler` to read the body another way, 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`. To discard a response -body, use `responseHandler: async (response) => { await response.body?.cancel(); }`. +under `fetchOptions`. The options passed to `emit()` apply to that Message only: +`headers` and `signal`. Unsupported transport or message options reject with a +`TypeError`. To discard a response body, use +`responseHandler: async (response) => { await response.body?.cancel(); }`. + +There is no timeout option: a deadline is an `AbortSignal` passed with the +message. ```js const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { diff --git a/src/index.ts b/src/index.ts index c7a6af45..945d198e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,12 +8,10 @@ import { ValidationError } from "./event/validation"; import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces"; import { - Options, TransportFunction, EmitterFunction, HTTPEmitterFunction, emitterFor, Emitter, + Options, TransportFunction, EmitterFunction, emitterFor, Emitter, } from "./transport/emitter"; import { - FetchHeadersInit, FetchRequestInit, HTTPResponseHandler, HTTPTransportError, - HTTPTransportErrorDetails, HTTPTransportErrorKind, - HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions, + HTTPResponseHandler, HTTPTransportError, HTTPTransportErrorKind, HTTPTransportOptions, HTTPTransportResponse, httpTransport } from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, @@ -56,15 +54,9 @@ export type { // From transport TransportFunction, EmitterFunction, - HTTPEmitterFunction, Options, - FetchHeadersInit, - FetchRequestInit, HTTPResponseHandler, - HTTPTransportErrorDetails, HTTPTransportErrorKind, - HTTPTransportFunction, HTTPTransportOptions, - HTTPTransportResponse, - HTTPTransportSendOptions + HTTPTransportResponse }; diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 601e7880..813564ff 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -6,9 +6,6 @@ import { CloudEvent } from "../event/cloudevent"; 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 @@ -34,17 +31,6 @@ 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 @@ -57,20 +43,6 @@ 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} @@ -85,9 +57,6 @@ export function emitterFor( * @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?: Options, -): EmitterFunction; export function emitterFor( fn: TransportFunction, options: Options = emitterDefaults, ): EmitterFunction { diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 84fbdffe..529f8e01 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -8,10 +8,10 @@ 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]>; +type FetchRequestInit = NonNullable[1]>; /** The header forms accepted by the Fetch implementation in the current environment */ -export type FetchHeadersInit = NonNullable; +type FetchHeadersInit = NonNullable; /** Turns a Fetch response into the body value handed back by the HTTP transport */ export type HTTPResponseHandler = (response: Response) => Promise; @@ -22,24 +22,21 @@ export type HTTPResponseHandler = (response: Response) => Promise; export interface HTTPTransportOptions { /** 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 reading it as text */ responseHandler?: HTTPResponseHandler; } -/** - * 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; -} +type HTTPMessageOptions = Pick, "headers" | "signal">; + +const supportedTransportOptions: Record, true> = { + headers: true, fetchOptions: true, responseHandler: true, +}; + +const supportedSendOptions: Record = { + headers: true, signal: true, +}; /** What the built-in HTTP transport hands back for a send the sink accepted */ export type HTTPTransportResponse = @@ -58,14 +55,6 @@ export type HTTPTransportResponse = bodyError: unknown; }; -/** - * 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 } @@ -83,7 +72,7 @@ export type HTTPTransportErrorKind = "http-status" | "aborted" | "network"; /** * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind} */ -export type HTTPTransportErrorDetails = Pick; +type HTTPTransportErrorDetails = Pick; /** * A request failure reported by the built-in HTTP transport @@ -125,28 +114,34 @@ export class HTTPTransportError extends Error { * 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, since the abort signals no longer apply + * Every response body is passed to the configured response handler, or read as text by + * default. A handler may return `response.body` to hand the stream to the caller, who must + * cancel it if it is not read to the end. The message's abort signal no longer applies * 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, abort and response behavior - * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_ + * @param {HTTPTransportOptions} options headers, Fetch options and response behavior + * @returns {TransportFunction} a function that sends CloudEvents to _sink_ */ export function httpTransport( - sink: string | URL, options: HTTPTransportOptions = {}, -): HTTPTransportFunction { + sink: string | URL, options?: HTTPTransportOptions, +): TransportFunction, HTTPMessageOptions>; +export function httpTransport( + sink: string | URL, options: HTTPTransportOptions = {}, +): TransportFunction, HTTPMessageOptions> { + for (const name of Object.keys(options)) { + if (!Object.prototype.hasOwnProperty.call(supportedTransportOptions, name)) { + throw new TypeError(`unsupported HTTP transport option: ${name}`); + } + } 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, (response) => response.text()); + const responseHandler = responseHandlerFrom(options.responseHandler); const fetchOptions: FetchRequestInit = { ...options.fetchOptions, // fetch reads an explicitly undefined redirect as absent, so the default comes last @@ -154,15 +149,19 @@ export function httpTransport( method: "POST", }; - const send = async ( - message: Message, sendOptions?: Options, - ): Promise> => { + async function send( + message: Message, sendOptions?: HTTPMessageOptions, + ): Promise> { + for (const name of Object.keys(sendOptions ?? {})) { + if (!Object.prototype.hasOwnProperty.call(supportedSendOptions, name)) { + throw new TypeError(`unsupported HTTP send option: ${name}`); + } + } const sendSignal = signalFrom(sendOptions?.signal); const headers = requestHeaders( sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), ); - const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler); - const combinedSignal = combineSignals(transportSignal, sendSignal); + const combinedSignal = combineSignals(sendSignal); try { let response: Response; try { @@ -191,24 +190,20 @@ export function httpTransport( } finally { combinedSignal.dispose(); } - }; + } - // the overloads type the per-send handler this implementation cannot - return send as HTTPTransportFunction; + return send; } /** - * Check a response handler and fall back to the handler from the enclosing transport + * Check a response handler and fall back to reading the body as text * - * @param {unknown} value the handler supplied at this layer, if any - * @param {HTTPResponseHandler} fallback the handler inherited from the transport or default + * @param {unknown} value the configured handler, if any * @returns {HTTPResponseHandler} the handler to run */ -function responseHandlerFrom( - value: unknown, fallback: HTTPResponseHandler, -): HTTPResponseHandler { +function responseHandlerFrom(value: unknown): HTTPResponseHandler { if (value === undefined || value === null) { - return fallback; + return (response) => response.text(); } if (typeof value !== "function") { throw new TypeError("options.responseHandler must be a function"); diff --git a/src/transport/signal.ts b/src/transport/signal.ts index 14dd50a2..0760a1ec 100644 --- a/src/transport/signal.ts +++ b/src/transport/signal.ts @@ -4,9 +4,8 @@ */ /** - * 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 + * Check a signal supplied to a message or 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 diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts index 52313346..64c85a2c 100644 --- a/test/integration/emitter_factory_test.ts +++ b/test/integration/emitter_factory_test.ts @@ -70,6 +70,15 @@ function gotEmitter(message: Message, options?: Options): Promise { } describe("emitterFor() defaults", () => { + it("Preserves custom transport options and return types", async () => { + const transport: TransportFunction = async (_message, options) => { + expect(options?.routingKey).to.equal("orders"); + return 42; + }; + const result: number = await emitterFor(transport)(fixture, { routingKey: "orders" }); + expect(result).to.equal(42); + }); + 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"]); diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index bdc4f707..b643b6a3 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -13,12 +13,14 @@ import { json, text } from "stream/consumers"; import { runInNewContext } from "vm"; import { - CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, HTTPTransportError, + CONSTANTS, Headers as CloudEventHeaders, HTTPTransportError, HTTPTransportOptions, Message, Mode, emitterFor, httpTransport, } from "../../src"; import { assertStructured, fixture } from "./emitter_factory_test"; type Handler = (request: IncomingMessage, response: ServerResponse) => void | Promise; +type FetchRequestInit = NonNullable[1]>; +type FetchHeadersInit = NonNullable; // nock cannot intercept fetch, so these tests use a local server @@ -56,6 +58,7 @@ describe("Built-in HTTP transport", () => { expect(response.headers.get("x-request-id")).to.equal("receipt-7"); expect(response.bodyUsed).to.equal(true); expect(body).to.equal("accepted"); + expectType(body); 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); @@ -148,7 +151,7 @@ describe("Built-in HTTP transport", () => { }); }); - it("Infers bodies from transport and per-send response handlers", async () => { + it("Infers bodies from the transport response handler", async () => { await withServer((_request, response) => { response.writeHead(202); response.end("42"); @@ -159,15 +162,15 @@ describe("Built-in HTTP transport", () => { const configured = await emit(fixture); expectType<{ receipt: number } | undefined>(configured.body); - // @ts-expect-error the transport handler determines the default body type + // @ts-expect-error the transport handler determines the body type expectType(configured.body); expect(configured.body).to.deep.equal({ receipt: 42 }); - const overridden = await emit(fixture, { responseHandler: (response) => response.text() }); - 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"); + const withOptions = await emit(fixture, { + headers: { "x-request-id": "order-123" }, signal: new AbortController().signal, + }); + expectType<{ receipt: number } | undefined>(withOptions.body); + expect(withOptions.body).to.deep.equal({ receipt: 42 }); }); }); @@ -176,10 +179,10 @@ describe("Built-in HTTP transport", () => { response.writeHead(202); response.end("not needed"); }, async (url) => { - const emit = emitterFor(httpTransport(url)); - const result = await emit(fixture, { + const emit = emitterFor(httpTransport(url, { responseHandler: async (response) => { await response.body?.cancel(); }, - }); + })); + const result = await emit(fixture); expectType(result.body); expect(result.body).to.equal(undefined); @@ -191,11 +194,10 @@ 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, + const emit = emitterFor(httpTransport(url, { responseHandler: async (response) => response.body, - }); + })); + const result = await emit(fixture, { signal: controller.signal }); expectType | null | undefined>(result.body); expect(result.response.bodyUsed).to.equal(false); @@ -212,10 +214,10 @@ describe("Built-in HTTP transport", () => { response.writeHead(202); response.end("not a receipt"); }, async (url) => { - const emit = emitterFor(httpTransport(url)); - const result = await emit(fixture, { + const emit = emitterFor(httpTransport(url, { responseHandler: async () => { throw handlerError; }, - }); + })); + const result = await emit(fixture); expect(result.response.status).to.equal(202); expect(result.response.bodyUsed).to.equal(true); @@ -344,8 +346,9 @@ describe("Built-in HTTP transport", () => { }); }); - it("Passes standard Fetch options to fetch", async () => { + it("Passes standard and platform-specific Fetch options to fetch", async () => { let receivedOptions: FetchRequestInit | undefined; + const platformOptions = { dispatcher: {} }; await withFetch(async (_input, options) => { receivedOptions = options; @@ -353,6 +356,7 @@ describe("Built-in HTTP transport", () => { }, async () => { const emit = emitterFor(httpTransport("https://events.example.com/orders", { fetchOptions: { + ...platformOptions, cache: "no-store", credentials: "include", redirect: "error", @@ -365,6 +369,7 @@ describe("Built-in HTTP transport", () => { expect(receivedOptions?.credentials).to.equal("include"); expect(receivedOptions?.redirect).to.equal("error"); expect(receivedOptions?.method).to.equal("POST"); + expect(receivedOptions).to.include(platformOptions); }); it("Keeps the manual redirect when fetchOptions leaves redirect undefined", async () => { @@ -509,14 +514,13 @@ describe("Built-in HTTP transport", () => { response.writeHead(202); response.write("partial"); }, async (url) => { - const emit = emitterFor(httpTransport(url)); - const emitted = emit(fixture, { - signal: controller.signal, + const emit = emitterFor(httpTransport(url, { responseHandler: async (response) => { markHandlerStarted(); return response.text(); }, - }); + })); + const emitted = emit(fixture, { signal: controller.signal }); await handlerStarted; controller.abort(new DOMException("deadline exceeded", "TimeoutError")); @@ -544,28 +548,36 @@ describe("Built-in HTTP transport", () => { }); }); - it("Reports an abort from the transport signal", async () => { + it("Reports an abort from a per-send 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)); + await withServer(() => controller.abort(new Error("send stopped")), async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: controller.signal })); expect(error.kind).to.equal("aborted"); expect(error.response).to.equal(undefined); - expect((error.cause as Error).message).to.equal("shutting down"); + expect((error.cause as Error).message).to.equal("send stopped"); }); }); - it("Reports an abort from a per-send signal", async () => { + it("Keeps the transport available after one concurrent message is aborted", async () => { const controller = new AbortController(); - - await withServer(() => controller.abort(), async (url) => { + await withServer((request, response) => { + if (request.headers["ce-id"] === "cancelled") { + controller.abort(new Error("cancel one message")); + } else { + acceptEvent(request, response); + } + }, 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); + const cancelled = transportErrorFrom(emit(fixture.cloneWith({ id: "cancelled" }), { + signal: controller.signal, + })); + const other = emit(fixture.cloneWith({ id: "other" })); + expect((await cancelled).kind).to.equal("aborted"); + expect((await other).response.status).to.equal(204); + expect((await emit(fixture)).response.status).to.equal(204); }); }); @@ -589,55 +601,14 @@ describe("Built-in HTTP transport", () => { controller.abort(); await withServer(() => { requestCount++; }, async (url) => { - const emit = emitterFor(httpTransport(url, { signal: controller.signal })); - const error = await transportErrorFrom(emit(fixture)); + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: controller.signal })); 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; @@ -649,8 +620,8 @@ describe("Built-in HTTP transport", () => { 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)); + const emit = emitterFor(httpTransport("https://events.example.com")); + const emitted = Array.from({ length: 20 }, () => emit(fixture, { signal: controller.signal })); try { expect(getEventListeners(controller.signal, "abort")).to.have.length(1); @@ -664,9 +635,9 @@ describe("Built-in HTTP transport", () => { }); }); - it("Fans a shared transport abort out to every concurrent send", async () => { + it("Fans a shared message signal out to every concurrent send", async () => { const controller = new AbortController(); - const reason = new Error("transport shutting down"); + const reason = new Error("shutting down"); await withFetch(async (_input, options) => new Promise((_resolve, reject) => { const signal = options?.signal; @@ -676,8 +647,10 @@ describe("Built-in HTTP transport", () => { } 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))); + const emit = emitterFor(httpTransport("https://events.example.com")); + const emitted = Array.from({ length: 20 }, () => transportErrorFrom(emit(fixture, { + signal: controller.signal, + }))); try { expect(getEventListeners(controller.signal, "abort")).to.have.length(1); @@ -695,29 +668,50 @@ describe("Built-in HTTP transport", () => { }); }); - 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); + it("Rejects unsupported transport options when the transport is created", () => { + for (const options of [ + { signal: new AbortController().signal }, { timeout: 1000 }, { timeout: undefined }, + { method: "GET" }, { url: "https://other.example" }, { mode: Mode.STRUCTURED }, + { redirect: "follow" }, { responseHanlder: async () => "receipt" }, + ]) { + const [name] = Object.keys(options); + expect(() => httpTransport(UNREACHABLE_URL, options as never)) + .to.throw(TypeError, `unsupported HTTP transport option: ${name}`); + } + expect(() => httpTransport(UNREACHABLE_URL, { + // @ts-expect-error unsupported transport options are rejected statically + timeout: 1000, + })).to.throw(TypeError, "unsupported HTTP transport option: timeout"); + }); + + it("Rejects unsupported message options in the transport and its emitter", async () => { + await withoutFetch(async () => { + const send = httpTransport(UNREACHABLE_URL); + const emit = emitterFor(send); + for (const options of [ + { fetchOptions: { redirect: "follow" } }, { method: "GET" }, + { url: "https://other.example" }, { mode: Mode.STRUCTURED }, + { responseHandler: async () => 42 }, { responseHandler: undefined }, { timeout: 1000 }, + ]) { + await rejects(send({ headers: {}, body: "" }, options as never), TypeError); + await rejects(emit(fixture, options as never), TypeError); + } + // @ts-expect-error Fetch options belong to the transport + await rejects(emit(fixture, { fetchOptions: { redirect: "follow" } }), TypeError); + // @ts-expect-error the HTTP method is fixed by the transport + await rejects(send({ headers: {}, body: "" }, { method: "GET" }), TypeError); + // @ts-expect-error the sink is fixed by the transport + await rejects(emit(fixture, { url: "https://other.example" }), TypeError); + // @ts-expect-error mode is configured when creating the emitter + await rejects(emit(fixture, { mode: Mode.STRUCTURED }), TypeError); + // @ts-expect-error response handlers are configured when creating the transport + await rejects(emit(fixture, { responseHandler: async () => 42 }), TypeError); + }); }); it("Rejects a per-send header that is not a valid header", async () => { @@ -774,9 +768,6 @@ 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"); @@ -826,6 +817,16 @@ async function withFetch(stub: typeof globalThis.fetch, run: () => Promise } } +// Runs with a Fetch that accepts everything, then checks that nothing reached it +async function withoutFetch(run: () => Promise): Promise { + let requests = 0; + await withFetch(async () => { + requests++; + return new Response(null, { status: 204 }); + }, run); + expect(requests).to.equal(0); +} + export async function withServer(handler: Handler, run: (url: string) => Promise): Promise { const server = createServer((request, response) => { Promise.resolve(handler(request, response)).catch((error) => { From d55fea07f88c013b552b4e9bb36d2b35f5b8bdf6 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 18 Oct 2026 17:39:36 +0900 Subject: [PATCH 6/8] fix: validate CloudEvents header overrides Signed-off-by: kahirokunn --- README.md | 7 + src/schema/validator.ts | 12 + src/transport/emitter.ts | 2 +- src/transport/http/headers.ts | 149 ++++++++++++ src/transport/http/index.ts | 47 +++- test/integration/http_transport_test.ts | 298 +++++++++++++++++++++++- 6 files changed, 500 insertions(+), 15 deletions(-) create mode 100644 src/schema/validator.ts create mode 100644 src/transport/http/headers.ts diff --git a/README.md b/README.md index 9b73c03b..abaac90f 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,13 @@ under `fetchOptions`. The options passed to `emit()` apply to that Message only: `TypeError`. To discard a response body, use `responseHandler: async (response) => { await response.body?.cancel(); }`. +Existing `ce-*` and `content-type` headers cannot be changed. New CloudEvents +headers must be valid and, in structured mode, match the body. + +Other headers, such as `authorization`, `traceparent` and `baggage`, can be added +or overridden. `host` and `content-length` are controlled by Fetch and are +rejected. Invalid headers reject with a `TypeError` before the request is sent. + There is no timeout option: a deadline is an `AbortSignal` passed with the message. diff --git a/src/schema/validator.ts b/src/schema/validator.ts new file mode 100644 index 00000000..e482b581 --- /dev/null +++ b/src/schema/validator.ts @@ -0,0 +1,12 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import { ErrorObject } from "ajv"; +import compiled from "./v1"; + +type SchemaValidator = ((data: unknown) => boolean) & { errors?: ErrorObject[] | null }; + +// The generated validator sets `errors` at runtime, but its inferred type omits it. +export const validate: SchemaValidator = compiled; diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 813564ff..103e6ef6 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -15,7 +15,7 @@ import { EventEmitter } from "events"; 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 for this send, taking precedence over the ones the binding produced except CloudEvents ones */ headers?: THeaders; [key: string]: string | Record | unknown; } diff --git a/src/transport/http/headers.ts b/src/transport/http/headers.ts new file mode 100644 index 00000000..c0bbaef6 --- /dev/null +++ b/src/transport/http/headers.ts @@ -0,0 +1,149 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import CONSTANTS from "../../constants"; +import { V1, V03 } from "../../event/cloudevent"; +import { ValidationError } from "../../event/validation"; +import { Message } from "../../message"; +import { validate } from "../../schema/validator"; + +const SHARED_ATTRIBUTES = ["id", "source", "specversion", "type", "subject", "time", "datacontenttype"]; + +/** + * Check CloudEvents header additions after all caller headers have been merged. + * Messages without added CloudEvents headers and unrelated HTTP headers do not trigger event validation. + * Changes to existing headers have already been rejected by then. + * + * @param {Message} message the message whose body will be sent unchanged + * @param {Headers} original the headers produced by the binding + * @param {Headers} headers the headers that will be sent + * @returns {void} + */ +export function validateHeaderAdditions( + message: Message, original: Headers, headers: Headers, +): void { + const added = new Set(); + headers.forEach((value, name) => { + if (name.startsWith(CONSTANTS.EXTENSIONS_PREFIX) && !original.has(name)) { + added.add(name); + } + }); + const typeAdded = headers.get(CONSTANTS.HEADER_CONTENT_TYPE) !== original.get(CONSTANTS.HEADER_CONTENT_TYPE); + if (added.size === 0 && !typeAdded) { + return; + } + + const type = mediaType(headers); + const attributeValue = (name: string, value: string): string => + added.has(name) ? headerAttribute(value) : value; + if (type.startsWith(CONSTANTS.MIME_CE)) { + if (type !== CONSTANTS.MIME_CE_JSON || typeof message.body !== "string") { + if (added.size > 0) { + throw new TypeError("cannot verify CloudEvents headers against the structured body"); + } + return; + } + let payload: Record; + try { + payload = (JSON.parse(message.body) ?? {}) as Record; + } catch { + throw new TypeError("invalid JSON CloudEvent body"); + } + const version = payload.specversion; + validateAttributes(payload, version); + const attributes = { ...payload }; + headers.forEach((value, name) => { + if (!name.startsWith(CONSTANTS.EXTENSIONS_PREFIX) + || (!typeAdded && !added.has(name))) { + return; + } + const attribute = attributeName(name, version); + const decoded = attributeValue(name, value); + attributes[attribute] = decoded; + const expected = payload[attribute]; + const matches = decoded === String(expected) + || (attribute === CONSTANTS.CE_ATTRIBUTES.TIME + && new Date(decoded).getTime() === new Date(expected as string).getTime()); + if (!Object.prototype.hasOwnProperty.call(payload, attribute) || !matches) { + throw new TypeError(`${name} does not match the structured CloudEvent body`); + } + }); + validateAttributes(attributes, version); + return; + } + + const rawVersion = headers.get(CONSTANTS.CE_HEADERS.SPEC_VERSION); + const version = rawVersion === null ? null : attributeValue(CONSTANTS.CE_HEADERS.SPEC_VERSION, rawVersion); + if (!isKnownVersion(version)) { + return; + } + const attributes: Record = Object.create(null); + headers.forEach((value, name) => { + if (name.startsWith(CONSTANTS.EXTENSIONS_PREFIX)) { + attributes[attributeName(name, version)] = attributeValue(name, value); + } + }); + if (headers.has(CONSTANTS.HEADER_CONTENT_TYPE)) { + attributes[CONSTANTS.CE_ATTRIBUTES.CONTENT_TYPE] = headers.get(CONSTANTS.HEADER_CONTENT_TYPE); + } + validateAttributes(attributes, version); +} + +function isKnownVersion(version: unknown): boolean { + return version === V1 || version === V03; +} + +function mediaType(headers: Headers): string { + return (headers.get(CONSTANTS.HEADER_CONTENT_TYPE) ?? "").split(";", 1)[0].trim().toLowerCase(); +} + +function attributeName(name: string, version: unknown): string { + const attribute = name.slice(CONSTANTS.EXTENSIONS_PREFIX.length); + // 1.0 sends datacontenttype as content-type, never as a ce- header + if (version === V1 && (!/^[a-z0-9]+$/.test(attribute) || attribute === CONSTANTS.CE_ATTRIBUTES.CONTENT_TYPE)) { + throw new TypeError(`invalid CloudEvents 1.0 header: ${name}`); + } + return attribute; +} + +// Accept older quoted header forms before percent-decoding. +function headerAttribute(value: string): string { + const unquoted = value.startsWith("\"") && value.endsWith("\"") + ? value.slice(1, -1).replace(/\\(.)/g, "$1") + : value; + try { + return decodeURIComponent(unquoted); + } catch { + throw new TypeError("CloudEvents header values must use valid UTF-8 percent encoding"); + } +} + +// Validate raw attributes before constructor defaults or parser corrections. +function validateAttributes(attributes: Record, version: unknown): void { + if (typeof version === "string" && version !== "" && !isKnownVersion(version)) { + return; + } + let checked = attributes; + if (version === V03) { + // Only the shared attributes use the 1.0 schema. 0.3 also permits map + // extensions and has different schema and content-encoding attributes. + checked = {}; + for (const name of SHARED_ATTRIBUTES) { + if (Object.prototype.hasOwnProperty.call(attributes, name)) { + checked[name] = attributes[name]; + } + } + } + if (!validate(checked)) { + throw new ValidationError("invalid HTTP CloudEvents attributes", validate.errors); + } + if (version === V1) { + for (const name of Object.keys(attributes)) { + if (name !== CONSTANTS.STRUCTURED_ATTRS_1.DATA_BASE64 && !/^[a-z0-9]+$/.test(name)) { + throw new TypeError(`invalid CloudEvents attribute name: ${name}`); + } + } + } +} diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 529f8e01..d352154d 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -4,8 +4,10 @@ */ import { Headers as CloudEventHeaders, Message } from "../../message"; +import CONSTANTS from "../../constants"; import { Options, TransportFunction } from "../emitter"; import { combineSignals, signalFrom } from "../signal"; +import { validateHeaderAdditions } from "./headers"; /** The request options accepted by the Fetch implementation in the current environment */ type FetchRequestInit = NonNullable[1]>; @@ -20,7 +22,7 @@ export type HTTPResponseHandler = (response: Response) => Promise; * Options applied to every request sent by an HTTP transport */ export interface HTTPTransportOptions { - /** HTTP headers applied to every request, per-send headers take precedence */ + /** HTTP headers applied to every request, per-send headers take precedence but neither can change CloudEvents ones */ headers?: FetchHeadersInit; /** Fetch options applied to every request, apart from the ones this transport controls */ fetchOptions?: Omit; @@ -159,7 +161,7 @@ export function httpTransport( } const sendSignal = signalFrom(sendOptions?.signal); const headers = requestHeaders( - sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), + sinkHeaders, message, transportHeaders, headersFrom(sendOptions?.headers), ); const combinedSignal = combineSignals(sendSignal); try { @@ -253,21 +255,26 @@ function validateHTTPURL(sink: string | URL): 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 {Message} message 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, + sinkHeaders: Headers | undefined, message: Message, 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, message.headers); + if (transportHeaders === undefined && sendHeaders === undefined) { + return headers; + } + const original = new Headers(headers); setHeaders(headers, transportHeaders); setHeaders(headers, sendHeaders); + validateHeaderAdditions(message, original, headers); return headers; } @@ -334,7 +341,7 @@ function headersFrom(source: unknown): Headers | undefined { } /** - * Copy headers onto a target, replacing any that are already there + * Merge headers, rejecting changes to CloudEvents headers and the ones Fetch controls * * @param {Headers} target the headers being built * @param {CloudEventHeaders|Headers} source the headers to copy, undefined ones are ignored @@ -344,16 +351,34 @@ 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; - } + const headers = source instanceof Headers ? source : normalizeHeaders(source); + headers.forEach((value, name) => { + if (name === "host" || name === "content-length") { + throw new TypeError(`${name} is controlled by Fetch`); + } + if ((name.startsWith(CONSTANTS.EXTENSIONS_PREFIX) || name === CONSTANTS.HEADER_CONTENT_TYPE) + && target.has(name)) { + throw new TypeError(`cannot override CloudEvents header: ${name}`); + } + target.set(name, value); + }); +} + +/** + * Collect the headers of a Message, so names are lowercased and a repeated name keeps its last value + * + * @param {CloudEventHeaders} source the headers of the Message + * @returns {Headers} the headers, without the undefined ones + */ +function normalizeHeaders(source: CloudEventHeaders): Headers { + const headers = new Headers(); for (const [name, value] of Object.entries(source as Record)) { const headerValue = headerValueFrom(value); if (headerValue !== undefined) { - target.set(name, headerValue); + headers.set(name, headerValue); } } + return headers; } /** diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index b643b6a3..e2c4bfa6 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -13,8 +13,8 @@ import { json, text } from "stream/consumers"; import { runInNewContext } from "vm"; import { - CONSTANTS, Headers as CloudEventHeaders, HTTPTransportError, HTTPTransportOptions, - Message, Mode, emitterFor, httpTransport, + CloudEvent, CONSTANTS, Headers as CloudEventHeaders, HTTPTransportError, HTTPTransportOptions, + HTTP, Message, Mode, emitterFor, httpTransport, } from "../../src"; import { assertStructured, fixture } from "./emitter_factory_test"; @@ -229,17 +229,27 @@ describe("Built-in HTTP transport", () => { it("Applies transport headers and lets per-send headers override them", async () => { const { received: receivedHeaders, handler } = headerCollector(); + const tracing = { + traceparent: `00-${"a".repeat(32)}-${"b".repeat(16)}-01`, + tracestate: "vendor=old", baggage: "tenant=store-42", + }; + const sendTracing = { + traceparent: `00-${"c".repeat(32)}-${"d".repeat(16)}-01`, + tracestate: "vendor=new", baggage: "tenant=store-99,region=jp", + }; + await withServer(handler, async (url) => { const emit = emitterFor(httpTransport(url, { headers: new Headers({ authorization: "Bearer transport-token", "x-tenant-id": "store-42", + ...tracing, }), })); await emit(fixture); await emit(fixture, { - headers: { "x-tenant-id": "store-99", "x-request-id": "order-123" }, + headers: { "x-tenant-id": "store-99", "x-request-id": "order-123", ...sendTracing }, }); expect(receivedHeaders[0]?.authorization).to.equal("Bearer transport-token"); @@ -247,11 +257,293 @@ describe("Built-in HTTP transport", () => { 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"); + expect(receivedHeaders[0]).to.include(tracing); + expect(receivedHeaders[1]).to.include(sendTracing); // a header neither layer set keeps the value the binding produced expect(receivedHeaders[1]?.["ce-type"]).to.equal(fixture.type); }); }); + it("Accepts new binary attributes", async () => { + const { received, handler } = headerCollector(); + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url, { + headers: { "ce-tenant": "store-42" }, + })); + await emit(fixture, { + headers: { + "ce-region": "jp", "ce-subject": "order-123", + }, + }); + expect(received[0]?.["ce-id"]).to.equal(fixture.id); + expect(received[0]?.["ce-tenant"]).to.equal("store-42"); + expect(received[0]?.["ce-region"]).to.equal("jp"); + expect(received[0]?.["ce-subject"]).to.equal("order-123"); + expect(received[0]?.["content-type"]).to.equal(CONSTANTS.DEFAULT_CONTENT_TYPE); + }); + }); + + it("Rejects duplicate CloudEvents headers even with identical values", async () => { + const repeated: Array<[Mode, string, string]> = [ + [Mode.BINARY, "CE-ID", fixture.id], + [Mode.BINARY, "ce-id", ` ${fixture.id} `], + [Mode.BINARY, "ce-lunch", String(fixture.lunch)], + [Mode.BINARY, "Content-Type", CONSTANTS.DEFAULT_CONTENT_TYPE], + [Mode.STRUCTURED, "Content-Type", CONSTANTS.DEFAULT_CE_CONTENT_TYPE], + ]; + await withoutFetch(async () => { + for (const [mode, name, value] of repeated) { + const headers = { [name]: value }; + const error = { name: "TypeError", message: `cannot override CloudEvents header: ${name.toLowerCase()}` }; + await rejects(emitterFor(httpTransport(UNREACHABLE_URL, { headers }), { mode })(fixture), error); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL), { mode })(fixture, { headers }), error); + } + }); + }); + + it("Rejects changes to existing CloudEvents headers at either option level", async () => { + const changes = { + "CE-ID": "another-id", "ce-source": "/another-source", "ce-type": "another.type", + "ce-specversion": "0.3", "ce-lunch": "ramen", "content-type": "text/plain", + }; + await withoutFetch(async () => { + for (const [name, value] of Object.entries(changes)) { + const error = { name: "TypeError", message: `cannot override CloudEvents header: ${name.toLowerCase()}` }; + await rejects(emitterFor(httpTransport(UNREACHABLE_URL, { headers: { [name]: value } }))(fixture), error); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL))(fixture, { headers: { [name]: value } }), error); + } + for (const specversion of ["0.3", "11.8"]) { + await rejects(emitterFor(httpTransport(UNREACHABLE_URL))(fixture.cloneWith({ specversion }), { + headers: { "ce-id": "another-id" }, + }), TypeError); + } + }); + }); + + it("Rejects transport overrides even if the per-send headers restore the original value", async () => { + await withoutFetch(async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL, { headers: { "ce-id": "another-id" } })); + await rejects(emit(fixture, { headers: { "ce-id": fixture.id } }), { + name: "TypeError", message: "cannot override CloudEvents header: ce-id", + }); + }); + }); + + it("Rejects per-send overrides of CloudEvents attributes added by the transport", async () => { + await withoutFetch(async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL, { headers: { "ce-tenant": "store-42" } })); + for (const value of ["store-42", "store-99"]) { + await rejects(emit(fixture, { headers: { "CE-TENANT": value } }), { + name: "TypeError", message: "cannot override CloudEvents header: ce-tenant", + }); + } + }); + }); + + it("Rejects host and content-length from Messages and either option level", async () => { + await withoutFetch(async () => { + for (const [name, value] of [["Host", "other.example.com"], ["Content-Length", "1"]]) { + const error = { name: "TypeError", message: `${name.toLowerCase()} is controlled by Fetch` }; + const message = HTTP.binary(fixture); + message.headers[name] = value; + await rejects(httpTransport(UNREACHABLE_URL)(message), error); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL, { headers: { [name]: value } }))(fixture), error); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL))(fixture, { headers: { [name]: value } }), error); + } + }); + }); + + it("Keeps binding attribute values when validating additional headers", async () => { + const { received, handler } = headerCollector(); + const headers = { "ce-tenant": "shop" }; + await withServer(handler, async (url) => { + for (const source of ["/orders/a%20b", "/orders/%E3%81%82"]) { + const event = fixture.cloneWith({ source, tracestate: "vendor=25%" }); + await emitterFor(httpTransport(url, { headers }))(event); + await emitterFor(httpTransport(url))(event, { headers }); + for (const requestHeaders of received.slice(-2)) { + expect(requestHeaders["ce-source"]).to.equal(source); + expect(requestHeaders["ce-tracestate"]).to.equal("vendor=25%"); + } + } + expect(received).to.have.length(4); + }); + }); + + it("Rejects invalid CloudEvents header additions before calling Fetch", async () => { + const invalid: FetchHeadersInit[] = [ + { "ce-subject": "" }, + { "ce-dataschema": "/relative" }, + { "ce-bad-extension": "value" }, + { "ce-data_base64": "ZGF0YQ==" }, + { "ce-datacontenttype": "text/plain" }, + ]; + await withoutFetch(async () => { + for (const headers of invalid) { + await rejects(emitterFor(httpTransport(UNREACHABLE_URL, { headers }))(fixture), TypeError); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL))(fixture, { headers }), TypeError); + } + const message = HTTP.binary(fixture); + delete message.headers["ce-time"]; + await rejects(httpTransport(UNREACHABLE_URL)(message, { headers: { "ce-time": "not a timestamp" } }), TypeError); + }); + }); + + it("Accepts structured metadata headers that match the body", async () => { + const { received, handler } = headerCollector(); + const event = new CloudEvent({ + id: "event-42", source: "/account/%2F", type: "order.created", tenant: "store-42", + time: "2026-10-18T00:00:00Z", + }); + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url), { mode: Mode.STRUCTURED }); + await emit(event, { + headers: new Headers({ + "CE-ID": event.id, + "ce-source": "/account/%252F", + "ce-type": event.type, + "ce-time": "2026-10-18T09:00:00+09:00", + "ce-tenant": "store-42", + }), + }); + expect(received[0]?.["ce-id"]).to.equal(event.id); + expect(received[0]?.["ce-tenant"]).to.equal(event.tenant); + expect(received[0]?.["content-type"]).to.equal(CONSTANTS.DEFAULT_CE_CONTENT_TYPE); + }); + }); + + it("Rejects content-type changes including casing and media type parameters", async () => { + const error = { name: "TypeError", message: "cannot override CloudEvents header: content-type" }; + await withoutFetch(async () => { + for (const mode of [Mode.BINARY, Mode.STRUCTURED]) { + const type = mode === Mode.BINARY ? CONSTANTS.DEFAULT_CONTENT_TYPE : CONSTANTS.DEFAULT_CE_CONTENT_TYPE; + for (const value of [type.toUpperCase(), `${type}; charset=UTF-8`]) { + const headers = { "Content-Type": value }; + await rejects(emitterFor(httpTransport(UNREACHABLE_URL, { headers }), { mode })(fixture), error); + await rejects(emitterFor(httpTransport(UNREACHABLE_URL), { mode })(fixture, { headers }), error); + } + } + }); + }); + + it("Validates added structured metadata without reinterpreting existing headers", async () => { + const event = fixture.cloneWith({ source: "/orders/a%20b", tracestate: "vendor=25%", tenant: "shop" }); + const message = HTTP.structured(event); + message.headers["ce-source"] = "/orders/a%2520b"; + message.headers["ce-tracestate"] = "vendor=25%25"; + const { received, handler } = headerCollector(); + await withServer(handler, async (url) => { + await httpTransport(url)(message, { headers: { "ce-tenant": "shop" } }); + expect(received[0]?.["ce-source"]).to.equal("/orders/a%2520b"); + expect(received[0]?.["ce-tracestate"]).to.equal("vendor=25%25"); + expect(received[0]?.["ce-tenant"]).to.equal("shop"); + }); + }); + + it("Checks structured additions against the body without imposing 1.0 rules on other versions", async () => { + const { received, handler } = headerCollector(); + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url), { mode: Mode.STRUCTURED }); + for (const specversion of ["0.3", "11.8"]) { + const event = new CloudEvent({ + ...fixture, specversion, tenant: "store-42", "map-key": "value", + }, false); + await emit(event, { headers: { "ce-tenant": "store-42", "ce-map-key": "value" } }); + await rejects(emit(event, { headers: { "ce-tenant": "store-99" } }), { + name: "TypeError", message: "ce-tenant does not match the structured CloudEvent body", + }); + } + expect(received).to.have.length(2); + }); + }); + + it("Rejects headers that conflict with the structured body before calling Fetch", async () => { + await withoutFetch(async () => { + const invalid: FetchHeadersInit[] = [ + { "ce-id": "different-event" }, + { "ce-tenant": "not-in-the-body" }, + { "ce-datacontenttype": "application/json" }, + ]; + const structured = emitterFor(httpTransport(UNREACHABLE_URL), { mode: Mode.STRUCTURED }); + for (const headers of invalid) { + await rejects(structured(fixture, { headers }), TypeError); + const configured = emitterFor(httpTransport(UNREACHABLE_URL, { headers }), { mode: Mode.STRUCTURED }); + await rejects(configured(fixture), TypeError); + } + const midnight = fixture.cloneWith({ time: "2026-10-18T00:00:00Z" }); + await rejects(structured(midnight, { headers: { "ce-time": "2026-10-18" } }), TypeError); + }); + }); + + it("Forwards opaque structured bodies unread and rejects metadata additions it cannot verify", async () => { + let reads = 0; + const stream = new ReadableStream({ + pull: () => { reads++; }, + }, { highWaterMark: 0 }); + const messages: Message[] = [ + { headers: { "content-type": CONSTANTS.DEFAULT_CE_CONTENT_TYPE }, body: stream }, + { headers: { "content-type": "application/cloudevents+avro" }, body: new Uint8Array([1, 2, 3]) }, + ]; + let requests = 0; + await withFetch(async (_input, options) => { + expect(options?.body).to.equal(messages[requests++].body); + return new Response(null, { status: 204 }); + }, async () => { + const send = httpTransport(UNREACHABLE_URL); + for (const message of messages) { + await rejects(send(message, { headers: { "ce-id": fixture.id } }), { + name: "TypeError", message: "cannot verify CloudEvents headers against the structured body", + }); + await send(message, { headers: { baggage: "tenant=store-42" } }); + } + }); + expect(requests).to.equal(messages.length); + expect(reads).to.equal(0); + expect(stream.locked).to.equal(false); + }); + + it("Does not apply 1.0 extension naming or reserved header rules to other versions", async () => { + const { received, handler } = headerCollector(); + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url)); + await emit(fixture.cloneWith({ specversion: "0.3" }), { + headers: { "ce-map-key": "value", "ce-datacontenttype": "legacy-value" }, + }); + await emit(fixture.cloneWith({ specversion: "11.8" }), { + headers: { "ce-datacontenttype": "future-value" }, + }); + expect(received[0]?.["ce-map-key"]).to.equal("value"); + expect(received[1]?.["ce-datacontenttype"]).to.equal("future-value"); + }); + }); + + it("Accepts encoded and quoted attribute values without rewriting the wire headers", async () => { + const { received, handler } = headerCollector(); + await withServer(handler, async (url) => { + const message = HTTP.binary(fixture); + delete message.headers["ce-specversion"]; + for (const version of ["1%2E0", "\"1.0\""]) { + await httpTransport(url)(message, { + headers: { + "ce-subject": "order%20%E2%82%AC", "ce-tenant": "\"quoted\\\"tenant\"", "ce-specversion": version, + }, + }); + } + expect(received[0]?.["ce-subject"]).to.equal("order%20%E2%82%AC"); + expect(received[0]?.["ce-tenant"]).to.equal("\"quoted\\\"tenant\""); + expect(received.map((headers) => headers["ce-specversion"])).to.deep.equal(["1%2E0", "\"1.0\""]); + }); + }); + + it("Rejects malformed UTF-8 encodings in added attribute headers", async () => { + await withoutFetch(async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + await rejects(emit(fixture, { headers: { "ce-tenant": "%FF" } }), { + name: "TypeError", message: "CloudEvents header values must use valid UTF-8 percent encoding", + }); + }); + }); + it("Sends credentials from the sink URL as an authorization header", async () => { const { received: receivedHeaders, handler } = headerCollector(); From 9366f99c220125b123b01f1b853252c16ace2d3f Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 18 Oct 2026 17:40:12 +0900 Subject: [PATCH 7/8] refactor: type the compiled schema validator errors Signed-off-by: kahirokunn --- src/event/spec.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/event/spec.ts b/src/event/spec.ts index 042da8ed..5ea3adbb 100644 --- a/src/event/spec.ts +++ b/src/event/spec.ts @@ -7,13 +7,13 @@ import { ValidationError } from "./validation"; import { CloudEventV1 } from "./interfaces"; import { V1 } from "./cloudevent"; -import validate from "../schema/v1"; +import { validate } from "../schema/validator"; export function validateCloudEvent(event: CloudEventV1): boolean { if (event.specversion === V1) { if (!validate(event)) { - throw new ValidationError("invalid payload", (validate as any).errors); + throw new ValidationError("invalid payload", validate.errors); } } else { return false; From 77c8f4099384110814ab8ed1f21a9b2c058ac8c8 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 18 Oct 2026 17:40:53 +0900 Subject: [PATCH 8/8] refactor: remove redundant type assertions Signed-off-by: kahirokunn --- src/event/cloudevent.ts | 4 ++-- src/message/mqtt/index.ts | 2 +- src/transport/http/index.ts | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/event/cloudevent.ts b/src/event/cloudevent.ts index 16c722b8..f585735c 100644 --- a/src/event/cloudevent.ts +++ b/src/event/cloudevent.ts @@ -62,10 +62,10 @@ export class CloudEvent implements CloudEventV1 { delete properties.time; this.type = properties.type as string; - delete (properties as any).type; + delete properties.type; this.source = properties.source as string; - delete (properties as any).source; + delete properties.source; this.specversion = (properties.specversion) || V1; delete properties.specversion; diff --git a/src/message/mqtt/index.ts b/src/message/mqtt/index.ts index f576c567..121323d4 100644 --- a/src/message/mqtt/index.ts +++ b/src/message/mqtt/index.ts @@ -101,7 +101,7 @@ function MQTTMessageFactory(contentType: string, headers: Record)) { + for (const [name, value] of Object.entries(source)) { const headerValue = headerValueFrom(value); if (headerValue !== undefined) { headers.set(name, headerValue);