Async Iteration Patterns in JavaScript and TypeScript

// The protocol in TypeScript terms
interface AsyncIterator<T> {
  next(): Promise<IteratorResult<T>>;
  return?(): Promise<IteratorResult<T>>;
  throw?(e: unknown): Promise<IteratorResult<T>>;
}
 
interface AsyncIterable<T> {
  [Symbol.asyncIterator](): AsyncIterator<T>;
}
 
// Manual implementation — paginated API client
class PaginatedFetcher implements AsyncIterable<Record<string, unknown>[]> {
  constructor(
    private baseUrl: string,
    private pageSize: number = 50
  ) {}
 
  [Symbol.asyncIterator](): AsyncIterator<Record<string, unknown>[]> {
    let page = 1;
    let hasMore = true;
    const { baseUrl, pageSize } = this;
 
    return {
      async next() {
        if (!hasMore) {
          return { value: undefined, done: true };
        }
 
        const url = `${baseUrl}?page=${page}&limit=${pageSize}`;
        const response = await fetch(url);
        const data = await response.json();
 
        page++;
        hasMore = data.items.length === pageSize;
 
        return { value: data.items, done: false };
      },
    };
  }
}
 
// Clean consumption with for-await-of
async function processAllUsers() {
  const users = new PaginatedFetcher('https://api.example.com/users', 50);
 
  for await (const batch of users) {
    for (const user of batch) {
      await processUser(user);
    }
  }
  // Automatically stops when API returns fewer items than pageSize
}

// ❌ Callback-based approach — nested, hard to compose
function fetchAllPages(
  url: string,
  callback: (items: unknown[]) => void,
  done: () => void
) {
  let page = 1;
  function fetchNext() {
    fetch(`${url}?page=${page}`)
      .then((r) => r.json())
      .then((data) => {
        callback(data.items);
        if (data.hasMore) {
          page++;
          fetchNext();
        } else {
          done();
        }
      });
  }
  fetchNext();
}
 
// ✅ Async generator — flat, composable, readable
async function* fetchAllPages(url: string, pageSize = 50) {
  let page = 1;
  let hasMore = true;
 
  while (hasMore) {
    const response = await fetch(`${url}?page=${page}&limit=${pageSize}`);
    const data = await response.json();
 
    yield* data.items; // Yield each item individually
    hasMore = data.items.length === pageSize;
    page++;
  }
}
 
// Consume directly
for await (const user of fetchAllPages('https://api.example.com/users')) {
  console.log(user.name);
}

// Utility: async map
async function* asyncMap<T, U>(
  source: AsyncIterable<T>,
  transform: (item: T) => U | Promise<U>
): AsyncGenerator<U> {
  for await (const item of source) {
    yield await transform(item);
  }
}
 
// Utility: async filter
async function* asyncFilter<T>(
  source: AsyncIterable<T>,
  predicate: (item: T) => boolean | Promise<boolean>
): AsyncGenerator<T> {
  for await (const item of source) {
    if (await predicate(item)) {
      yield item;
    }
  }
}
 
// Utility: async take (limit number of items)
async function* asyncTake<T>(
  source: AsyncIterable<T>,
  limit: number
): AsyncGenerator<T> {
  let count = 0;
  for await (const item of source) {
    if (count >= limit) return;
    yield item;
    count++;
  }
}
 
// Utility: async batch (group items into fixed-size chunks)
async function* asyncBatch<T>(
  source: AsyncIterable<T>,
  size: number
): AsyncGenerator<T[]> {
  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;
  }
}

// Compose into a processing pipeline
async function processActiveUsers() {
  const allUsers = fetchAllPages('https://api.example.com/users');
 
  // Pipeline: fetch → filter active → transform → batch → insert
  const activeUsers = asyncFilter(allUsers, (u: any) => u.isActive);
  const enriched = asyncMap(activeUsers, async (u: any) => ({
    ...u,
    lastSeen: await fetchLastActivity(u.id),
  }));
  const batches = asyncBatch(enriched, 100);
 
  for await (const batch of batches) {
    await bulkInsertToDatabase(batch);
    console.log(`Inserted ${batch.length} users`);
  }
}
 
// Each stage processes one item at a time
// Memory usage stays constant regardless of total user count

async function* databaseCursor(query: string): AsyncGenerator<Record<string, unknown>> {
  const connection = await getConnection();
  const cursor = await connection.query(query);
 
  try {
    while (cursor.hasNext()) {
      yield await cursor.next();
    }
  } finally {
    // Cleanup runs whether iteration completes normally,
    // breaks early, or throws an error
    await cursor.close();
    await connection.release();
    console.log('Database cursor and connection cleaned up');
  }
}
 
// All of these trigger the finally block:
 
// 1. Normal completion
for await (const row of databaseCursor('SELECT * FROM users')) {
  processRow(row);
}
// finally runs after last row
 
// 2. Early break
for await (const row of databaseCursor('SELECT * FROM users')) {
  if (row.id === targetId) break; // finally runs immediately
}
 
// 3. Exception
try {
  for await (const row of databaseCursor('SELECT * FROM users')) {
    throw new Error('Processing failed'); // finally runs before catch
  }
} catch (err) {
  console.error(err);
}

// ❌ No cleanup — connection and cursor leak on early exit
async function* leakyGenerator(query: string) {
  const conn = await getConnection();
  const cursor = await conn.query(query);
  while (cursor.hasNext()) {
    yield await cursor.next();
  }
  // If consumer breaks early, these never run:
  await cursor.close();
  await conn.release();
}
 
// ✅ Always use try/finally in generators that acquire resources
async function* safeGenerator(query: string) {
  const conn = await getConnection();
  const cursor = await conn.query(query);
  try {
    while (cursor.hasNext()) {
      yield await cursor.next();
    }
  } finally {
    await cursor.close();
    await conn.release();
  }
}

import { createReadStream } from 'fs';
import { createInterface } from 'readline';
 
// Process a multi-gigabyte log file line by line
async function* readLines(filePath: string): AsyncGenerator<string> {
  const stream = createReadStream(filePath, { encoding: 'utf-8' });
  const rl = createInterface({ input: stream, crlfDelay: Infinity });
 
  for await (const line of rl) {
    yield line;
  }
}
 
// Compose: read → parse → filter → aggregate
async function analyzeErrorLogs(logPath: string) {
  const lines = readLines(logPath);
  const errors = asyncFilter(lines, (line: string) => line.includes('ERROR'));
  const parsed = asyncMap(errors, (line: string) => {
    const match = line.match(/\[(\d{4}-\d{2}-\d{2})\] ERROR: (.+)/);
    return match ? { date: match[1], message: match[2] } : null;
  });
  const validErrors = asyncFilter(parsed, (e): e is NonNullable<typeof e> => e !== null);
 
  const errorCounts = new Map<string, number>();
  for await (const error of validErrors) {
    const count = errorCounts.get(error.message) ?? 0;
    errorCounts.set(error.message, count + 1);
  }
 
  return errorCounts;
  // Processes gigabytes of logs with constant memory usage
}