Fetch, Streams, and AbortController

3 collapsed lines
1
// Imports and setup (collapsed)
2
const API_BASE = "https://api.example.com"
3
const signal = AbortSignal.timeout(5000)
4

5
// Key pattern: structured request with explicit configuration
6
const response = await fetch(`${API_BASE}/users`, {
7
  method: "POST",
8
  headers: { "Content-Type": "application/json" },
9
  body: JSON.stringify({ name: "Alice", role: "admin" }),
10
  credentials: "include", // Send cookies cross-origin
11
  signal, // 5-second timeout
12
})
13

14
// Error handling pattern (collapsed)
5 collapsed lines
15
if (!response.ok) {
16
  throw new Error(`HTTP ${response.status}: ${response.statusText}`)
17
}
18

19
const user = await response.json()

1
const response = await fetch("/api/data")
2
const data1 = await response.json()
3
const data2 = await response.json() // ❌ TypeError: body already consumed

1
// Option 1: Clone the response before consuming
2
const response = await fetch("/api/data")
3
const clone = response.clone() // Creates independent stream via tee()
4
const data1 = await response.json()
5
const data2 = await clone.json() // ✅ Works
6

7
// Option 2: Read once, use multiple times
8
const response = await fetch("/api/data")
9
const data = await response.json()
10
// Use `data` multiple times (it's a plain object now)

1
// ❌ Common mistake: assuming 404 throws
2
try {
3
  const response = await fetch("/nonexistent")
4
  const data = await response.json() // May throw parsing error, not 404
5
} catch (e) {
6
  // Only catches network failures or JSON parse errors
7
}
8

9
// ✅ Correct pattern: check response.ok
10
const response = await fetch("/api/data")
11
if (!response.ok) {
12
  // response.ok is false for status outside 200-299
13
  throw new Error(`HTTP ${response.status}`)
14
}
15
const data = await response.json()

2 collapsed lines
1
// Type definitions (collapsed)
2
type Chunk = Uint8Array | string
3

4
const stream = new ReadableStream<Chunk>(
5
  {
6
    // Called once when stream is created
7
    start(controller) {
8
      // Initial setup, return Promise to delay first pull()
9
    },
10

11
    // Called when internal queue needs more data
12
    // For pull sources: fetch data here
13
    // For push sources: often unused (data pushed via enqueue)
14
    pull(controller) {
15
      // controller.enqueue(chunk) - add data
16
      // controller.close() - signal end of data
17
      // controller.error(e) - signal failure
18
      // controller.desiredSize - bytes/chunks wanted (backpressure signal)
19
    },
20

21
    // Called when consumer cancels (e.g., reader.cancel())
6 collapsed lines
22
    cancel(reason) {
23
      // Cleanup: close connections, release resources
24
      // reason contains cancellation cause
25
    },
26
  },
27
  // Queuing strategy
28
  {
29
    highWaterMark: 3, // Queue up to 3 chunks before backpressure
30
    size: (chunk) => 1, // Each chunk counts as 1 (or return byte length)
31
  },
32
)

4 collapsed lines
1
// Simulated slow consumer example
2
// This demonstrates backpressure without any explicit coordination
3
const CHUNK_SIZE = 1024
4

5
const stream = new ReadableStream({
6
  start(controller) {
7
    let offset = 0
8

9
    const produceChunk = () => {
10
      // Key insight: check desiredSize before producing
11
      if (controller.desiredSize !== null && controller.desiredSize <= 0) {
12
        // Consumer is slow—wait before producing more
13
        // This prevents memory exhaustion automatically
14
        setTimeout(produceChunk, 100)
15
        return
16
      }
17

18
      const chunk = generateData(offset, CHUNK_SIZE)
19
      controller.enqueue(chunk)
20
      offset += CHUNK_SIZE
21

22
      if (offset < TOTAL_SIZE) {
23
        // Schedule next chunk
24
        setTimeout(produceChunk, 0)
25
      } else {
26
        controller.close()
27
      }
28
    }
29

6 collapsed lines
30
    produceChunk()
31
  },
32
})
33

34
// Helper function (collapsed)
35
function generateData(offset: number, size: number): Uint8Array {
36
  return new Uint8Array(size).fill(offset % 256)
37
}

1
const reader = stream.getReader()
2

3
try {
4
  while (true) {
5
    const { done, value } = await reader.read()
6
    if (done) break
7
    processChunk(value)
8
  }
9
} finally {
10
  reader.releaseLock() // Always release to allow other readers
11
}

1
// Locks stream during iteration
2
for await (const chunk of stream) {
3
  processChunk(chunk)
4
}
5
// Stream is automatically cancelled on break/throw unless preventCancel: true
6

7
// To continue reading after early exit:
8
for await (const chunk of stream.values({ preventCancel: true })) {
9
  if (foundTarget) break // Stream remains open
10
}

1
// Pipe handles reading, backpressure, and cleanup automatically
2
await stream.pipeTo(writableStream)

1
const reader1 = stream.getReader()
2
const reader2 = stream.getReader() // ❌ TypeError: stream is locked
3

4
reader1.releaseLock()
5
const reader2 = stream.getReader() // ✅ Now works

1
const [stream1, stream2] = originalStream.tee()
2

3
// Both receive the same data independently
4
// Common use: stream to browser AND cache simultaneously

3 collapsed lines
1
// DOM setup (collapsed)
2
const outputElement = document.getElementById("output")!
3
let lineCount = 0
4

5
const writableStream = new WritableStream({
6
  // Called once when stream is created
7
  start(controller) {
8
    // Initialize sink, return Promise to delay writes
9
  },
10

11
  // Called for each chunk written
12
  write(chunk, controller) {
13
    // Process chunk (can be async)
14
    outputElement.textContent += chunk
15
    lineCount++
16

17
    // controller.error(e) to abort with error
18
  },
19

20
  // Called when writer.close() is invoked
21
  close() {
22
    console.log(`Finished writing ${lineCount} lines`)
23
  },
24

6 collapsed lines
25
  // Called on writer.abort() or upstream error
26
  abort(reason) {
27
    console.error("Stream aborted:", reason)
28
    // Cleanup resources
29
  },
30
})
31

32
const writer = writableStream.getWriter()
33
await writer.write("Line 1\n")
34
await writer.write("Line 2\n")
35
await writer.close()

1
const uppercaseTransform = new TransformStream({
2
  transform(chunk, controller) {
3
    // Transform and forward
4
    controller.enqueue(chunk.toUpperCase())
5
  },
6

7
  flush(controller) {
8
    // Called after all input processed, before output closes
9
    // Useful for emitting buffered data
10
  },
11
})
12

13
// Use with pipeThrough
14
const result = await fetch("/text.txt")
15
const stream = response.body.pipeThrough(new TextDecoderStream()).pipeThrough(uppercaseTransform)
16

17
for await (const line of stream) {
18
  console.log(line) // UPPERCASE TEXT
19
}

1
await readableStream.pipeTo(writableStream, {
2
  preventClose: false, // Default: close destination when source closes
3
  preventAbort: false, // Default: abort destination on source error
4
  preventCancel: false, // Default: cancel source on destination error
5
  signal: abortSignal, // Cancel pipe on abort
6
})

1
const controller = new AbortController()
2
const signal = controller.signal
3

4
// Pass signal to cancellable operation
5
const response = await fetch("/slow-endpoint", { signal })
6

7
// Cancel from anywhere
8
controller.abort()
9
// OR with reason
10
controller.abort(new Error("User navigated away"))

1
try {
2
  const response = await fetch("/api/data", {
3
    signal: AbortSignal.timeout(5000), // 5-second timeout
4
  })
5
  const data = await response.json()
6
} catch (e) {
7
  if (e.name === "TimeoutError") {
8
    // Timeout-specific handling
9
    console.error("Request timed out")
10
  } else if (e.name === "AbortError") {
11
    // Manual abort (different from timeout)
12
    console.error("Request was cancelled")
13
  }
14
}

1
const controller = new AbortController()
2
const timeoutSignal = AbortSignal.timeout(10000)
3

4
// Abort on user action OR timeout
5
const response = await fetch("/api/data", {
6
  signal: AbortSignal.any([controller.signal, timeoutSignal]),
7
})
8

9
// User can still cancel manually
10
cancelButton.onclick = () => controller.abort()

1
async function processItems(items: Item[], signal: AbortSignal) {
2
  for (const item of items) {
3
    // Check before each expensive operation
4
    signal.throwIfAborted()
5
    await expensiveProcess(item)
6
  }
7
}

1
const controller = new AbortController()
2
controller.abort()
3

4
// These both reject immediately with AbortError
5
await fetch("/a", { signal: controller.signal }) // ❌ Rejected
6
await fetch("/b", { signal: controller.signal }) // ❌ Rejected
7

8
// Create new controller for each operation that needs independent cancellation

3 collapsed lines
1
// Helper types (collapsed)
2
interface ProcessOptions {
3
  signal?: AbortSignal
4
}
5
type ProcessResult = { processed: number }
6

7
async function processWithAbort(data: AsyncIterable<Chunk>, options: ProcessOptions = {}): Promise<ProcessResult> {
8
  const { signal } = options
9
  let processed = 0
10

11
  // Check if already aborted
12
  signal?.throwIfAborted()
13

14
  for await (const chunk of data) {
15
    // Check before each chunk
16
    if (signal?.aborted) {
17
      throw signal.reason
18
    }
19

20
    await processChunk(chunk)
21
    processed++
22
  }
23

24
  return { processed }
25
}
26

27
// Listen for abort during long operations
8 collapsed lines
28
function longOperation(signal: AbortSignal): Promise<void> {
29
  return new Promise((resolve, reject) => {
30
    // Handle abort
31
    signal.addEventListener("abort", () => reject(signal.reason), { once: true })
32

33
    // Do work...
34
    setTimeout(resolve, 10000)
35
  })
36
}

4 collapsed lines
1
// Progress callback type (collapsed)
2
type ProgressCallback = (loaded: number, total: number | null) => void
3
interface DownloadResult {
4
  data: Uint8Array
5
  size: number
6
}
7

8
async function downloadWithProgress(
9
  url: string,
10
  onProgress: ProgressCallback,
11
  signal?: AbortSignal,
12
): Promise<DownloadResult> {
13
  const response = await fetch(url, { signal })
14
  if (!response.ok) throw new Error(`HTTP ${response.status}`)
15

16
  // Total may be null if server doesn't send Content-Length
17
  const total = response.headers.get("Content-Length")
18
  const totalBytes = total ? parseInt(total, 10) : null
19

20
  const reader = response.body!.getReader()
21
  const chunks: Uint8Array[] = []
22
  let loaded = 0
23

24
  while (true) {
25
    const { done, value } = await reader.read()
26
    if (done) break
27

28
    chunks.push(value)
29
    loaded += value.length
30
    onProgress(loaded, totalBytes)
31
  }
9 collapsed lines
32

33
  // Concatenate chunks into single array
34
  const result = new Uint8Array(loaded)
35
  let offset = 0
36
  for (const chunk of chunks) {
37
    result.set(chunk, offset)
38
    offset += chunk.length
39
  }
40

41
  return { data: result, size: loaded }
42
}

2 collapsed lines
1
// Import (collapsed)
2
type LineCallback = (line: string) => void
3

4
async function processLines(response: Response, onLine: LineCallback): Promise<void> {
5
  const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader()
6

7
  let buffer = ""
8

9
  while (true) {
10
    const { done, value } = await reader.read()
11

12
    if (done) {
13
      // Emit final line if buffer has content
14
      if (buffer) onLine(buffer)
15
      break
16
    }
17

18
    buffer += value
19
    const lines = buffer.split("\n")
20

21
    // Process all complete lines
22
    for (let i = 0; i < lines.length - 1; i++) {
23
      onLine(lines[i])
24
    }
25

26
    // Keep incomplete line in buffer
27
    buffer = lines[lines.length - 1]
28
  }
29
}
6 collapsed lines
30

31
// Usage with Server-Sent Events format
32
await processLines(response, (line) => {
33
  if (line.startsWith("data: ")) {
34
    const data = JSON.parse(line.slice(6))
35
    handleEvent(data)
36
  }
37
})

3 collapsed lines
1
// Type definitions (collapsed)
2
type JSONValue = string | number | boolean | null | JSONValue[] | { [key: string]: JSONValue }
3

4
async function* streamNDJSON<T extends JSONValue>(response: Response): AsyncGenerator<T> {
5
  const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader()
6

7
  let buffer = ""
8

9
  while (true) {
10
    const { done, value } = await reader.read()
11

12
    if (done) {
13
      if (buffer.trim()) {
14
        yield JSON.parse(buffer) as T
15
      }
16
      return
17
    }
18

19
    buffer += value
20
    const lines = buffer.split("\n")
21
    buffer = lines.pop()! // Keep incomplete line
22

23
    for (const line of lines) {
24
      if (line.trim()) {
25
        yield JSON.parse(line) as T
26
      }
27
    }
28
  }
29
}
30

31
// Usage
32
for await (const record of streamNDJSON<UserRecord>(response)) {
33
  await processUser(record)
34
}

5 collapsed lines
1
// Feature detection for streaming uploads
2
const supportsRequestStreams = (() => {
3
  let duplexAccessed = false
4
  const hasContentType = new Request("", {
5
    body: new ReadableStream(),
6
    method: "POST",
7
    get duplex() {
8
      duplexAccessed = true
9
      return "half"
10
    },
11
  }).headers.has("Content-Type")
12
  // If duplex getter was called AND Content-Type wasn't auto-set to text/plain,
13
  // browser supports streaming
14
  return duplexAccessed && !hasContentType
15
})()
16

17
async function uploadStream(
18
  url: string,
19
  dataGenerator: AsyncGenerator<Uint8Array>,
20
  signal?: AbortSignal,
21
): Promise<Response> {
22
  const stream = new ReadableStream({
23
    async pull(controller) {
24
      const { done, value } = await dataGenerator.next()
25
      if (done) {
26
        controller.close()
27
      } else {
28
        controller.enqueue(value)
29
      }
30
    },
31
  })
32

33
  return fetch(url, {
34
    method: "POST",
11 collapsed lines
35
    body: stream,
36
    duplex: "half", // Required for streaming body
37
    signal,
38
  })
39
}
40

41
// Example: stream file in chunks
42
async function* readFileChunks(file: File, chunkSize = 64 * 1024) {
43
  let offset = 0
44
  while (offset < file.size) {
45
    const slice = file.slice(offset, offset + chunkSize)
46
    const buffer = await slice.arrayBuffer()
47
    yield new Uint8Array(buffer)
48
    offset += chunkSize
49
  }
50
}
51

52
await uploadStream("/api/upload", readFileChunks(largeFile))

3 collapsed lines
1
// Progress types (collapsed)
2
type UploadProgress = { loaded: number; total: number }
3
type ProgressHandler = (progress: UploadProgress) => void
4

5
function uploadWithProgress(
6
  url: string,
7
  formData: FormData,
8
  onProgress: ProgressHandler,
9
  signal?: AbortSignal,
10
): Promise<Response> {
11
  return new Promise((resolve, reject) => {
12
    const xhr = new XMLHttpRequest()
13

14
    xhr.upload.addEventListener("progress", (e) => {
15
      if (e.lengthComputable) {
16
        onProgress({ loaded: e.loaded, total: e.total })
17
      }
18
    })
19

20
    xhr.addEventListener("load", () => {
21
      resolve(new Response(xhr.response, { status: xhr.status }))
22
    })
23

24
    xhr.addEventListener("error", () => reject(new Error("Upload failed")))
11 collapsed lines
25

26
    // Handle abort signal
27
    signal?.addEventListener("abort", () => {
28
      xhr.abort()
29
      reject(signal.reason)
30
    })
31

32
    xhr.open("POST", url)
33
    xhr.send(formData)
34
  })
35
}

4 collapsed lines
1
// BYOB is only available on byte streams (type: 'bytes')
2
// Example: efficient binary file processing
3

4
async function readWithBYOB(stream: ReadableStream<Uint8Array>): Promise<Uint8Array[]> {
5
  // Get BYOB reader (only works with byte streams)
6
  const reader = stream.getReader({ mode: "byob" })
7
  const chunks: Uint8Array[] = []
8

9
  // Reusable buffer—same memory gets reused across reads
10
  let buffer = new Uint8Array(65536) // 64KB buffer
11

12
  while (true) {
13
    // Read directly into our buffer
14
    const { done, value } = await reader.read(buffer)
15

16
    if (done) break
17

18
    // value is a view into the same ArrayBuffer, but potentially
19
    // a different length than buffer. Save the actual data.
20
    chunks.push(value.slice()) // Copy the filled portion
21

22
    // The original buffer is "detached"—we need a new view
23
    // value.buffer is the underlying ArrayBuffer
24
    buffer = new Uint8Array(value.buffer)
25
  }
26

27
  return chunks
28
}

5 collapsed lines
1
// Error types for discrimination
2
class NetworkError extends Error {
3
  name = "NetworkError"
4
}
5
class HTTPError extends Error {
6
  constructor(
7
    public status: number,
8
    public statusText: string,
9
  ) {
10
    super(`HTTP ${status}: ${statusText}`)
11
    this.name = "HTTPError"
12
  }
13
}
14

15
async function robustFetch<T>(url: string, options: RequestInit = {}): Promise<T> {
16
  const { signal, ...restOptions } = options
17

18
  // Combine user signal with timeout
19
  const timeoutSignal = AbortSignal.timeout(30000)
20
  const combinedSignal = signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal
21

22
  let response: Response
23

24
  try {
25
    response = await fetch(url, { ...restOptions, signal: combinedSignal })
26
  } catch (e) {
27
    if (e instanceof Error) {
28
      if (e.name === "TimeoutError") {
29
        throw new NetworkError("Request timed out after 30 seconds")
30
      }
31
      if (e.name === "AbortError") {
32
        throw e // Re-throw user-initiated abort
33
      }
34
    }
35
    throw new NetworkError(`Network request failed: ${e}`)
36
  }
37

38
  if (!response.ok) {
39
    throw new HTTPError(response.status, response.statusText)
40
  }
41

42
  // Handle empty responses
43
  const contentType = response.headers.get("Content-Type")
44
  if (!contentType?.includes("application/json")) {
11 collapsed lines
45
    throw new Error(`Expected JSON, got ${contentType}`)
46
  }
47

48
  return response.json() as Promise<T>
49
}
50

51
// Usage with proper error discrimination
52
try {
53
  const data = await robustFetch<UserData>("/api/user")
54
} catch (e) {
55
  if (e instanceof HTTPError && e.status === 404) {
56
    // Handle not found
57
  } else if (e instanceof NetworkError) {
58
    // Handle connectivity issues
59
  } else if (e instanceof Error && e.name === "AbortError") {
60
    // User cancelled—usually no action needed
61
  } else {
62
    // Unknown error
63
    throw e
64
  }
65
}

3 collapsed lines
1
// Retry configuration (collapsed)
2
interface RetryConfig {
3
  maxRetries: number
4
  delayMs: number
5
}
6

7
async function streamWithRetry(
8
  url: string,
9
  config: RetryConfig = { maxRetries: 3, delayMs: 1000 },
10
): Promise<Uint8Array> {
11
  let lastError: Error | undefined
12
  let receivedBytes = 0
13
  const chunks: Uint8Array[] = []
14

15
  for (let attempt = 0; attempt < config.maxRetries; attempt++) {
16
    try {
17
      const headers: HeadersInit = {}
18

19
      // Resume from where we left off
20
      if (receivedBytes > 0) {
21
        headers["Range"] = `bytes=${receivedBytes}-`
22
      }
23

24
      const response = await fetch(url, { headers })
25

26
      // Check if server supports range requests
27
      if (receivedBytes > 0 && response.status !== 206) {
28
        // Server doesn't support resume—start over
29
        chunks.length = 0
30
        receivedBytes = 0
31
      }
32

33
      const reader = response.body!.getReader()
34

35
      while (true) {
36
        const { done, value } = await reader.read()
37
        if (done) break
38

39
        chunks.push(value)
40
        receivedBytes += value.length
41
      }
42

43
      // Success—concatenate and return
44
      const result = new Uint8Array(receivedBytes)
45
      let offset = 0
46
      for (const chunk of chunks) {
47
        result.set(chunk, offset)
48
        offset += chunk.length
49
      }
50
      return result
51
    } catch (e) {
52
      lastError = e as Error
53
      await new Promise((r) => setTimeout(r, config.delayMs * (attempt + 1)))
54
    }
55
  }
56

57
  throw lastError
58
}

1
// Streaming uploads
2
const supportsUploadStreaming = (() => {
3
  try {
4
    let duplexAccessed = false
5
    new Request("", {
6
      body: new ReadableStream(),
7
      method: "POST",
8
      get duplex() {
9
        duplexAccessed = true
10
        return "half"
11
      },
12
    })
13
    return duplexAccessed
14
  } catch {
15
    return false
16
  }
17
})()
18

19
// AbortSignal.any()
20
const supportsSignalAny = typeof AbortSignal.any === "function"
21

22
// AbortSignal.timeout()
23
const supportsSignalTimeout = typeof AbortSignal.timeout === "function"
24

25
// Async iteration on streams
26
const supportsAsyncIteration = typeof ReadableStream.prototype[Symbol.asyncIterator] === "function"