jsonl-stream

A library for progressive parsing of JSON Lines (JSONL) — also known as NDJSON (Newline Delimited JSON) streams.

Specifically designed to process real-time data streams, such as AI model responses (LLM streaming like GPT/Gemini), massive log files, or any stream-based communication where network packets (chunks) arrive fragmented and incomplete over HTTP/HTTPS, WebSockets, gRPC, or local file system reads.

Key Features

• Smart Parsing: Uses an internal character-by-character decoder that detects incomplete structures and buffers them until the next chunk arrives.
• Complex Structures: Supports nested objects, complex arrays, strings with unicode escape characters (\uXXXX), booleans, null, and numbers in various notations (including scientific notation like -123.45e2).
• Stream API Compatibility: Exposes a JsonlStream class extending the native Node.js Transform class, integrating seamlessly with pipes and streams in object mode (readableObjectMode: true). Ideal for HTTP requests and other streaming protocols.
• Support for Diverse Line Endings: Works correctly with Windows (\r\n), Unix/modern macOS (\n), and classic macOS (\r) line endings, handling extra whitespaces, tabs, and indentation gracefully.
• Robust End-Of-File (EOF) Handling: Automatically processes and parses remaining buffered numbers or primitive values at the end of the stream, preventing data loss when the upstream source finishes.

Installation

npm install jsonl-stream

How to Use

The library is designed to be flexible. It works with the Node.js Stream API or by manually feeding chunks from persistent connections like WebSockets, gRPC, or Server-Sent Events (SSE).

1. Consuming data via HTTP Stream

Ideal for consuming APIs that send real-time data over HTTP/HTTPS.

Using fetch (Node.js 18+)

import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { JsonlStream } from "jsonl-stream";

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

async function consumeFetchStream() {
  const response = await fetch("https://api.example.com/stream");
  if (!response.body) {
    throw new Error("ReadableStream is not supported or empty");
  }

  // Type-safe conversion from fetch Web Stream to Node Stream
  const webStream = response.body as import("node:stream/web").ReadableStream<Uint8Array>;
  const readableStream = Readable.fromWeb(webStream);
  const jsonlParser = new JsonlStream<User>();

  try {
    // pipeline handles upstream and downstream errors automatically
    await pipeline(
      readableStream,
      jsonlParser,
      async function* (source) {
        for await (const item of source) {
          console.log("User received via HTTP:", item.name); // Typed as User
        }
      }
    );
    console.log("HTTP Stream finished!");
  } catch (error) {
    console.error("Error consuming the stream:", error);
  }
}

consumeFetchStream();

Using the https module

import https from "node:https";
import { pipeline } from "node:stream/promises";
import { JsonlStream } from "jsonl-stream";

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

https.get("https://api.example.com/stream-jsonl", async (response) => {
  const jsonlParser = new JsonlStream<User>();

  try {
    // Use pipeline to securely handle all errors in the stream chain
    await pipeline(
      response,
      jsonlParser,
      async function* (source) {
        for await (const item of source) {
          console.log("User received via HTTPS:", item.name); // Typed as User
        }
      }
    );
    console.log("HTTPS Stream successfully processed!");
  } catch (error) {
    console.error("Error in stream pipeline:", error);
  }
}).on("error", (error) => {
  console.error("Request connection error:", error);
});

2. Reading Local Files via Stream

Ideal for processing extremely large JSONL files without exceeding memory limits.

import { createReadStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { JsonlStream } from "jsonl-stream";

interface LogEntry {
  level: string;
  message: string;
  timestamp: string;
}

async function processFile() {
  const fileStream = createReadStream("data.jsonl");
  const jsonlParser = new JsonlStream<LogEntry>();

  try {
    await pipeline(
      fileStream,
      jsonlParser,
      async function* (source) {
        for await (const item of source) {
          console.log(`[${item.level}] Log read from file: ${item.message}`); // Typed as LogEntry
        }
      }
    );
    console.log("Local file fully processed!");
  } catch (error) {
    console.error("Error processing file:", error);
  }
}

processFile();

3. Configuration & Security Limits

You can configure limits and security boundaries by passing options to the JsonlStream constructor:

import { JsonlStream } from "jsonl-stream";

interface LogEntry {
  level: string;
  message: string;
  timestamp: string;
}

const jsonlParser = new JsonlStream<LogEntry>({
  maxDepth: 30,                     // Reject nesting levels above 30
  maxPayloadSizeBytes: 500 * 1024,  // Limit parsing buffer to 500KB
  maxFieldLength: 1024,             // Limit error context buffer length to 1024 characters
  maxItemsPerCall: 5000             // Limit how many records are queued from one write() at a time
});

Parameter Reference

| Parameter | Type | Default | Description | |-----------|------|---------|-------------| | maxDepth | number | 50 | Maximum nesting depth (arrays/objects) allowed in JSON records to prevent stack overflows and ReDoS attacks. | | maxPayloadSizeBytes | number | 1048576 (1MB) | Maximum size in bytes of a single JSONL record. Applies both to complete records and to incomplete data buffered across chunks, protecting against memory exhaustion DoS attacks. Enforced incrementally while parsing (not just after a record finishes), so a single oversized-but-well-formed record delivered in one write() is rejected before it is fully built in memory. | | maxFieldLength | number | 2048 (2KB) | Maximum length of the sanitized buffer excerpt attached to JsonParserError contexts. It does not limit the size of parsed fields — it only truncates diagnostic output, and sensitive keys (e.g. password, token, etc.) are redacted to prevent logging sensitive information. | | maxItemsPerCall | number | 10000 | Maximum number of records parsed and queued from a single write()/flush pass before the rest is deferred to a subsequent bounded pass. Bounds peak memory when a caller writes one very large buffer containing many small records, instead of relying entirely on natural chunk sizes from the transport. |

maxDepth is additionally capped at a hard ceiling of 1000, maxPayloadSizeBytes at 104857600 (100MB), maxFieldLength at 1048576 (1MB), and maxItemsPerCall at 1000000, regardless of the configured value — all four parameters must be finite positive numbers within these ceilings, and invalid configurations throw a JsonlConfigError at construction time instead of silently disabling protection.

Error handling

• JsonlConfigError (extends RangeError) is thrown synchronously — at new JsonlStream(...) or parseJsonlStream(...) call time — when a limit option is missing, non-finite, non-positive, or exceeds its hard ceiling. It carries code: "INVALID_CONFIG", plus param (the offending option name) and value (what was supplied), for programmatic handling instead of parsing the message text.

• JsonParserError (extends Error) is emitted asynchronously (as a stream "error" event, or thrown by parseJsonlStream) for data-level failures. It carries a structured code: JsonlErrorCode in addition to the human-readable message:

  | Code | Meaning | |------|---------| | UNEXPECTED_TOKEN | Malformed JSON syntax (bad literal, missing ,/:, unexpected character, etc.) | | INVALID_NUMBER | A number token doesn't conform to RFC 8259 grammar | | INVALID_ESCAPE | An invalid or malformed string escape sequence | | UNESCAPED_CONTROL_CHARACTER | A raw control character (U+0000–U+001F) inside a string | | MAX_DEPTH_EXCEEDED | Nesting exceeded maxDepth | | PAYLOAD_TOO_LARGE | A record (complete or still-incomplete) exceeded maxPayloadSizeBytes | | MULTIPLE_RECORDS_PER_LINE | Two JSON values found on the same line without a newline separator | | INCOMPLETE_AT_EOF | The stream ended mid-structure | | UNKNOWN | An unexpected internal error was wrapped as-is |

  Prefer branching on error.code over matching error.message, since message wording may change between versions.

Behavior notes

• Strict RFC 8259 parsing: leading zeros, trailing dots, unescaped control characters in strings, and multiple records on the same line are rejected. A UTF-8 BOM (U+FEFF) at the start of the input is also rejected, matching JSON.parse — strip it before writing if your source emits one.
• Incomplete data at EOF is an error: when the stream ends (or parseJsonlStream is called with isEnd = true) with an unterminated record, a JsonParserError is raised instead of silently dropping the partial data.
• Prototype pollution safe: a "__proto__" key becomes a plain own property of the parsed object and never mutates the object's prototype. Duplicate keys follow last-wins semantics, like JSON.parse.
• Error contexts include input excerpts: JsonParserError.context.buffer carries a redacted, truncated (maxFieldLength) excerpt of the offending input to aid debugging. Be aware of this if you forward errors to logs with strict data-retention rules.

🧪 Development and Testing

If you are developing or testing the library locally, use the following npm commands:

Build

Compiles TypeScript code, generating CJS, ESM bundles, and type definitions in the dist/ folder:

npm run build

Tests

Runs the full test suite using Vitest, including empirical memory-bound checks under large synthetic JSONL payloads (src/memory.test.ts):

npm test

Benchmarks

Reports parsing throughput (records/s, MB/s) for representative chunk sizes using Vitest's benchmarking mode:

npm run bench

Issues & Feedback

If you encounter any bugs, have questions, or would like to request a new feature, please feel free to open an issue on GitHub. Your feedback and contributions are highly welcome!

License

This project is licensed under the MIT License. See the LICENSE file for details.