async-streamify

Stream and serialize nested promises and async iterables over HTTP, workers, etc

async-streamify enables seamless transmission of complex async objects (including promises and async iterators) over HTTP and other text-based protocols, while maintaining their async behavior on the receiving end.

Features

• ✨ Stream promises and async iterators as they resolve/yield
• 🚀 Receive native promises and async iterables on the client
• 🔄 Support for deeply nested async objects (iterable in a promise, etc)
• 🎯 Type-safe serialization and deserialization
• 🌊 Automatic backpressure handling
• 📦 Zero dependencies
• 🛡️ Works in all modern runtimes (browser, bun, deno, node, edge).

Installation

• From npm with bun add | deno add | npm install | pnpm add | yarn add and async-streamify.
• From jsr with deno | npx jsr | yarn dlx jsr | pnpm dlx jsr | bunx jsr and add gadicc:@async-streamify.

Quick Start

server.ts

1import { AsyncResponse } from "async-streamify";
2
3// Helper functions; integers() generates a new integer every 200ms.
4const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); // deno-fmt-ignore
5async function* integers(max=10) { let i=0; while (i <= max) { yield i++; await sleep(200); }}
6
7// Create an object with mixed async values
8const data = () => ({
9  availability: "immediate",
10  promise: sleep(100).then(() => "completed"),
11  stream: integers(3),
12  nested: {
13    iteratorInPromise: sleep(100).then(() => integers(3)),
14  },
15});
16export type data = typeof data;
17
18// HTTP handler
19export function handler(request: Request) {
20  return new AsyncResponse(data(), {
21    headers: {
22      "Access-Control-Allow-Origin": "*",
23      "Cache-Control": "no-cache", // Recommended for streaming responses
24    },
25  });
26}

client.ts

1import { deserializeResponse } from "async-streamify";
2import type { data } from "./server";
3
4const response = await fetch("http://localhost:8000");
5const result = await deserializeResponse<ReturnType<data>>(response);
6
7// Values are received as they become available
8console.log(result.availability); // "immediate"
9
10result.promise.then((value) => {
11  console.log(value); // "completed" (after 100ms + network latency)
12});
13
14// Async iterators retain their native behaviour
15for await (const num of result.stream) {
16  console.log(num); // 0, 1, 2, 3 (streamed every 200ms)
17}
18
19// Nested values work and stream as they resolve
20for await (const num of (await result.nested.iteratorInPromise)) {
21  console.log(num); // 0, 1, 2, 3
22}

API Reference

AsyncResponse

Creates a streaming response that serializes an object containing promises and async iterables. The response is streamed as newline-delimited JSON (NDJSON).

1new AsyncResponse(data: any, init?: ResponseInit)

Parameters:

• data: The object to serialize and stream. Can contain promises, async iterables, and nested objects.
• init: Standard ResponseInit options. The response will automatically set:
  • Content-Type: application/x-ndjson
  • Additional headers can be provided through init.headers

deserializeResponse

Deserializes a streaming response back into an object with promises and async iterables.

1deserializeResponse<T>(response: Response): Promise<T>

Low-level API

For more control, you can use the serializer/deserializer directly:

1import { AsyncObjectSerializer, deserialize } from "async-streamify";
2
3const serializer = new AsyncObjectSerializer(data);
4for await (const chunk of serializer) {
5  // Send chunks over your transport
6  // Each chunk is a JSON object that should be serialized
7}
8
9const deserializedData = await deserialize<typeof data>(receivedStream);

For a full implementation example (for Response types with NDSON), see send/response.ts and receive/response.ts.

How It Works

1. The server serializes objects into a stream objects that contains either resolved values or references to pending async operations.
2. Values are transmitted as soon as they become available (provided the stream is ready for more, i.e., backpressure handling).
3. The AsyncResponse and deserializeResponse helpers further serialize via NDJSON (Newline Deliminited JSON) for HTTP streaming.
4. The client "reassembles" the stream back into native objects, promises, async iterables.

Protocol Details

The serialized stream consists of the original root object on the first line, with any async instances being substituted with a unique index and being resolved on future lines with [idx, value], i.e.:

1const object = { promise: sleep(100).then("resolved"; integers: integers(2) };
2console.log(await Array.fromAsync(new AsyncObjectSerializer(object)));
3[
4  { promise: { $promise: 1 }, integers: { $asyncIterable: 2 } },
5  [ 1, { $resolve: "resolved" } ],
6  [ 2, { value: 0, done: false }],
7  [ 2, { value: 1, done: false }],
8  [ 2, { value: 2, done: false }],
9  [ 2, { value: undefined, done: true }],
10]

Limitations

• Async generators on the client yield as fast as the stream can handle, not when explicitly requested on the client.
• Because of how promise chains work, if you provide a promise as the only item to serialize, if you call await reassemble(...) you'll get the result or thrown error back, not a promise. To work around, simply nest it, e.g. { promise: new Promise() }.
• Errors in promise rejections are (de-serialized), and error instanceof Error works. But obviously instances of custom errors cannot be sent over the wire, so instead check if error.name === "CustomError" vs error instanceof CustomError, etc.
• Errors in async generators are not handled yet (TODO)
• Circular references are not supported
• WebSocket and bi-directional streaming are not currently supported
• The transport must support streaming and handle backpressure correctly

TypeScript Support

The library is written in TypeScript and provides full type safety. Use the generic parameter in deserializeResponse<T> to ensure type-safe deserialization:

1interface MyData {
2  promise: Promise<string>;
3  stream: AsyncIterable<number>;
4}
5
6const data = await deserializeResponse<MyData>(response);
7// data is fully typed

License

Copyright (c) 2024 by Gadi Cohen. MIT Licensed.