Devalue Async

Extension of the great devalue package with ability to handle async values

About

Wrapper around devalue with ability to serialize and deserialize async values.

• Promises (both resolved and rejected)
• Async Iterables (async generators, async iterators)
• ReadableStreams (Web Streams API)
• All the goodness of devalue: cyclical references, undefined, Infinity, NaN, -0, regular expressions, dates, Map and Set, BigInt, custom types, etc.

Installation

npm install devalue devalue-async

Usage

There are two main functions: stringifyAsync and parseAsync.

Basic Example

import { parseAsync, stringifyAsync } from "devalue-async";

const source = {
	asyncIterable: (async function* () {
		yield 1;
		yield 2;
		yield 3;
		return "done!";
	})(),
	promise: Promise.resolve("Hello world"),
};

// Stringify to an async iterable of strings
const serialized = stringifyAsync(source);

// Reconstruct the original structure
const result = await parseAsync<typeof source>(serialized);

console.log(await result.promise); // 'Hello world'

// Iterate through the async iterable
for await (const value of result.asyncIterable) {
	console.log(value); // 1, 2, 3
}

Working with ReadableStreams

import { parseAsync, stringifyAsync } from "devalue-async";

const source = {
	stream: new ReadableStream({
		start(controller) {
			controller.enqueue("chunk 1");
			controller.enqueue("chunk 2");
			controller.close();
		},
	}),
};

const serialized = stringifyAsync(source);
const result = await parseAsync<typeof source>(serialized);

const reader = result.stream.getReader();
while (true) {
	const res = await reader.read();
	if (res.done) {
		break;
	}
	console.log(res.value); // 'chunk 1', 'chunk 2'
}

Error Handling

Devalue Async can handle errors in async operations:

import { parseAsync, stringifyAsync } from "devalue-async";

class CustomError extends Error {
	constructor(message: string) {
		super(message);
		this.name = "CustomError";
	}
}

const source = {
	asyncIterable: (async function* () {
		yield 1;
		yield 2;
		throw new CustomError("Async error");
	})(),
	promise: Promise.reject(new CustomError("Something went wrong")),
};

const serialized = stringifyAsync(source, {
	reducers: {
		CustomError: (error) => error instanceof CustomError && error.message,
	},
});

const result = await parseAsync<typeof source>(serialized, {
	revivers: {
		CustomError: (message) => new CustomError(message),
	},
});

try {
	await result.promise;
} catch (error) {
	console.log(error instanceof CustomError); // true
	console.log(error.message); // 'Something went wrong'
}

Coercing Unknown Errors

When dealing with errors that don't have registered reducers, use coerceError:

class GenericError extends Error {
	constructor(cause: unknown) {
		super("Generic error occurred", { cause });
		console.error("Encountered a unregistered error:", cause);
		this.name = "GenericError";
	}
}

const source = {
	asyncIterable: (async function* () {
		yield 1;
		yield 2;
		throw new Error("Generic error");
	})(),
};

const serialized = stringifyAsync(source, {
	coerceError: (error) => new GenericError(error),
	reducers: {
		GenericError: (error) => {
			if (error instanceof GenericError) {
				// Don't stringify the error
				return null;
			}
			return false;
		},
	},
});

const result = await parseAsync<typeof source>(serialized, {
	revivers: {
		GenericError: () => new Error("Unknown error"),
	},
});

Streaming Over HTTP

// Server
export type ApiResponse = ReturnType<typeof getData>;

function getApiResponse() {
	return {
		metrics: getMetricsStream(), // ReadableStream
		notifications: getNotificationStream(), // async iterable
		user: getUserData(), // promise
	};
}

app.get("/api/data", async (req, res) => {
	const data = getApiResponse();

	res.setHeader("Content-Type", "text/plain");
	for await (const chunk of stringifyAsync(data)) {
		res.write(chunk);
	}
	res.end();
});

// Client
import type { ApiResponse } from "./server";

const response = await fetch("/api/data");
const responseBody = response.body!.pipeThrough(new TextDecoderStream());

const result = await parseAsync<ApiResponse>(responseBody);

console.log(result.user);
for await (const notification of result.notifications) {
	console.log(notification);
}

Custom Types

Like devalue, you can handle custom types with reducers and revivers:

class Vector {
	constructor(
		public x: number,
		public y: number,
	) {}
}

const source = {
	vectors: (async function* () {
		yield new Vector(1, 2);
		yield new Vector(3, 4);
	})(),
};
const iterable = stringifyAsync(source, {
	reducers: {
		Vector: (value) => value instanceof Vector && [value.x, value.y],
	},
});

const result = await parseAsync<typeof source>(iterable, {
	revivers: {
		Vector: (value) => {
			const [x, y] = value as [number, number];
			return new Vector(x, y);
		},
	},
});

for await (const vector of result.vectors) {
	console.log(vector); // Vector { x: 1, y: 2 }, Vector { x: 3, y: 4 }
}

Nested async values

It's supported to have nested async values

interface Comment {
	content: string;
	user: string;
}

async function getComments(): Promise<Comment[]> {
	// ...
}

const source = {
	post: Promise.resolve({
		comments: getComments(),
		title: "post title",
	}),
};

const iterable = stringifyAsync(source);

const result = await parseAsync<typeof source>(iterable);

const post = await result.post;
console.log(post.title); // "post title"

const comments = await post.comments;
console.log(comments); // [ { content: "comment 1", user: "KATT" } ]

API Reference

stringifyAsync(value, options?)

Serializes a value containing async elements to an async iterable of strings.

Parameters:

• value: The value to serialize
• options.reducers?: Record of custom type reducers (same as devalue)
• options.coerceError?: Function to handle unregistered errors

Returns: AsyncIterable<string>

parseAsync(iterable, options?)

Reconstructs a value from a serialized async iterable.

Parameters:

• iterable: AsyncIterable<string> from stringifyAsync
• options.revivers?: Record of custom type revivers (same as devalue)

Returns: Promise<T>

Serialization Format

The serialization format is based on devalue's format with additional support for async values.

Whenever an async value is encountered, it is registered with a unique id each value will be passed later on in the stream.

All async values are raced so values are resolved as they are encountered.

Example

const source = () => ({
	asyncIterable: (async function* () {
		yield "hello";
		yield "world";

		return "return value";
	})(),
});

const iterable = stringifyAsync(source());

for await (const chunk of iterable) {
	console.log(chunk); // ->
	// 1st chunk:
	// [{"asyncIterable":1},["AsyncIterable",2],1]
	// the 1 at the end is the id of the async iterable

	// 2nd chunk:
	// [1,0,["hello"]]
	// the 1 at the start is the id of the async iterable
	// the 0 means it is a yielded value

	// 3rd chunk:
	// [1,0,["world"]]
	// the 1 at the start is the id of the async iterable
	// the 0 means it is a yielded value

	// 4th chunk:
	// [1,2,["return value"]]
	// the 1 at the start is the id of the async iterable
	// the 2 means that the async iterable has returned a value
}

Limitations

Referential integrity and deduplication across chunks in the stream is not supported and requires a fork of devalue to be able to track already seen values.

Development

See .github/CONTRIBUTING.md, then .github/DEVELOPMENT.md. Thanks! 💖

Contributors

Alex / KATT 💻 📖 🤔 🚇 🚧 🔧

💝 This package was templated with create-typescript-app using the Bingo framework.