TypeScript Advanced Async Patterns 2026: Signals, Cancellation and Streams

// Type-safe promise utilities
type Awaited<T> = T extends Promise<infer U> ? U : T;
type MaybePromise<T> = T | Promise<T>;

// allSettled with typed results
async function settleAll<T extends readonly unknown[]>(
  promises: { [K in keyof T]: Promise<T[K]> }
): Promise<{ [K in keyof T]: { ok: true; value: T[K] } | { ok: false; error: unknown } }> {
  const results = await Promise.allSettled(promises);
  return results.map(r =>
    r.status === 'fulfilled' ? { ok: true, value: r.value } : { ok: false, error: r.reason }
  ) as any;
}

// Race with timeout
async function withTimeout<T>(promise: Promise<T>, ms: number, message?: string): Promise<T> {
  const timeout = new Promise<never>((_, reject) =>
    setTimeout(() => reject(new Error(message ?? `Timed out after ${ms}ms`)), ms)
  );
  return Promise.race([promise, timeout]);
}

// Retry with exponential backoff
async function retry<T>(
  fn: () => Promise<T>,
  options: { maxAttempts?: number; delay?: number; backoff?: number } = {}
): Promise<T> {
  const { maxAttempts = 3, delay = 1000, backoff = 2 } = options;
  let lastError: unknown;
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastError = err;
      if (attempt < maxAttempts) {
        await new Promise(r => setTimeout(r, delay * Math.pow(backoff, attempt - 1)));
      }
    }
  }
  throw lastError;
}

// Usage
const [user, posts] = await settleAll([fetchUser(1), fetchPosts(1)] as const);
if (user.ok && posts.ok) {
  displayDashboard(user.value, posts.value);
}

class AsyncQueue<T> {
  private queue: Array<{ item: T; resolve: () => void }> = [];
  private workers = 0;
  private readonly maxWorkers: number;
  private readonly processor: (item: T) => Promise<void>;

  constructor(processor: (item: T) => Promise<void>, maxWorkers = 5) {
    this.processor = processor;
    this.maxWorkers = maxWorkers;
  }

  async add(item: T): Promise<void> {
    return new Promise(resolve => {
      this.queue.push({ item, resolve });
      this.processNext();
    });
  }

  private async processNext(): Promise<void> {
    if (this.workers >= this.maxWorkers || this.queue.length === 0) return;

    this.workers++;
    const { item, resolve } = this.queue.shift()!;

    try {
      await this.processor(item);
    } finally {
      resolve();
      this.workers--;
      this.processNext();
    }
  }

  async drain(): Promise<void> {
    while (this.queue.length > 0 || this.workers > 0) {
      await new Promise(r => setTimeout(r, 10));
    }
  }
}

// Usage
const emailQueue = new AsyncQueue<Email>(
  async (email) => { await sendEmail(email); },
  10 // max 10 concurrent emails
);

for (const user of users) {
  await emailQueue.add({ to: user.email, subject: "Welcome!" });
}
await emailQueue.drain();

// TC39 Stage 3 Signals (available in 2026)
import { Signal } from 'signal-polyfill';

// Reactive primitive
const count = new Signal.State(0);
const doubled = new Signal.Computed(() => count.get() * 2);
const greeting = new Signal.Computed(() => `Count is ${count.get()}, doubled: ${doubled.get()}`);

// In React (via @preact/signals-react or use-signals)
function Counter() {
  return (
    <div>
      <p>{greeting}</p>
      <button onClick={() => count.set(count.get() + 1)}>Increment</button>
    </div>
  );
}

// Server-side signals for streaming
const streamData = new Signal.State<string[]>([]);

// Push data as it arrives
async function streamFromAPI() {
  const response = await fetch('/api/stream');
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value);
    streamData.set([...streamData.get(), chunk]); // reactive update
  }
}

// Proper cancellation with AbortController
class CancellableFetcher {
  private controllers = new Map<string, AbortController>();

  async fetch<T>(key: string, url: string, options?: RequestInit): Promise<T> {
    // Cancel previous request with same key
    this.cancel(key);

    const controller = new AbortController();
    this.controllers.set(key, controller);

    try {
      const response = await fetch(url, {
        ...options,
        signal: controller.signal,
      });

      if (!response.ok) {
        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
      }

      return response.json() as Promise<T>;
    } catch (err) {
      if (err instanceof DOMException && err.name === 'AbortError') {
        throw new Error('Request cancelled');
      }
      throw err;
    } finally {
      this.controllers.delete(key);
    }
  }

  cancel(key: string): void {
    this.controllers.get(key)?.abort();
  }

  cancelAll(): void {
    this.controllers.forEach(c => c.abort());
    this.controllers.clear();
  }
}

// React hook with auto-cancellation
function useAsync<T>(fn: () => Promise<T>, deps: unknown[]) {
  const [state, setState] = useState<{
    data: T | null; loading: boolean; error: Error | null
  }>({ data: null, loading: true, error: null });

  useEffect(() => {
    const controller = new AbortController();
    let cancelled = false;

    fn()
      .then(data => !cancelled && setState({ data, loading: false, error: null }))
      .catch(err => !cancelled && setState({ data: null, loading: false, error: err }));

    return () => {
      cancelled = true;
      controller.abort();
    };
  }, deps);

  return state;
}

// Create an observable stream from events
async function* fromEvents<T>(
  emitter: EventTarget,
  event: string,
  signal?: AbortSignal
): AsyncGenerator<T> {
  const queue: T[] = [];
  let resolve: (() => void) | null = null;

  const handler = (e: Event) => {
    queue.push((e as CustomEvent<T>).detail);
    resolve?.();
    resolve = null;
  };

  emitter.addEventListener(event, handler);
  signal?.addEventListener('abort', () => emitter.removeEventListener(event, handler));

  try {
    while (!signal?.aborted) {
      if (queue.length > 0) {
        yield queue.shift()!;
      } else {
        await new Promise<void>(r => { resolve = r; });
      }
    }
  } finally {
    emitter.removeEventListener(event, handler);
  }
}

// Pipeline operators for async iterables
async function* map<T, U>(iterable: AsyncIterable<T>, fn: (x: T) => U): AsyncGenerator<U> {
  for await (const item of iterable) {
    yield fn(item);
  }
}

async function* filter<T>(iterable: AsyncIterable<T>, pred: (x: T) => boolean): AsyncGenerator<T> {
  for await (const item of iterable) {
    if (pred(item)) yield item;
  }
}

async function* take<T>(iterable: AsyncIterable<T>, n: number): AsyncGenerator<T> {
  let count = 0;
  for await (const item of iterable) {
    if (count++ >= n) break;
    yield item;
  }
}

// Usage: live data pipeline
const controller = new AbortController();
const events = fromEvents<SensorReading>(sensorHub, 'reading', controller.signal);
const filtered = filter(events, r => r.value > threshold);
const mapped = map(filtered, r => ({ ...r, normalized: r.value / maxValue }));
const firstTen = take(mapped, 10);

for await (const reading of firstTen) {
  dashboard.update(reading);
}