Ndjson unstable

Encodes and decodes newline-delimited JSON streams in Effect channels.

NDJSON stores one complete JSON value on each line. This module has helpers for byte streams, string streams, and schema-checked records, so streaming code can read or write one JSON record at a time.

15 exports Since v4.0.0 Source

Combinators

duplex

unstable

Since v4.0.0

Wraps a bidirectional byte channel with NDJSON encoding and decoding.

Details

Outgoing values are written as UTF-8 NDJSON bytes, and incoming bytes are parsed as NDJSON values.

Signature

declare const duplex: {
  (options?: {
    readonly ignoreEmptyLines?: boolean;
    readonly maxLineBytes?: number;
  }): <R, IE, OE, OutDone, InDone>(self: Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], OE, OutDone, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], NdjsonError | IE, InDone, R>) => Channel<readonly [unknown, unknown], NdjsonError | OE, OutDone, readonly [unknown, unknown], IE, InDone, R>;
  <R, IE, OE, OutDone, InDone>(self: Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], OE, OutDone, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], NdjsonError | IE, InDone, R>, options?: {
    readonly ignoreEmptyLines?: boolean;
    readonly maxLineBytes?: number;
  }): Channel<readonly [unknown, unknown], NdjsonError | OE, OutDone, readonly [unknown, unknown], IE, InDone, R>;
}

duplexSchema

unstable

Since v4.0.0

Wraps a bidirectional byte channel with schema-aware NDJSON encoding and decoding.

Details

Values sent to the wrapped channel are encoded with inputSchema; bytes received from it are parsed as NDJSON and decoded with outputSchema.

Signature

declare const duplexSchema: {
  <In extends Constraint, Out extends Constraint>(options: {
    readonly ignoreEmptyLines?: boolean;
    readonly inputSchema: In;
    readonly maxLineBytes?: number;
    readonly outputSchema: Out;
  }): <OutErr, OutDone, InErr, InDone, R>(self: Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], OutErr, OutDone, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], SchemaError | NdjsonError | InErr, InDone, R>) => Channel<readonly [Out["Type"], Out["Type"]], SchemaError | NdjsonError | OutErr, OutDone, readonly [In["Type"], In["Type"]], InErr, InDone, R | In["EncodingServices"] | Out["DecodingServices"]>;
  <Out extends Constraint, In extends Constraint, OutErr, OutDone, InErr, InDone, R>(self: Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], OutErr, OutDone, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], SchemaError | NdjsonError | InErr, InDone, R>, options: {
    readonly ignoreEmptyLines?: boolean;
    readonly inputSchema: In;
    readonly maxLineBytes?: number;
    readonly outputSchema: Out;
  }): Channel<readonly [Out["Type"], Out["Type"]], SchemaError | NdjsonError | OutErr, OutDone, readonly [In["Type"], In["Type"]], InErr, InDone, R | In["EncodingServices"] | Out["DecodingServices"]>;
}

duplexSchemaString

unstable

Since v4.0.0

Wraps a bidirectional string channel with schema-aware NDJSON encoding and decoding.

Details

Values sent to the wrapped channel are encoded with inputSchema; strings received from it are parsed as NDJSON and decoded with outputSchema.

Signature

declare const duplexSchemaString: {
  <In extends Constraint, Out extends Constraint>(options: {
    readonly ignoreEmptyLines?: boolean;
    readonly inputSchema: In;
    readonly outputSchema: Out;
  }): <OutErr, OutDone, InErr, InDone, R>(self: Channel<readonly [string, string], OutErr, OutDone, readonly [string, string], SchemaError | NdjsonError | InErr, InDone, R>) => Channel<readonly [Out["Type"], Out["Type"]], SchemaError | NdjsonError | OutErr, OutDone, readonly [In["Type"], In["Type"]], InErr, InDone, R | In["EncodingServices"] | Out["DecodingServices"]>;
  <Out extends Constraint, In extends Constraint, OutErr, OutDone, InErr, InDone, R>(self: Channel<readonly [string, string], OutErr, OutDone, readonly [string, string], SchemaError | NdjsonError | InErr, InDone, R>, options: {
    readonly ignoreEmptyLines?: boolean;
    readonly inputSchema: In;
    readonly outputSchema: Out;
  }): Channel<readonly [Out["Type"], Out["Type"]], SchemaError | NdjsonError | OutErr, OutDone, readonly [In["Type"], In["Type"]], InErr, InDone, R | In["EncodingServices"] | Out["DecodingServices"]>;
}

duplexString

unstable

Since v4.0.0

Wraps a bidirectional string channel with NDJSON encoding and decoding.

Details

Outgoing values are written as NDJSON strings, and incoming strings are parsed as NDJSON values.

Signature

declare const duplexString: {
  (options?: {
    readonly ignoreEmptyLines?: boolean;
  }): <R, IE, OE, OutDone, InDone>(self: Channel<readonly [string, string], OE, OutDone, readonly [string, string], NdjsonError | IE, InDone, R>) => Channel<readonly [unknown, unknown], NdjsonError | OE, OutDone, readonly [unknown, unknown], IE, InDone, R>;
  <R, IE, OE, OutDone, InDone>(self: Channel<readonly [string, string], OE, OutDone, readonly [string, string], NdjsonError | IE, InDone, R>, options?: {
    readonly ignoreEmptyLines?: boolean;
  }): Channel<readonly [unknown, unknown], NdjsonError | OE, OutDone, readonly [unknown, unknown], IE, InDone, R>;
}

Constructors

decode

unstable

Since v4.0.0

Creates a channel that decodes UTF-8 byte chunks and parses them as NDJSON.

Details

Lines may span input chunks. \n, \r\n and a bare \r end a line, and a leading UTF-8 BOM is skipped. ignoreEmptyLines skips empty lines and lines containing only spaces or tabs before JSON parsing.

The first bad line fails the channel after the lines before it have been emitted. A line that is not valid UTF-8 fails with reason: "InvalidUtf8" and a line that is not valid JSON fails with reason: "InvalidJson".

maxLineBytes limits the UTF-8 bytes of a line, not counting its terminator. A longer line fails with reason: "LineTooLarge" as soon as the limit is passed, without reading the rest of the line. Lines are unbounded by default.

Signature

declare function decode<IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean;
  readonly maxLineBytes?: number;
}): Channel<readonly [unknown, unknown], NdjsonError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done>

decodeSchema

unstable

Since v4.0.0

Creates an NDJSON byte decoder channel for values of a schema.

Details

The channel decodes UTF-8 bytes, parses each NDJSON line, and then decodes each parsed value with the schema.

Signature

declare function decodeSchema<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean;
  readonly maxLineBytes?: number;
}) => Channel<readonly [S["Type"], S["Type"]], SchemaError | NdjsonError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done, S["DecodingServices"]>

decodeSchemaString

unstable

Since v4.0.0

Creates an NDJSON string decoder channel for values of a schema.

Details

The channel parses each line as JSON and then decodes each parsed value with the schema.

Signature

declare function decodeSchemaString<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean;
}) => Channel<readonly [S["Type"], S["Type"]], SchemaError | NdjsonError | IE, Done, readonly [string, string], IE, Done, S["DecodingServices"]>

decodeString

unstable

Since v4.0.0

Creates a channel that parses NDJSON string chunks into values.

When to use

Use when NDJSON input arrives as string chunks and each complete line should be parsed into a JSON value.

Details

Lines may span input chunks. \n, \r\n and a bare \r end a line. The first line that fails to parse fails the channel with reason: "InvalidJson", after the lines before it have been emitted.

Gotchas

Set ignoreEmptyLines to skip empty lines and lines containing only spaces or tabs before calling JSON.parse; otherwise these lines fail as invalid JSON.

Signature

declare function decodeString<IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean;
}): Channel<readonly [unknown, unknown], NdjsonError | IE, Done, readonly [string, string], IE, Done>

encode

unstable

Since v4.0.0

Creates a channel that encodes chunks of values as UTF-8 NDJSON bytes.

Signature

declare function encode<IE = never, Done = unknown>(): Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], NdjsonError | IE, Done, readonly [unknown, unknown], IE, Done>

encodeSchema

unstable

Since v4.0.0

Creates an NDJSON byte encoder channel for values of a schema.

Details

Values are first encoded with the schema and then written as UTF-8 newline-delimited JSON.

Signature

declare function encodeSchema<S extends Constraint>(schema: S): <IE = never, Done = unknown>() => Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], SchemaError | NdjsonError | IE, Done, readonly [S["Type"], S["Type"]], IE, Done, S["EncodingServices"]>

encodeSchemaString

unstable

Since v4.0.0

Creates an NDJSON string encoder channel for values of a schema.

Details

Values are first encoded with the schema and then written as newline-delimited JSON strings.

Signature

declare function encodeSchemaString<S extends Constraint>(schema: S): <IE = never, Done = unknown>() => Channel<readonly [string, string], SchemaError | NdjsonError | IE, Done, readonly [S["Type"], S["Type"]], IE, Done, S["EncodingServices"]>

encodeString

unstable

Since v4.0.0

Creates a channel that encodes chunks of values as NDJSON strings.

Details

Each input item is JSON.stringify-encoded, separated by newlines, and the output chunk ends with a trailing newline.

Signature

declare function encodeString<IE = never, Done = unknown>(): Channel<readonly [string, string], NdjsonError | IE, Done, readonly [unknown, unknown], IE, Done>

Errors

NdjsonError

unstable

Since v4.0.0

Error raised when NDJSON encoding or decoding fails.

Details

The kind field identifies whether the failure happened while packing or unpacking, and cause preserves the original error.

Decoders also set reason to say why a line was rejected, and unterminated when that line is the final line of the input and has no line terminator.

Signature

declare class NdjsonError extends YieldableError<this> & {
  readonly _tag: "NdjsonError";
} & Readonly<{
  readonly cause: unknown;
  readonly kind: "Pack" | "Unpack";
  readonly reason?: NdjsonErrorReason;
  readonly unterminated?: boolean;
}> {
  constructor(args: {
    readonly cause: unknown;
    readonly kind: "Pack" | "Unpack";
    readonly reason?: NdjsonErrorReason;
    readonly unterminated?: boolean;
  });
  readonly "~effect/encoding/Ndjson/NdjsonError": "~effect/encoding/Ndjson/NdjsonError";
  message: "Pack" | "Unpack";
}

NdjsonErrorReason

type unstable

Since v4.0.6

Why a decoder rejected a line.

Details

InvalidJson: the line is not valid JSON.
InvalidUtf8: the line is not valid UTF-8 (byte decoders only).
LineTooLarge: the line is longer than maxLineBytes (byte decoders only).

Signature

type NdjsonErrorReason = "InvalidJson" | "InvalidUtf8" | "LineTooLarge"

Guards

isNdjsonError

unstable

Since v4.0.0

Checks whether a value is an NdjsonError.

Signature

declare function isNdjsonError(u: unknown): u is NdjsonError