Stream API

/**
 * Stream interface for HTTP requests
 */
interface GotStream {
  /**
   * Creates a duplex stream for HTTP requests
   * @param url - Request URL (optional if specified in options)
   * @param options - Stream request options
   * @returns Duplex stream
   */
  (url?: string | URL, options?: StreamOptions): Request;
  
  // HTTP method variants
  get(url?: string | URL, options?: StreamOptions): Request;
  post(url?: string | URL, options?: StreamOptions): Request;
  put(url?: string | URL, options?: StreamOptions): Request;
  patch(url?: string | URL, options?: StreamOptions): Request;
  head(url?: string | URL, options?: StreamOptions): Request;
  delete(url?: string | URL, options?: StreamOptions): Request;
}

type StreamOptions = OptionsInit & {
  isStream?: true;
};

import got from "got";
import { pipeline } from "node:stream";
import { createWriteStream, createReadStream } from "node:fs";

// Download a file using streams
await pipeline(
  got.stream("https://example.com/large-file.zip"),
  createWriteStream("download.zip")
);

// Upload a file using streams
await pipeline(
  createReadStream("upload.txt"),
  got.stream.post("https://api.example.com/upload", {
    headers: { "Content-Type": "text/plain" }
  })
);

// Stream JSON processing
const stream = got.stream("https://api.example.com/data.json");
stream.pipe(/* JSON parser */).pipe(/* processor */);

/**
 * Request stream class extending Node.js Duplex stream
 */
class Request extends Duplex {
  /**
   * Request options
   */
  options: Options;
  
  /**
   * Response object (available after 'response' event)
   */
  response?: IncomingMessageWithTimings;
  
  /**
   * Request URL
   */
  requestUrl?: URL;
  
  /**
   * Whether request is aborted
   */
  isAborted: boolean;
}

/**
 * Event interface for request streams
 */
interface RequestEvents<T> {
  /**
   * Emitted when the request is sent
   */
  on(event: "request", listener: (request: ClientRequest) => void): T;
  
  /**
   * Emitted when response headers are received
   */
  on(event: "response", listener: (response: IncomingMessageWithTimings) => void): T;
  
  /**
   * Emitted on redirect
   */
  on(event: "redirect", listener: (updatedOptions: Options, plainResponse: PlainResponse) => void): T;
  
  /**
   * Emitted for upload progress
   */
  on(event: "uploadProgress", listener: (progress: Progress) => void): T;
  
  /**
   * Emitted for download progress
   */
  on(event: "downloadProgress", listener: (progress: Progress) => void): T;
  
  /**
   * Emitted on request error
   */
  on(event: "error", listener: (error: RequestError) => void): T;
  
  /**
   * Emitted before retry
   */
  on(event: "retry", listener: (retryCount: number, error: RequestError) => void): T;
}

interface Progress {
  /**
   * Number of bytes transferred
   */
  transferred: number;
  
  /**
   * Total number of bytes (if known)
   */
  total?: number;
  
  /**
   * Transfer percentage (0-1)
   */
  percent?: number;
}

import got from "got";
import { createWriteStream } from "node:fs";

// Download with progress tracking
const downloadStream = got.stream("https://example.com/large-file.zip");
const fileStream = createWriteStream("download.zip");

downloadStream.on("downloadProgress", (progress) => {
  console.log(`Downloaded: ${progress.transferred} bytes`);
  if (progress.total) {
    const percentage = Math.round((progress.transferred / progress.total) * 100);
    console.log(`Progress: ${percentage}%`);
  }
});

downloadStream.on("response", (response) => {
  console.log(`Status: ${response.statusCode}`);
  console.log(`Content-Length: ${response.headers['content-length']}`);
});

downloadStream.on("error", (error) => {
  console.error("Download failed:", error.message);
});

downloadStream.pipe(fileStream);

/**
 * Upload progress tracking interface
 */
interface UploadProgress extends Progress {
  /**
   * Number of bytes uploaded
   */
  transferred: number;
  
  /**
   * Total bytes to upload
   */
  total?: number;
}

import got from "got";
import { createReadStream } from "node:fs";

// Upload file with progress tracking
const fileStream = createReadStream("large-file.zip");
const uploadStream = got.stream.post("https://api.example.com/upload");

uploadStream.on("uploadProgress", (progress) => {
  console.log(`Uploaded: ${progress.transferred} bytes`);
  if (progress.total) {
    const percentage = Math.round((progress.transferred / progress.total) * 100);
    console.log(`Upload progress: ${percentage}%`);
  }
});

uploadStream.on("response", (response) => {
  console.log(`Upload completed with status: ${response.statusCode}`);
});

fileStream.pipe(uploadStream);

interface StreamRequestOptions extends OptionsInit {
  /**
   * Enable stream mode
   */
  isStream: true;
  
  /**
   * Decompress response automatically
   * @default true
   */
  decompress?: boolean;
  
  /**
   * Buffer size for stream operations
   */
  buffer?: boolean;
  
  /**
   * Request timeout settings
   */
  timeout?: Delays;
}

/**
 * Utility functions for stream operations
 */
namespace StreamUtils {
  /**
   * Check if an object is a stream
   */
  function isStream(object: unknown): object is NodeJS.ReadableStream;
  
  /**
   * Get content length from stream or buffer
   */
  function getBodySize(body: unknown): number | undefined;
}

import got from "got";
import { Transform } from "node:stream";
import { createWriteStream } from "node:fs";

// Transform stream for processing data
const transformStream = new Transform({
  transform(chunk, encoding, callback) {
    // Process chunk data
    const processed = chunk.toString().toUpperCase();
    callback(null, processed);
  }
});

// Chain streams with transformation
const downloadStream = got.stream("https://api.example.com/data.txt");
const fileStream = createWriteStream("processed-data.txt");

downloadStream
  .pipe(transformStream)
  .pipe(fileStream);

// Handle backpressure and errors
downloadStream.on("error", (error) => {
  console.error("Download error:", error);
  transformStream.destroy();
  fileStream.destroy();
});

// JSON streaming with custom parser
const jsonStream = got.stream("https://api.example.com/large-dataset.json", {
  responseType: "json"
});

jsonStream.on("data", (chunk) => {
  try {
    const data = JSON.parse(chunk.toString());
    console.log("Received data:", data);
  } catch (error) {
    console.error("JSON parse error:", error);
  }
});