AbortController & Streaming Fetch

AbortController
  ├── .signal  ──────────────────► fetch(url, { signal })
  │                                  └── rejects with AbortError when aborted
  └── .abort() ── triggers ──────► signal.aborted = true
                                   signal.dispatchEvent("abort")

response.body (ReadableStream)
  └── .getReader()          → ReadableStreamDefaultReader
        └── .read()         → Promise<{ value: Uint8Array, done: boolean }>
              ├── value     → raw bytes of the current chunk
              └── done      → true when the stream is closed

// utils/fetch-with-abort.ts

export async function fetchWithAbort<T>(
  url: string,
  signal: AbortSignal,
): Promise<T> {
  const response = await fetch(url, { signal });

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

  return response.json() as Promise<T>;
}

// components/UserProfile.tsx
"use client";

import { useEffect, useState } from "react";
import { fetchWithAbort } from "@/utils/fetch-with-abort";

interface User {
  id: string;
  name: string;
  email: string;
}

export function UserProfile({ userId }: { userId: string }) {
  const [user, setUser] = useState<User | null>(null);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    const controller = new AbortController();

    fetchWithAbort<User>(`/api/users/${userId}`, controller.signal)
      .then(setUser)
      .catch((err) => {
        // AbortError is intentional — don't surface it as a UI error
        if (err.name !== "AbortError") {
          setError(err.message);
        }
      });

    // Cleanup: abort the request if userId changes or component unmounts.
    // The cleanup runs before the next effect — so a rapid userId change
    // cancels the previous request before starting a new one.
    return () => controller.abort();
  }, [userId]);

  if (error) return <p>Error: {error}</p>;
  if (!user) return <p>Loading...</p>;
  return <p>{user.name}</p>;
}

// components/ProductSearch.tsx
"use client";

import { useEffect, useState } from "react";

interface Product {
  id: string;
  name: string;
  price: number;
}

export function ProductSearch() {
  const [query, setQuery] = useState("");
  const [results, setResults] = useState<Product[]>([]);
  const [isLoading, setIsLoading] = useState(false);

  useEffect(() => {
    if (!query.trim()) {
      setResults([]);
      return;
    }

    const controller = new AbortController();
    setIsLoading(true);

    fetch(`/api/products/search?q=${encodeURIComponent(query)}`, {
      signal: controller.signal,
    })
      .then((res) => {
        if (!res.ok) throw new Error(`Search failed: ${res.status}`);
        return res.json() as Promise<Product[]>;
      })
      .then((data) => {
        setResults(data);
        setIsLoading(false);
      })
      .catch((err) => {
        if (err.name !== "AbortError") {
          console.error("Search error:", err.message);
          setIsLoading(false);
        }
        // AbortError: a newer request superseded this one — do nothing
      });

    // Each query change aborts the previous request before starting a new one
    return () => controller.abort();
  }, [query]);

  return (
    <div>
      <input
        value={query}
        onChange={(e) => setQuery(e.target.value)}
        placeholder="Search products..."
      />
      {isLoading && <span>Searching...</span>}
      <ul>
        {results.map((p) => (
          <li key={p.id}>
            {p.name} — ${p.price}
          </li>
        ))}
      </ul>
    </div>
  );
}

// lib/fetch-with-timeout.ts

export async function fetchWithTimeout<T>(
  url: string,
  options: RequestInit & { timeoutMs?: number } = {},
): Promise<T> {
  const { timeoutMs = 5000, ...fetchOptions } = options;

  // AbortSignal.timeout() is a newer static method that creates a signal
  // that aborts automatically after the given duration.
  // It's cleaner than manually managing setTimeout + AbortController.
  const signal = AbortSignal.timeout(timeoutMs);

  try {
    const response = await fetch(url, { ...fetchOptions, signal });

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

    return response.json() as Promise<T>;
  } catch (error) {
    if (error instanceof DOMException) {
      if (error.name === "TimeoutError") {
        throw new Error(`Request to ${url} timed out after ${timeoutMs}ms`);
      }
      if (error.name === "AbortError") {
        throw new Error(`Request to ${url} was cancelled`);
      }
    }
    throw error;
  }
}

// lib/abortable-operation.ts

export function combineSignals(...signals: AbortSignal[]): AbortSignal {
  const controller = new AbortController();

  for (const signal of signals) {
    // If any input signal is already aborted, abort immediately
    if (signal.aborted) {
      controller.abort(signal.reason);
      break;
    }

    // Otherwise, listen for abort on each signal
    signal.addEventListener("abort", () => {
      controller.abort(signal.reason);
    });
  }

  return controller.signal;
}

// Usage
async function loadDataWithControls(url: string, userSignal: AbortSignal) {
  const timeoutSignal = AbortSignal.timeout(8000);

  // Aborts if EITHER the user cancels OR the timeout fires
  const combinedSignal = combineSignals(userSignal, timeoutSignal);

  const response = await fetch(url, { signal: combinedSignal });
  return response.json();
}

// app/chat/page.tsx
"use client";

import { useState, useRef } from "react";

export default function ChatPage() {
  const [prompt, setPrompt] = useState("");
  const [output, setOutput] = useState("");
  const [isStreaming, setIsStreaming] = useState(false);
  const controllerRef = useRef<AbortController | null>(null);

  async function startStream() {
    // Abort any existing stream before starting a new one
    controllerRef.current?.abort();
    const controller = new AbortController();
    controllerRef.current = controller;

    setOutput("");
    setIsStreaming(true);

    try {
      const response = await fetch("/api/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ prompt }),
        signal: controller.signal,
      });

      if (!response.ok || !response.body) {
        throw new Error(`Stream failed: ${response.status}`);
      }

      const reader = response.body.getReader();
      // { stream: true } handles multi-byte characters across chunk boundaries
      const decoder = new TextDecoder("utf-8");

      while (true) {
        const { value, done } = await reader.read();

        if (done) break;

        const chunk = decoder.decode(value, { stream: true });
        setOutput((prev) => prev + chunk);
      }
    } catch (err) {
      if (err instanceof Error && err.name !== "AbortError") {
        console.error("Stream error:", err.message);
      }
    } finally {
      setIsStreaming(false);
    }
  }

  function stopStream() {
    controllerRef.current?.abort();
  }

  return (
    <div>
      <textarea
        value={prompt}
        onChange={(e) => setPrompt(e.target.value)}
        placeholder="Enter your prompt..."
        rows={4}
      />
      <div>
        <button onClick={startStream} disabled={isStreaming}>
          {isStreaming ? "Generating..." : "Generate"}
        </button>
        {isStreaming && <button onClick={stopStream}>Stop</button>}
      </div>
      <pre style={{ whiteSpace: "pre-wrap" }}>{output}</pre>
    </div>
  );
}

// app/api/chat/route.ts

import { NextRequest } from "next/server";

export async function POST(req: NextRequest) {
  const { prompt } = await req.json();

  // Simulate streaming token output — in production, pipe from OpenAI or similar
  const words =
    `You asked: "${prompt}". Here is a streamed response word by word.`.split(
      " ",
    );

  const stream = new ReadableStream({
    async start(controller) {
      for (const word of words) {
        controller.enqueue(new TextEncoder().encode(word + " "));
        // Simulate per-token delay
        await new Promise((r) => setTimeout(r, 60));
      }
      controller.close();
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/plain; charset=utf-8",
      // Disable buffering on proxies — required for true streaming
      "X-Accel-Buffering": "no",
    },
  });
}

// lib/stream-processor.ts

async function processStream(
  stream: ReadableStream<Uint8Array>,
  onChunk: (text: string) => void,
  signal: AbortSignal,
): Promise<void> {
  const reader = stream.getReader();
  const decoder = new TextDecoder("utf-8");

  // Register abort handler to cancel the read loop
  signal.addEventListener("abort", () => reader.cancel(), { once: true });

  try {
    while (true) {
      const { value, done } = await reader.read();

      if (done) break;
      if (signal.aborted) break;

      const text = decoder.decode(value, { stream: true });

      // For expensive per-chunk work, batch renders using rAF
      // instead of calling setState on every single chunk
      requestAnimationFrame(() => onChunk(text));
    }
  } finally {
    // Always release the reader lock — even if an error occurs mid-stream
    reader.releaseLock();
  }
}

// ❌ Signal is already aborted — new fetch rejects immediately
controller.abort();
fetch(url, { signal: controller.signal }); // rejects before the request is even sent

// ✅ Fresh controller for each request
const newController = new AbortController();
fetch(url, { signal: newController.signal });

if (!response.body) {
  throw new Error("Response has no body to stream");
}
const reader = response.body.getReader();