Stream Collection

/**
 * Converts a stream to a byte array.
 * Supports both Node.js Readable streams and Web API ReadableStreams.
 */
const streamCollector: StreamCollector;

/**
 * Type definition for the stream collector function
 */
type StreamCollector = (stream: Readable | ReadableStream): Promise<Uint8Array>;

import { streamCollector } from "@smithy/node-http-handler";
import { Readable } from "stream";
import { createReadStream } from "fs";

// Collect from Node.js Readable stream
const fileStream = createReadStream("data.txt");
const fileBytes = await streamCollector(fileStream);
console.log(fileBytes); // Uint8Array containing file contents

// Collect from string-based stream  
const textStream = Readable.from(["Hello", " ", "World"]);
const textBytes = await streamCollector(textStream);
const text = new TextDecoder().decode(textBytes);
console.log(text); // "Hello World"

// Collect from Web API ReadableStream (Node.js 18+)
const response = await fetch("https://api.example.com/data");
if (response.body) {
  const responseBytes = await streamCollector(response.body);
  const responseText = new TextDecoder().decode(responseBytes);
  console.log(responseText);
}

// Collect from HTTP response stream
import { NodeHttpHandler } from "@smithy/node-http-handler";
import { HttpRequest } from "@smithy/protocol-http";

const handler = new NodeHttpHandler();
const request = new HttpRequest({
  method: "GET",
  hostname: "api.example.com",
  path: "/large-file"
});

const { response } = await handler.handle(request);
if (response.body) {
  const bodyBytes = await streamCollector(response.body);
  console.log(`Downloaded ${bodyBytes.length} bytes`);
}

/**
 * Internal writable stream for collecting data chunks.
 * Extends Node.js Writable stream.
 */
class Collector extends Writable {
  /**
   * Array of collected buffer chunks
   */
  bufferedBytes: Buffer[];
  
  /**
   * Create a new collector instance - inherits Writable constructor
   * @param options - WritableOptions for the underlying Writable stream
   */
  constructor(options?: WritableOptions);
  
  /**
   * Stream write implementation
   * @param chunk - Data chunk to write
   * @param encoding - Character encoding (if applicable)
   * @param callback - Completion callback
   */
  _write(chunk: Buffer, encoding: string, callback: (err?: Error) => void): void;
}

// Traditional Node.js streams
const nodeStream = new Readable({
  read() {
    this.push("chunk 1");
    this.push("chunk 2");
    this.push(null); // End stream
  }
});

const bytes = await streamCollector(nodeStream);

// Web API streams (Node.js 18+)
const webStream = new ReadableStream({
  start(controller) {
    controller.enqueue(new TextEncoder().encode("Hello"));
    controller.enqueue(new TextEncoder().encode(" World"));
    controller.close();
  }
});

const bytes = await streamCollector(webStream);

import { streamCollector } from "@smithy/node-http-handler";
import { Readable } from "stream";

// Handle stream errors
const errorStream = new Readable({
  read() {
    // Simulate an error
    this.emit("error", new Error("Stream processing failed"));
  }
});

try {
  const bytes = await streamCollector(errorStream);
} catch (error) {
  console.error("Stream collection failed:", error.message);
}

// Handle network stream errors
import { NodeHttpHandler } from "@smithy/node-http-handler";

const handler = new NodeHttpHandler();
try {
  const { response } = await handler.handle(request);
  const bodyBytes = await streamCollector(response.body);
} catch (error) {
  if (error.code === "ECONNRESET") {
    console.error("Connection was reset during stream collection");
  } else if (error.code === "ETIMEDOUT") {
    console.error("Stream collection timed out");
  } else {
    console.error("Stream collection error:", error.message);
  }
}

// From Node.js stream module
interface Readable {
  pipe<T extends Writable>(destination: T): T;
  on(event: string, listener: Function): this;
  // ... other Readable properties and methods
}

// From stream/web module (Node.js 16.5.0+)
interface ReadableStream<R = any> {
  getReader(): ReadableStreamDefaultReader<R>;
  // ... other ReadableStream properties and methods
}

// From Node.js stream module
abstract class Writable {
  abstract _write(chunk: any, encoding: string, callback: (error?: Error | null) => void): void;
  end(): void;
  on(event: string, listener: Function): this;
  // ... other Writable properties and methods
}

// From Node.js stream module  
interface WritableOptions {
  highWaterMark?: number;
  objectMode?: boolean;
  // ... other Writable options
}