Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 163 additions & 0 deletions API_TRANSITION_GUIDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -100,3 +100,166 @@ 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

In the 11.0.0 release, the built-in HTTP transport was rewritten on top of the Fetch API.

#### HTTP Transport

`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, 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`. 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");
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
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

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.

#### Emitter Options

`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`.

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);
```
196 changes: 182 additions & 14 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -57,10 +58,166 @@ 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);
```

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 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,
`error.body` is absent and the error carries a `cause`.

```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 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. Wrap an emitter with `withTimeout()`
when every send should have the same time limit:

```js
import { emitterFor, httpTransport, withTimeout } from "cloudevents";

const emit = withTimeout(
emitterFor(httpTransport("https://my.receiver.com/endpoint", {
headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` },
responseHandler: (response) => response.json(),
})),
10000,
);

async function main() {
await emit(ce);
}

main().catch(console.error);
```

For a deadline which varies by event, pass `AbortSignal.timeout()` to that
`emit()` call instead. If both are present, the caller's signal and the timeout
can each abort the send. The timeout passed to `withTimeout()` must be a whole
number from 0 to 2,147,483,647 milliseconds.

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.

`httpTransport()` sends each event once. Applications which can accept
at-least-once delivery can add retries by wrapping the emitter. `maxAttempts`
counts the first send, so this emitter makes at most five requests:

```js
import {
CloudEvent, emitterFor, httpTransport, withRetry, withTimeout,
} from "cloudevents";

const emit = withRetry(
withTimeout(
emitterFor(httpTransport("https://events.example.com/orders")),
10000,
),
{ maxAttempts: 5 },
);
const orderCreated = new CloudEvent({
type: "com.example.order.created",
source: "/orders",
data: { orderId: "order-123" },
});

async function main() {
await emit(orderCreated);
}

main().catch(console.error);
```

The default retry policy handles failures which are commonly temporary:

| Failure reported by the emitter | Retried by default |
| ------------------------------- | ------------------ |
| Network failure before a response | Yes |
| HTTP 408, 425, 429, 500, 502, 503 or 504 | Yes |
| An aborted send or another HTTP status | No |
| An error from a custom transport | No |

The wrapper order decides which work the timeout covers:

| Goal | Composition | What the timeout covers |
| ---- | ----------- | ----------------------- |
| Limit each attempt to 10 seconds | `withRetry(withTimeout(base, 10000))` | One fresh timeout per attempt; retry backoff is excluded |
| Limit the whole delivery to 30 seconds | `withTimeout(withRetry(base), 30000)` | Every attempt and retry backoff share one timeout |

A timed-out HTTP send is reported as an abort, which the default policy does
not retry. When an attempt fails earlier with a retryable network or HTTP error,
the next attempt receives a fresh timeout in the first composition above.

Retries use randomized exponential backoff. For HTTP 429 and 503, a valid
`Retry-After` header takes precedence, and `maxRetryDelay` caps whatever a delay
asks for - 30 seconds by default, so a distant `Retry-After` cannot park an
emitter for hours. Pass `shouldRetry(error, context)` or
`retryDelay(error, context)` to replace either policy. The exported
`isRetryableHTTPError()` and `defaultRetryDelay()` functions let a custom policy
reuse the defaults.

The signal passed with the event also ends a wait between attempts. The emitter
then rejects with what that signal carries, rather than with the failure which
led to the retry.

A network failure can happen after the receiver accepted an event but before
its response arrived. Every attempt therefore uses the same CloudEvent and the
same `source` and `id`, but the receiver still needs to recognize duplicate
delivery.

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
has properties for `headers` and `body`, allowing greater flexibility
Expand Down Expand Up @@ -109,20 +266,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:
Expand Down
Loading