Building Event Streaming Pipelines with AsyncGenerator — Escaping Callback Hell
# TypeScript AsyncGenerator Event Streaming Pipeline
## Overview
This article introduces event streaming pipeline design patterns using TypeScript's AsyncGenerator as an alternative to callbacks and EventEmitter. We'll cover practical implementations including yield* delegation, backpressure, and combination with the Strategy pattern.
---
## Why AsyncGenerator?
```typescript
// ❌ Traditional callback approach: pyramid of doom
eventEmitter.on('data', (chunk) => {
transform(chunk, (err, result) => {
if (err) handleError(err);
sink(result, (err) => {
if (err) handleError(err);
});
});
});
// ❌ Promise chain: loses streaming characteristics
const results = await fetchAll(); // waits for everything
results.forEach(process);
// ✅ AsyncGenerator: streaming + async/await syntax
for await (const event of pipeline(source())) {
process(event);
}
```
**Key advantages:**
- Lazy evaluation (processes one item at a time)
- Native backpressure
- Composable with standard iteration protocol
- Type-safe end-to-end
---
## Core Type Definitions
```typescript
// Fundamental pipeline types
type AsyncStream<T> = AsyncGenerator<T, void, unknown>;
type Transform<A, B> = (source: AsyncStream<A>) => AsyncStream<B>;
type Sink<T, R> = (source: AsyncStream<T>) => Promise<R>;
// Event types for domain modeling
interface BaseEvent<T extends string, P = unknown> {
type: T;
payload: P;
timestamp: number;
id: string;
}
type UserEvent =
| BaseEvent<'user.created', { userId: string; email: string }>
| BaseEvent<'user.updated', { userId: string; changes: Record<string, unknown> }>
| BaseEvent<'user.deleted', { userId: string }>;
type OrderEvent =
| BaseEvent<'order.placed', { orderId: string; amount: number }>
| BaseEvent<'order.shipped', { orderId: string; trackingId: string }>
| BaseEvent<'order.cancelled', { orderId: string; reason: string }>;
type DomainEvent = UserEvent | OrderEvent;
```
---
## Building Blocks: Primitive Generators
```typescript
// Utility to create unique IDs
const createId = () => Math.random().toString(36).slice(2);
// 1. Source: generates events from various origins
async function* fromEventEmitter<T>(
emitter: EventEmitter,
eventName: string,
errorEvent = 'error'
): AsyncStream<T> {
const queue: Array<T | Error> = [];
let resolve: (() => void) | null = null;
let done = false;
const onData = (data: T) => {
queue.push(data);
resolve?.();
resolve = null;
};
const onError = (err: Error) => {
queue.push(err);
resolve?.();
resolve = null;
};
const onEnd = () => {
done = true;
resolve?.();
resolve = null;
};
emitter.on(eventName, onData);
emitter.on(errorEvent, onError);
emitter.on('end', onEnd);
try {
while (!done || queue.length > 0) {
if (queue.length === 0) {
await new Promise<void>((r) => { resolve = r; });
}
while (queue.length > 0) {
const item = queue.shift()!;
if (item instanceof Error) throw item;
yield item;
}
}
} finally {
emitter.off(eventName, onData);
emitter.off(errorEvent, onError);
emitter.off('end', onEnd);
}
}
// 2. Source: interval-based ticker
async function* interval(ms: number, signal?: AbortSignal): AsyncStream<number> {
let count = 0;
while (!signal?.aborted) {
yield count++;
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(resolve, ms);
signal?.addEventListener('abort', () => {
clearTimeout(timer);
reject(new DOMException('Aborted', 'AbortError'));
}, { once: true });
});
}
}
// 3. Source: from iterable
async function* fromIterable<T>(iterable: Iterable<T> | AsyncIterable<T>): AsyncStream<T> {
yield* iterable;
}
// 4. Source: merge multiple streams (fan-in)
async function* merge<T>(...streams: AsyncStream<T>[]): AsyncStream<T> {
const channel = new AsyncChannel<T>();
const producers = streams.map(async (stream) => {
try {
for await (const item of stream) {
await channel.send(item);
}
} finally {
channel.producerDone();
}
});
// Track completion
Promise.all(producers).then(() => channel.close());
yield* channel;
}
```
---
## AsyncChannel: Backpressure Implementation
```typescript
// Core component enabling backpressure
class AsyncChannel<T> {
private queue: T[] = [];
private waiters: Array<(value: IteratorResult<T>) => void> = [];
private producers = 0;
private closed = false;
private readonly capacity: number;
private sendWaiters: Array<() => void> = [];
constructor(capacity = 16) {
this.capacity = capacity;
this.producers = 0;
}
registerProducer() {
this.producers++;
}
producerDone() {
this.producers--;
if (this.producers === 0) {
this.close();
}
}
async send(value: T): Promise<void> {
// Block when buffer is full (backpressure)
while (this.queue.length >= this.capacity) {
await new Promise<void>((resolve) => {
this.sendWaiters.push(resolve);
});
}
if (this.waiters.length > 0) {
// Deliver directly to waiting consumer
const waiter = this.waiters.shift()!;
waiter({ value, done: false });
} else {
this.queue.push(value);
}
}
close() {
this.closed = true;
// Notify all waiting consumers
for (const waiter of this.waiters) {
waiter({ value: undefined as never, done: true });
}
this.waiters = [];
}
// Implement AsyncIterator protocol
[Symbol.asyncIterator](): AsyncIterator<T> {
return {
next: (): Promise<IteratorResult<T>> => {
if (this.queue.length > 0) {
const value = this.queue.shift()!;
// Notify blocked senders
const sendWaiter = this.sendWaiters.shift();
sendWaiter?.();
return Promise.resolve({ value, done: false });
}
if (this.closed) {
return Promise.resolve({ value: undefined as never, done: true });
}
// Wait for next value
return new Promise((resolve) => {
this.waiters.push((result) => {
if (!result.done) {
const sendWaiter = this.sendWaiters.shift();
sendWaiter?.();
}
resolve(result);
});
});
},
return: async (): Promise<IteratorResult<T>> => {
this.close();
return { value: undefined as never, done: true };
}
};
}
}
```
---
## Transform Operators
```typescript
// Operator library (modeled after RxJS but sync-free)
// map
function map<A, B>(fn: (value: A) => B | Promise<B>): Transform<A, B> {
return async function* (source) {
for await (const item of source) {
yield await fn(item);
}
};
}
// filter with type narrowing
function filter<A, B extends A>(predicate: (value: A) => value is B): Transform<A, B>;
function filter<A>(predicate: (value: A) => boolean | Promise<boolean>): Transform<A, A>;
function filter<A>(predicate: (value: A) => boolean | Promise<boolean>): Transform<A, A> {
return async function* (source) {
for await (const item of source) {
if (await predicate(item)) {
yield item;
}
}
};
}
// flatMap: expand each item into a stream
function flatMap<A, B>(fn: (value: A) => AsyncIterable<B> | Iterable<B>): Transform<A, B> {
return async function* (source) {
for await (const item of source) {
yield* fn(item);
}
};
}
// buffer: batch items
function buffer<T>(size: number): Transform<T, T[]> {
return async function* (source) {
let batch: T[] = [];
for await (const item of source) {
batch.push(item);
if (batch.length >= size) {
yield batch;
batch = [];
}
}
if (batch.length > 0) yield batch;
};
}
// throttle: rate limiting
function throttle<T>(ms: number): Transform<T, T> {
return async function* (source) {
let lastEmit = 0;
for await (const item of source) {
const now = Date.now();
const elapsed = now - lastEmit;
if (elapsed < ms) {
await new Promise((r) => setTimeout(r, ms - elapsed));
}
lastEmit = Date.now();
yield item;
}
};
}
// retry: error recovery
function retry<T>(
maxAttempts: number,
backoff: (attempt: number) => number = (n) => Math.min(1000 * 2 ** n, 30000)
): Transform<T, T> {
return async function* (source) {
let attempt = 0;
while (true) {
try {
for await (const item of source) {
attempt = 0; // reset on success
yield item;
}
break;
} catch (err) {
if (attempt >= maxAttempts) throw err;
await new Promise((r) => setTimeout(r, backoff(attempt)));
attempt++;
}
}
};
}
// deduplicate: suppress consecutive duplicates
function deduplicate<T, K = T>(keyFn: (value: T) => K = (v) => v as unknown as K): Transform<T, T> {
return async function* (source) {
let lastKey: K | typeof SENTINEL = SENTINEL;
const SENTINEL = Symbol('sentinel');
for await (const item of source) {
const key = keyFn(item);
if (key !== lastKey) {
lastKey = key;
yield item;
}
}
};
}
```
---
## Pipeline Composition with yield*
```typescript
// Pipeline builder using yield* delegation
class Pipeline<T> {
constructor(private stream: AsyncStream<T>) {}
static from<T>(source: AsyncStream<T>): Pipeline<T> {
return new Pipeline(source);
}
pipe<U>(transform: Transform<T, U>): Pipeline<U> {
return new Pipeline(transform(this.stream));
}
// yield* delegation: combine multiple pipelines
concat(...others: Pipeline<T>[]): Pipeline<T> {
const self = this.stream;
const otherStreams = others.map((p) => p.stream);
async function* concatenated(): AsyncStream<T> {
yield* self; // delegate to first stream
for (const stream of otherStreams) {
yield* stream; // sequentially delegate to remaining
}
}
return new Pipeline(concatenated());
}
// Fan-out: broadcast to multiple consumers
broadcast(count: number): Pipeline<T>[] {
const channels = Array.from({ length: count }, () => new AsyncChannel<T>());
const self = this.stream;
// Background task distributes to all channels
(async () => {
try {
for await (const item of self) {
await Promise.all(channels.map((ch) => ch.send(item)));
}
} finally {
channels.forEach((ch) => ch.close());
}
})();
return channels.map((ch) => Pipeline.from(ch as unknown as AsyncStream<T>));
}
async collect(): Promise<T[]> {
const results: T[] = [];
for await (const item of this.stream) {
results.push(item);
}
return results;
}
async forEach(fn: (item: T) => void | Promise<void>): Promise<void> {
for await (const item of this.stream) {
await fn(item);
}
}
[Symbol.asyncIterator]() {
return this.stream[Symbol.asyncIterator]();
}
}
```
---
## Strategy Pattern Integration
```typescript
// Processing strategy interface
interface ProcessingStrategy<T, R> {
process(event: T): Promise<R>;
canHandle(event: T): boolean;
}
// Concrete strategies
class UserCreatedStrategy implements ProcessingStrategy<UserEvent, void> {
canHandle(event: UserEvent): boolean {
return event.type === 'user.created';
}
async process(event: UserEvent): Promise<void> {
if (event.type !== 'user.created') return;
const { userId, email } = event.payload;
console.log(`Creating user: ${userId} (${email})`);
// send welcome email, setup profile, etc.
}
}
class OrderPlacedStrategy implements ProcessingStrategy<OrderEvent, void> {
canHandle(event: OrderEvent): boolean {
return event.type === 'order.placed';
}
async process(event: OrderEvent): Promise<void> {
if (event.type !== 'order.placed') return;
const { orderId, amount } = event.payload;
console.log(`Processing order ${orderId}: $${amount}`);
// inventory check, payment processing, etc.
}
}
// Strategy registry: dynamic dispatch
class StrategyRegistry<T> {
private strategies: Array<ProcessingStrategy<T, unknown>> = [];
register(strategy: ProcessingStrategy<T, unknown>): this {
this.strategies.push(strategy);
return this;
}
// Returns transform operator
asTransform(): Transform<T, { event: T; handled: boolean }> {
const strategies = this.strategies;
return async function* (source) {
for await (const event of source) {
const handler = strategies.find((s) => s.canHandle(event));
if (handler) {
await handler.process(event);
yield { event, handled: true };
} else {
yield { event, handled: false };
}
}
};
}
}
// Usage
const userRegistry = new StrategyRegistry<UserEvent>()
.register(new UserCreatedStrategy());
const orderRegistry = new StrategyRegistry<OrderEvent>()
.register(new OrderPlacedStrategy());
```
---
## Complete Real-World Example
```typescript
// Real-time order processing system
async function buildOrderProcessingPipeline(signal: AbortSignal) {
const eventEmitter = new EventEmitter();
// Source: events from message broker (Kafka/RabbitMQ equivalent)
const rawEvents = fromEventEmitter<unknown>(eventEmitter, 'message');
// Parsing and validation
function parseEvent(raw: unknown): DomainEvent | null {
try {
const event = raw as DomainEvent;
if (!event.type || !event.payload) return null;
return event;
} catch {
return null;
}
}
// Type guards
function isOrderEvent(event: DomainEvent): event is OrderEvent {
return event.type.startsWith('order.');
}
function isUserEvent(event: DomainEvent): event is UserEvent {
return event.type.startsWith('user.');
}
// Build pipeline
const parsedPipeline = Pipeline.from(rawEvents)
.pipe(map(parseEvent))
.pipe(filter((e): e is DomainEvent => e !== null))
.pipe(retry(3));
// Fan-out: separate order and user streams
const [orderStream, userStream, auditStream] = parsedPipeline.broadcast(3);
// Order processing pipeline
const orderPipeline = orderStream
.pipe(filter(isOrderEvent))
.pipe(throttle(100)) // max 10 orders/second
.pipe(buffer(10)) // process in batches of 10
.pipe(flatMap(async function* (batch) { // expand batches
for (const order of batch) {
yield order;
}
}))
.pipe(orderRegistry.asTransform())
.pipe(map(({ event, handled }) => ({
orderId: (event.payload as { orderId: string }).orderId,
handled,
processedAt: Date.now(),
})));
// User processing pipeline
const userPipeline = userStream
.pipe(filter(isUserEvent))
.pipe(deduplicate((e) => `${e.type}:${(e.payload as { userId: string }).userId}`))
.pipe(userRegistry.asTransform());
// Audit pipeline: write all events to log
const auditPipeline = auditStream
.pipe(map((event) => ({
...event,
auditedAt: Date.now(),
})));
// Run all pipelines concurrently
await Promise.all([
orderPipeline.forEach((result) => {
console.log(`Order processed:`, result);
}),
userPipeline.forEach(({ event, handled }) => {
console.log(`User event ${handled ? 'handled' : 'skipped'}:`, event.type);
}),
auditPipeline.forEach((event) => {
// write to audit log
console.log(`AUDIT:`, JSON.stringify(event));
}),
]);
return eventEmitter; // return for testing
}
```
---
## Testing Patterns
```typescript
// Test helpers
async function* mockStream<T>(items: T[], delay = 0): AsyncStream<T> {
for (const item of items) {
if (delay > 0) await new Promise((r) => setTimeout(r, delay));
yield item;
}
}
async function collectN<T>(stream: AsyncStream<T>, n: number): Promise<T[]> {
const results: T[] = [];
for await (const item of stream) {
results.push(item);
if (results.length >= n) break;
}
return results;
}
// Unit test examples
describe('Pipeline operators', () => {
test('map transforms values', async () => {
const source = mockStream([1, 2, 3]);
const mapped = map((x: number) => x * 2)(source);
const results = await Pipeline.from(mapped).collect();
expect(results).toEqual([2, 4, 6]);
});
test('filter removes non-matching values', async () => {
const source = mockStream([1, 2, 3, 4, 5]);
const filtered = filter((x: number) => x % 2 === 0)(source);
const results = await Pipeline.from(filtered).collect();
expect(results).toEqual([2, 4]);
});
test('buffer batches correctly', async () => {
const source = mockStream([1, 2, 3, 4, 5]);
const buffered = buffer<number>(2)(source);
const results = await Pipeline.from(buffered).collect();
expect(results).toEqual([[1, 2], [3, 4], [5]]);
});
test('backpressure limits throughput', async () => {
const channel = new AsyncChannel<number>(2); // capacity 2
const sent: number[] = [];
// Producer: send 10 items
const producer = (async () => {
for (let i = 0; i < 10; i++) {
await channel.send(i);
sent.push(i);
}
channel.close();
})();
// Slow consumer
const consumed: number[] = [];
const consumer = (async () => {
for await (const item of channel) {
await new Promise((r) => setTimeout(r, 10)); // slow processing
consumed.push(item);
}
})();
await Promise.all([producer, consumer]);
expect(consumed).toHaveLength(10);
// Verify backpressure was applied (producer blocked)
expect(sent.length).toBe(10);
});
});
```
---
## Performance Considerations
```typescript
// Measure pipeline throughput
async function benchmark<T>(
stream: AsyncStream<T>,
label: string
): Promise<void> {
let count = 0;
const start = performance.now();
for await (const _ of stream) {
count++;
}
const elapsed = performance.now() - start;
console.log(`${label}: ${count} items in ${elapsed.toFixed(2)}ms`);
console.log(`Throughput: ${(count / elapsed * 1000).toFixed(0)} items/sec`);
}
// Parallel processing for CPU-intensive transforms
function parallelMap<A, B>(
fn: (value: A) => Promise<B>,
concurrency: number
): Transform<A, B> {
return async function* (source) {
const channel = new AsyncChannel<B>(concurrency);
let activeWorkers = 0;
let sourceExhausted = false;
const processItem = async (item: A) => {
const result = await fn(item);
await channel.send(result);
activeWorkers--;
if (sourceExhausted && activeWorkers === 0) {
channel.close();
}
};
// Consume source and dispatch workers
(async () => {
for await (const item of source) {
activeWorkers++;
processItem(item); // intentionally not awaited
// Wait when concurrency limit reached
while (activeWorkers >= concurrency) {
await new Promise((r) => setTimeout(r, 0));
}
}
sourceExhausted = true;
if (activeWorkers === 0) channel.close();
})();
yield* channel;
};
}
```
---
## Summary
| Pattern | Benefit | Use Case |
|---|---|---|
| AsyncGenerator | Lazy evaluation + backpressure | Streaming data sources |
| yield* delegation | Composable pipelines | Sequential stream chaining |
| AsyncChannel | Bounded buffer | Producer/consumer decoupling |
| Strategy + Transform | Runtime extensibility | Event type dispatch |
| broadcast() | Fan-out | Multi-consumer scenarios |
AsyncGenerator-based pipelines provide **type safety**, **composability**, and **backpressure** in a way that's difficult to achieve with traditional EventEmitter or callback patterns. They're particularly well-suited for systems that process high-volume event streams.
2026.07.24
This page has been translated by machine translation. View original
Introduction
Running multi-step processing on the server and wanting to report the progress of each step to the browser in real time ── this is a common requirement in web application development.
For example, when a user submits a form, three steps run on the server side ── "validation → API call → post-processing" ── and each step's start, completion, and errors are sequentially notified to the browser via SSE (Server-Sent Events).
AsyncGenerator (asynchronous generator) turned out to be ideal for implementing this "asynchronous processing that generates events in sequence." This article compares the differences with callbacks and EventEmitter using actual code, and explains the advantages of AsyncGenerator.
Prerequisites / Environment
- TypeScript / Node.js
- SvelteKit (server-side processing)
A Common Challenge: "What Do You Do When Step 2 Fails?"
The differences between the three approaches are not visible with just the happy-path code. The differences emerge in behavior when an error occurs in the middle of a step.
We compare using the following common scenario:
- Step 1: Parameter validation
- Step 2: External API call (assumed to fail here)
- Step 3: Post-processing of the result
- After all steps complete, clean up session information
Callback
function processTask(
input: TaskInput,
onStepStart: (step: string) => void,
onStepComplete: (step: string, result: unknown) => void,
onError: (error: Error) => void,
onComplete: (result: unknown) => void
) {
const sessionId = createSession();
onStepStart("validate");
validate(input)
.then((validated) => {
onStepComplete("validate", validated);
onStepStart("api_call");
return callExternalApi(validated);
})
.then((apiResult) => {
onStepComplete("api_call", apiResult);
onStepStart("postprocess");
return postprocess(apiResult);
})
.then((final) => {
onStepComplete("postprocess", final);
deleteSession(sessionId); // cleanup
onComplete(final);
})
.catch((error) => {
deleteSession(sessionId); // cleanup is needed here too
onError(error);
});
}
Problems:
- 5 callback arguments. They grow every time a step is added
deleteSessionmust be written in both.thenand.catch(forgetting causes leaks)- As the Promise chain grows deeper, nesting becomes deeper
EventEmitter
const processor = new TaskProcessor();
// Register event listeners (before execution)
processor.on("step_start", (step) => sendSSE("step_start", step));
processor.on("step_complete", (step, result) => sendSSE("step_complete", { step, result }));
processor.on("error", (err) => {
sendSSE("error", err.message);
// Clean up here?
// But can we access sessionId?
});
processor.on("complete", (result) => {
sendSSE("complete", result);
// Cleanup here too
});
// Execute processing (separate from listener registration)
processor.process(input);
Problems:
- Listener registration and execution are separated → readers must scroll up and down to follow the code
- Cleanup on error must be written in both
errorandcomplete - Event names are strings → typos won't cause type errors
- Forgetting to remove listeners causes memory leaks
AsyncGenerator
async function* processTask(input: TaskInput): AsyncIterable<ProcessingEvent> {
const sessionId = createSession();
try {
yield { type: "step_start", step: "validate" };
const validated = await validate(input);
yield { type: "step_complete", step: "validate", result: validated };
yield { type: "step_start", step: "api_call" };
const apiResult = await callExternalApi(validated); // ← even if an exception occurs here...
yield { type: "step_complete", step: "api_call", result: apiResult };
yield { type: "step_start", step: "postprocess" };
const final = await postprocess(apiResult);
yield { type: "step_complete", step: "postprocess", result: final };
yield { type: "complete", result: final };
} finally {
deleteSession(sessionId); // ← just one place in finally
}
}
With try/finally, cleanup is consolidated in one place. Even if an exception occurs at step 2, finally is guaranteed to execute.
The calling side is also intuitive:
try {
for await (const event of processTask(input)) {
sendSSE(event.type, event);
}
} catch (error) {
sendSSE("error", error);
}
Overview
Implementation: AsyncGenerator × SSE Streaming
From here, I'll introduce the implementation adopted in an actual project.
Event Type Definitions
Define the events emitted by the pipeline using a Discriminated Union. When you branch with switch on the type field, properties are automatically narrowed within each case.
interface ProcessingEvent {
type: "step_start" | "step_complete" | "step_error" | "progress" | "complete" | "error";
stepId?: string;
processorType?: string;
progress?: number;
data?: unknown;
error?: string;
timestamp: Date;
}
Processing Engine
This is the core of the processing engine. Define an async generator function with async *, and return events from each step via yield.
class ProcessingEngine {
private activeProcesses = new Map<string, ProcessingContext>();
async *process(
config: ProcessingConfig,
parameters: Record<string, unknown>
): AsyncIterable<ProcessingEvent> {
const sessionId = this.generateSessionId();
const context: ProcessingContext = {
sessionId,
parameters,
startTime: new Date()
};
this.activeProcesses.set(sessionId, context);
try {
// Select strategy based on processing type
const strategy = this.selectStrategy(config);
// Transparently delegate strategy events
yield* strategy.execute(config, context);
} catch (error) {
yield {
type: "error" as const,
error: error instanceof Error ? error.message : "Unknown error",
timestamp: new Date()
};
} finally {
// Always release session, whether successful or not
this.activeProcesses.delete(sessionId);
}
}
}
The key is the try/finally pattern. Sessions are tracked in the activeProcesses map and reliably released in finally when processing completes. With EventEmitter, cleanup would need to be written in both complete and error, but with AsyncGenerator only one place is needed.
Delegation with yield*
yield* is a generator function keyword that transparently forwards all values produced by another generator.
yield* strategy.execute(config, context);
This means the events yielded by strategy.execute() are passed directly to the caller of process(). Even if you change the internal implementation of a processing strategy, no changes to the engine's code are needed.
Implementing Processing Strategies
Implement each processing type as an individual class. Putting event-emission helpers in the base class reduces yield boilerplate.
abstract class ProcessingStrategy {
abstract execute(
config: ProcessingConfig,
context: ProcessingContext
): AsyncIterable<ProcessingEvent>;
// Helper: generate step_start event
protected async *emitStepStart(stepId: string): AsyncIterable<ProcessingEvent> {
yield { type: "step_start", stepId, timestamp: new Date() };
}
// Helper: generate step_complete event
protected async *emitStepComplete(
stepId: string,
data?: unknown
): AsyncIterable<ProcessingEvent> {
yield { type: "step_complete", stepId, data, timestamp: new Date() };
}
// Helper: generate error event
protected async *emitError(error: string): AsyncIterable<ProcessingEvent> {
yield { type: "error", error, timestamp: new Date() };
}
}
The helper methods themselves are also defined with async *, and the calling side delegates with yield*. This has the same effect as writing yield statements directly, and lets you consolidate event generation logic in the base class.
Here is an example implementation of a concrete processing strategy:
class SingleStepStrategy extends ProcessingStrategy {
async *execute(
config: ProcessingConfig,
context: ProcessingContext
): AsyncIterable<ProcessingEvent> {
const stepId = "main";
try {
// Notify step start
yield* this.emitStepStart(stepId);
// Expand parameters and process
const result = this.buildPayload(config, context.parameters);
// Notify step completion (including processing result)
yield* this.emitStepComplete(stepId, result);
// Notify overall completion
yield { type: "complete", data: result, timestamp: new Date() };
} catch (error) {
const message = error instanceof Error ? error.message : "Processing failed";
yield* this.emitError(message);
}
}
}
Delegating to helpers with yield* and returning events directly with plain yield ── being able to mix both is the flexibility of AsyncGenerator.
SSE Endpoint: The Consumer Side of AsyncGenerator
This is where the true value of AsyncGenerator shines. Receive events with for await...of and stream them directly to the client in SSE format.
// SvelteKit API endpoint
export const POST: RequestHandler = async ({ request }) => {
const { config, parameters } = await request.json();
const stream = new ReadableStream({
async start(controller) {
const encoder = new TextEncoder();
try {
for await (const event of engine.process(config, parameters)) {
// Convert event to SSE format and send
const data = `event: processing\ndata: ${JSON.stringify(event)}\n\n`;
controller.enqueue(encoder.encode(data));
}
// Completion notification
const done = `event: complete\ndata: ${JSON.stringify({ complete: true })}\n\n`;
controller.enqueue(encoder.encode(done));
} catch (error) {
const errorEvent = `event: error\ndata: ${JSON.stringify({
error: error instanceof Error ? error.message : "Processing failed"
})}\n\n`;
controller.enqueue(encoder.encode(errorEvent));
} finally {
controller.close();
}
}
});
return new Response(stream, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive"
}
});
};
Why AsyncGenerator and ReadableStream work well together:
for await...ofreceives events one at a time, converts them to SSE format, and simply callscontroller.enqueue- The loop naturally waits until the generator
yields the next event (backpressure) - The engine's
try/finallyreleases the session; the endpoint'sfinallycallscontroller.close()── each layer manages its own resources
Extending Processing Strategies
When adding a new processing type, just create a class that inherits from ProcessingStrategy and register it with the engine. No changes to the engine or SSE endpoint are needed.
class MultiStepStrategy extends ProcessingStrategy {
async *execute(
config: ProcessingConfig,
context: ProcessingContext
): AsyncIterable<ProcessingEvent> {
const steps = config.steps ?? [];
for (let i = 0; i < steps.length; i++) {
yield* this.emitStepStart(steps[i].id);
const result = await this.executeStep(steps[i], context);
yield* this.emitStepComplete(steps[i].id, result);
yield {
type: "progress",
progress: ((i + 1) / steps.length) * 100,
timestamp: new Date()
};
// Pass the result to the next step
context.parameters = { ...context.parameters, previousResult: result };
}
yield { type: "complete", timestamp: new Date() };
}
}
Delegation via yield* makes the combination with the Strategy pattern feel natural. The engine only needs to call yield* strategy.execute(), without needing to know how many events or what kind of events the strategy will yield.
Comparison Summary of the Three Approaches
| Characteristic | Callback | EventEmitter | AsyncGenerator |
|---|---|---|---|
| Processing flow | Distributed across .then chains |
Listener registration and execution are separated | Linear top-to-bottom flow |
| Cleanup on error | Written in both .then and .catch |
Written in both error and complete |
Only one place in finally |
| Type safety | Individual types per callback argument | Event names are strings (vulnerable to typos) | Unified with ProcessingEvent type |
| Composition | Nesting hell | Listener proliferation | Transparently delegated with yield* |
| Backpressure | None | None | Automatically controlled with for await...of |
| SSE integration | enqueue inside callbacks |
enqueue inside listeners |
for await...of + enqueue |
Backpressure is an easy-to-overlook advantage. for await...of pauses the generator until the consumer requests the next event. The problem of events accumulating without limit on the server side for slow-network clients is naturally resolved.
Conclusion
AsyncGenerator is the ideal pattern for "asynchronous processing that generates multiple events in sequence."
async *defines an async generator functionyieldemits an eventyield*delegates to another generator → works naturally with the Strategy patternfor await...ofconsumes events → integrates naturally withReadableStream+ SSEtry/finallyensures reliable resource cleanup → no risk of forgetting
Especially in combination with SSE streaming, event generation (generator) and event delivery (SSE endpoint) are cleanly separated, resulting in a design where each layer can focus on managing its own resources.
