From 277e295a2fd0a6fdd659234717caa40754fee5b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ernesto=20Garc=C3=ADa?= Date: Thu, 27 Aug 2026 12:33:20 -0400 Subject: [PATCH] fix: respect output stream backpressure --- CODEBASE.md | 3 ++- README.md | 6 +++--- src/auth/account.test.ts | 34 ++++++++++++++++++------------ src/auth/account.ts | 16 ++++++++------ src/auth/status.test.ts | 9 ++++++-- src/auth/status.ts | 4 ++-- src/commands/update.test.ts | 10 ++++++++- src/commands/update.ts | 10 ++++----- src/ids.test.ts | 22 +++++++++---------- src/ids.ts | 15 ++++++------- src/index.ts | 2 +- src/json.test.ts | 35 +++++++++++++++++++++++++++++-- src/json.ts | 29 +++++++++++++++---------- src/options.ts | 8 +++---- src/stream.test.ts | 24 +++++++++++++++++++++ src/stream.ts | 42 +++++++++++++++++++++++++++++++++++++ 16 files changed, 200 insertions(+), 69 deletions(-) create mode 100644 src/stream.test.ts create mode 100644 src/stream.ts diff --git a/CODEBASE.md b/CODEBASE.md index c9eaf29..ff208c9 100644 --- a/CODEBASE.md +++ b/CODEBASE.md @@ -60,7 +60,8 @@ src/ ├─ index.ts # Root barrel (the `.` export) ├─ errors.ts # CliError + CliErrorCode aggregator + getErrorMessage ├─ config.ts # XDG config I/O; CoreConfig / UpdateChannel / ConfigErrorCode -├─ json.ts # formatJson / formatNdjson (throw on non-serializable) +├─ json.ts # formatJson / formatNdjson / outputNdjson +├─ stream.ts # Internal backpressure-aware buffered line writer ├─ options.ts # ViewOptions type + emitView (json/ndjson/human dispatch) ├─ empty.ts # printEmpty (machine-aware empty-state output) ├─ global-args.ts # parseGlobalArgs + spinner/accessible gate factories + stripUserFlag diff --git a/README.md b/README.md index b10350b..333b734 100644 --- a/README.md +++ b/README.md @@ -20,8 +20,8 @@ npm install @doist/cli-core | `empty` | `printEmpty` | Print an empty-state message gated on `--json` / `--ndjson` / `--ids-only` so machine consumers never see human strings on stdout. | | `errors` | `CliError` | Typed CLI error class with `code` and exit-code mapping. | | `global-args` | `parseGlobalArgs`, `stripUserFlag`, `createGlobalArgsStore`, `createAccessibleGate`, `createSpinnerGate`, `getProgressJsonlPath`, `isProgressJsonlEnabled` | Parse well-known global flags (`--json`, `--ndjson`, `--ids-only`, `--quiet`, `--verbose`, `--accessible`, `--no-spinner`, `--progress-jsonl`, `--user `) and derive predicates from them. `stripUserFlag` removes `--user` tokens from argv so the cleaned array can be forwarded to Commander when the flag has no root-program attachment. | -| `ids` | `formatIds`, `outputIds` | Format or emit one stable string or numeric ID per line. Empty results stay silent; optional pagination notices go to stderr. | -| `json` | `formatJson`, `formatNdjson` | Stable JSON / newline-delimited JSON formatting for stdout. | +| `ids` | `formatIds`, `outputIds` | Format or emit one stable string or numeric ID per line. Empty results stay silent; optional pagination notices go to stderr. Async output waits for stdout backpressure. | +| `json` | `formatJson`, `formatNdjson`, `outputNdjson` | Stable JSON formatting plus backpressure-aware newline-delimited JSON output. | | `markdown` (subpath) | `preloadMarkdown`, `renderMarkdown`, `TerminalRendererOptions` | Lazy-init terminal markdown renderer. **Requires** `marked` and `marked-terminal-renderer` as peer-deps — install only if your CLI uses this subpath. | | `options` | `OUTPUT_MODES`, `OutputMode`, `ViewOptions`, `ListViewOptions`, `resolveOutputMode` | Canonical output-mode contracts; `ListViewOptions` adds `idsOnly?`, and the resolver rejects conflicting machine-output flags. | | `spinner` | `createSpinner` | Loading spinner factory wrapping `yocto-spinner` with disable gates. | @@ -336,7 +336,7 @@ attachRefreshTokenViewCommand(auth, { `attachLogoutCommand` snapshots `store.active(ref)` when either `--user ` is supplied or one of the consumer hooks (`revokeToken` / `onCleared`) needs the prior account, calls `store.clear(ref)`, awaits `revokeToken({ token, account, ref, view, flags })` for best-effort server-side revocation, emits `✓ Logged out` (human) or `{ "ok": true }` (`--json`, silent under `--ndjson`), and finally fires `onCleared({ account, ref, view, flags })`. `ref` is the parsed `--user` argument (or `undefined`) so consumers can distinguish "nothing was stored" (`account: null`, `ref: undefined`) from "cleared an unreadable record by ref" (`account: null`, `ref: "me"`). `revokeToken` failures are always swallowed; the pre-flight snapshot's error contract is covered in the `--user ` section below. The exported `AttachLogoutRevokeContext` is the ctx type for typing standalone revoke implementations. -`attachStatusCommand` reads the active credential — preferring `store.activeBundle` when `fetchLive` is supplied, so the access token and the full bundle come from a single keyring read — optionally runs `fetchLive` (consumer translates 401 → `CliError('NO_TOKEN', …)`), then dispatches to `renderJson` (`--json` / `--ndjson` via `formatJson` / `formatNdjson`, defaults to the account itself, **only invoked in machine-output mode**) or `renderText` (human mode, string or array of lines). When the store is empty it throws `CliError('NOT_AUTHENTICATED', 'Not signed in.')` unless `onNotAuthenticated` is supplied. `fetchLive` receives `{ account, token, bundle?, view, flags }` — `bundle` carries the refresh-side metadata (expiry, refresh token) when the store implements `activeBundle`, so a consumer can render expiry without a second read. +`attachStatusCommand` reads the active credential — preferring `store.activeBundle` when `fetchLive` is supplied, so the access token and the full bundle come from a single keyring read — optionally runs `fetchLive` (consumer translates 401 → `CliError('NO_TOKEN', …)`), then dispatches to `renderJson` (`--json` / `--ndjson` via `formatJson` / `outputNdjson`, defaults to the account itself, **only invoked in machine-output mode**) or `renderText` (human mode, string or array of lines). When the store is empty it throws `CliError('NOT_AUTHENTICATED', 'Not signed in.')` unless `onNotAuthenticated` is supplied. `fetchLive` receives `{ account, token, bundle?, view, flags }` — `bundle` carries the refresh-side metadata (expiry, refresh token) when the store implements `activeBundle`, so a consumer can render expiry without a second read. Both attachers strip the standard `--json` / `--ndjson` / `--user` registrar flags from the parsed options and pass the remainder to their callbacks as `flags` — same escape hatch `attachLoginCommand` uses, so a consumer can chain e.g. `.option('--full')` and read it in `revokeToken` / `onCleared` / `renderText` / `fetchLive` / `renderJson` / `onNotAuthenticated`. diff --git a/src/auth/account.test.ts b/src/auth/account.test.ts index 01b3a6d..858259f 100644 --- a/src/auth/account.test.ts +++ b/src/auth/account.test.ts @@ -3,7 +3,11 @@ import { describe, expect, it, vi } from 'vitest' import { CliError } from '../errors.js' import { formatJson, formatNdjson } from '../json.js' -import { buildProgram, installCapturedConsole } from '../test-support/cli-harness.js' +import { + buildProgram, + installCapturedConsole, + installCapturedStream, +} from '../test-support/cli-harness.js' import { type TestAccount as Account, alanGrant, @@ -77,6 +81,7 @@ function buildRemove( describe('attachAccountListCommand', () => { const logSpy = installCapturedConsole() + const stdoutSpy = installCapturedStream() it('renders default human lines with a (default) marker only on the default entry', async () => { const { program } = buildList() @@ -168,11 +173,12 @@ describe('attachAccountListCommand', () => { await program.parseAsync(['node', 'cli', 'account', 'list', '--ndjson']) - const emitted = logSpy().mock.calls.map((call: unknown[]) => call[0]) - expect(emitted).toEqual([ - formatNdjson([{ account: alanGrant, isDefault: true }]), - formatNdjson([{ account: ellieSattler, isDefault: false }]), - ]) + expect(stdoutSpy()).toHaveBeenCalledWith( + `${formatNdjson([ + { account: alanGrant, isDefault: true }, + { account: ellieSattler, isDefault: false }, + ])}\n`, + ) }) it('shapes each --ndjson line via renderJson, matching the --json accounts entries', async () => { @@ -196,11 +202,12 @@ describe('attachAccountListCommand', () => { isDefault: false, flags: {}, }) - const emitted = logSpy().mock.calls.map((call: unknown[]) => call[0]) - expect(emitted).toEqual([ - formatNdjson([{ name: 'Alan Grant', isDefault: true }]), - formatNdjson([{ name: 'Ellie Sattler', isDefault: false }]), - ]) + expect(stdoutSpy()).toHaveBeenCalledWith( + `${formatNdjson([ + { name: 'Alan Grant', isDefault: true }, + { name: 'Ellie Sattler', isDefault: false }, + ])}\n`, + ) }) it('prefers --json over --ndjson when both flags are passed', async () => { @@ -426,6 +433,7 @@ describe('attachAccountUseCommand', () => { describe('attachAccountCurrentCommand', () => { const logSpy = installCapturedConsole() + const stdoutSpy = installCapturedStream() it('renders the default human line with a (default) marker for the active account', async () => { const { program } = buildCurrent() @@ -491,8 +499,8 @@ describe('attachAccountCurrentCommand', () => { await program.parseAsync(['node', 'cli', 'account', 'current', '--ndjson']) - expect(logSpy()).toHaveBeenCalledWith( - formatNdjson([{ account: alanGrant, isDefault: true }]), + expect(stdoutSpy()).toHaveBeenCalledWith( + `${formatNdjson([{ account: alanGrant, isDefault: true }])}\n`, ) }) diff --git a/src/auth/account.ts b/src/auth/account.ts index 911b55d..0094a98 100644 --- a/src/auth/account.ts +++ b/src/auth/account.ts @@ -1,6 +1,6 @@ import type { Command } from 'commander' import { CliError } from '../errors.js' -import { formatJson, formatNdjson } from '../json.js' +import { formatJson, outputNdjson } from '../json.js' import { type ViewOptions, emitView } from '../options.js' import type { AccountRef, @@ -172,7 +172,7 @@ export function attachAccountListCommand { + await emitView(view, payload, () => { const ctx: AttachAccountListContext = { accounts, default: defaultRef, @@ -224,7 +224,7 @@ export function attachAccountUseCommand entry.isDefault)?.account.id ?? ref) : ref - emitView(view, { ok: true, default: resolvedDefault }, () => [ + await emitView(view, { ok: true, default: resolvedDefault }, () => [ `✓ Default account set to ${ref}`, ]) } @@ -307,7 +307,11 @@ export function attachAccountCurrentCommand { + await emitView(view, { ok: true, removed: cleared.account.id }, () => { const name = cleared.account.label ?? cleared.account.id const removedLine = `✓ Removed ${name}${ctx.wasDefault ? ' (default)' : ''}` const text = options.renderText ? options.renderText(ctx) : removedLine diff --git a/src/auth/status.test.ts b/src/auth/status.test.ts index 5424eda..b75ff2c 100644 --- a/src/auth/status.test.ts +++ b/src/auth/status.test.ts @@ -3,7 +3,11 @@ import { describe, expect, it, vi } from 'vitest' import { CliError } from '../errors.js' import { formatJson, formatNdjson } from '../json.js' -import { buildProgram, installCapturedConsole } from '../test-support/cli-harness.js' +import { + buildProgram, + installCapturedConsole, + installCapturedStream, +} from '../test-support/cli-harness.js' import { type TestAccount as Account, type TokenStoreHarness, @@ -44,6 +48,7 @@ function build( describe('attachStatusCommand', () => { const logSpy = installCapturedConsole() + const stdoutSpy = installCapturedStream() it('emits renderText output in plain mode', async () => { const { program, renderText } = build() @@ -97,7 +102,7 @@ describe('attachStatusCommand', () => { await program.parseAsync(['node', 'cli', 'auth', 'status', '--ndjson']) - expect(logSpy()).toHaveBeenCalledWith(formatNdjson([account])) + expect(stdoutSpy()).toHaveBeenCalledWith(`${formatNdjson([account])}\n`) }) it('does not invoke renderJson in human mode', async () => { diff --git a/src/auth/status.ts b/src/auth/status.ts index c86d9d0..e2d2eba 100644 --- a/src/auth/status.ts +++ b/src/auth/status.ts @@ -1,6 +1,6 @@ import type { Command } from 'commander' import { CliError } from '../errors.js' -import { formatJson, formatNdjson } from '../json.js' +import { formatJson, outputNdjson } from '../json.js' import type { ViewOptions } from '../options.js' import type { AccountRef, @@ -149,7 +149,7 @@ export function attachStatusCommand( } if (view.ndjson) { const payload = options.renderJson ? options.renderJson({ account, flags }) : account - console.log(formatNdjson([payload])) + await outputNdjson([payload]) return } const text = options.renderText({ account, view, flags }) diff --git a/src/commands/update.test.ts b/src/commands/update.test.ts index c0dd529..27b6978 100644 --- a/src/commands/update.test.ts +++ b/src/commands/update.test.ts @@ -79,10 +79,14 @@ function mockSpawnError(error: Error) { } let consoleSpy: ReturnType +let stdoutSpy: ReturnType beforeEach(() => { chalk.level = 0 consoleSpy = vi.spyOn(console, 'log').mockImplementation(() => {}) + stdoutSpy = vi + .spyOn(process.stdout, 'write') + .mockImplementation((() => true) as typeof process.stdout.write) mockReadConfigOrThrow.mockReset().mockResolvedValue({}) mockUpdateConfigOrThrow.mockReset().mockResolvedValue(undefined) mockSpawn.mockClear() @@ -187,7 +191,11 @@ describe('update --check', () => { ])('emits machine envelope under %s', async (flag, parse) => { mockFetchOk('99.99.99') await createProgram().parseAsync(['node', 'td', 'update', '--check', flag]) - expect(parse(consoleSpy.mock.calls[0][0] as string)).toEqual({ + const output = + flag === '--ndjson' + ? stdoutSpy.mock.calls.map((call: unknown[]) => String(call[0])).join('') + : String(consoleSpy.mock.calls[0][0]) + expect(parse(output)).toEqual({ currentVersion: '1.0.0', latestVersion: '99.99.99', channel: 'stable', diff --git a/src/commands/update.ts b/src/commands/update.ts index 8e2bb1f..8444d01 100644 --- a/src/commands/update.ts +++ b/src/commands/update.ts @@ -274,7 +274,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions): const channel = await getConfiguredUpdateChannel(options.configPath) if (cmd.channel) { - emitView(view, { channel }, () => [`Update channel: ${formatChannel(channel)}`]) + await emitView(view, { channel }, () => [`Update channel: ${formatChannel(channel)}`]) return } @@ -314,7 +314,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions): const updateAvailable = !upToDate && isNewer(currentVersion, latestVersion) if (cmd.check) { - emitView(view, { currentVersion, latestVersion, channel, updateAvailable }, () => { + await emitView(view, { currentVersion, latestVersion, channel, updateAvailable }, () => { const channelLine = ` Channel: ${formatChannel(channel)}` const headline = upToDate ? `${chalk.green('✓')} Already up to date (v${currentVersion})` @@ -327,7 +327,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions): } if (upToDate) { - emitView(view, { currentVersion, latestVersion, channel, installed: false }, () => [ + await emitView(view, { currentVersion, latestVersion, channel, installed: false }, () => [ `${chalk.green('✓')} Already up to date${label} (v${currentVersion})`, ]) return @@ -405,7 +405,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions): via: brew ? ('brew' as const) : pm, ...(brew && installedVersion ? { installedVersion } : {}), } - emitView(view, summary, () => { + await emitView(view, summary, () => { const lines = [ `${chalk.green('✓')} ${brew ? 'brew upgrade complete' : `Updated to v${latestVersion}`}${label}`, ] @@ -442,7 +442,7 @@ async function runSwitch( await updateConfigOrThrow(options.configPath, { update_channel: channel }) - emitView(view, { channel }, () => { + await emitView(view, { channel }, () => { if (channel === 'pre-release') { return [ `${chalk.green('✓')} Update channel set to ${formatChannel(channel)}`, diff --git a/src/ids.test.ts b/src/ids.test.ts index a642962..5eb2059 100644 --- a/src/ids.test.ts +++ b/src/ids.test.ts @@ -16,27 +16,27 @@ describe('outputIds', () => { vi.restoreAllMocks() }) - it('writes IDs to stdout in one block', () => { - const log = vi.spyOn(console, 'log').mockImplementation(() => {}) + it('writes IDs to stdout in one block', async () => { + const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true) - outputIds([{ id: 'a' }, { id: 'b' }], (item) => item.id) + await outputIds([{ id: 'a' }, { id: 'b' }], (item) => item.id) - expect(log).toHaveBeenCalledOnce() - expect(log).toHaveBeenCalledWith('a\nb') + expect(write).toHaveBeenCalledOnce() + expect(write).toHaveBeenCalledWith('a\nb\n') }) - it('writes nothing to stdout for no results', () => { - const log = vi.spyOn(console, 'log').mockImplementation(() => {}) + it('writes nothing to stdout for no results', async () => { + const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true) - outputIds([], (item: { id: string }) => item.id) + await outputIds([], (item: { id: string }) => item.id) - expect(log).not.toHaveBeenCalled() + expect(write).not.toHaveBeenCalled() }) - it('writes pagination notices to stderr', () => { + it('writes pagination notices to stderr', async () => { const error = vi.spyOn(console, 'error').mockImplementation(() => {}) - outputIds([], (item: { id: string }) => item.id, 'More results exist.') + await outputIds([], (item: { id: string }) => item.id, 'More results exist.') expect(error).toHaveBeenCalledWith('More results exist.') }) diff --git a/src/ids.ts b/src/ids.ts index 4b24e65..fbddc8a 100644 --- a/src/ids.ts +++ b/src/ids.ts @@ -1,18 +1,19 @@ +import { writeLines } from './stream.js' + /** Format stable IDs as one value per line, with no trailing newline. */ export function formatIds(items: readonly T[], getId: (item: T) => string | number): string { return items.map((item) => String(getId(item))).join('\n') } /** - * Write stable IDs to stdout and an optional pagination notice to stderr. - * Empty results write nothing to stdout. + * Write stable IDs to stdout in bounded chunks, waiting when the stream applies + * backpressure. Empty results write nothing; pagination notices go to stderr. */ -export function outputIds( - items: readonly T[], +export async function outputIds( + items: Iterable, getId: (item: T) => string | number, paginationNotice = '', -): void { - const output = formatIds(items, getId) - if (output) console.log(output) +): Promise { + await writeLines(items, (item) => String(getId(item))) if (paginationNotice) console.error(paginationNotice) } diff --git a/src/index.ts b/src/index.ts index 4b16eb1..3a3ba56 100644 --- a/src/index.ts +++ b/src/index.ts @@ -33,7 +33,7 @@ export type { GlobalArgsStore, SpinnerGateOptions, } from './global-args.js' -export { formatJson, formatNdjson } from './json.js' +export { formatJson, formatNdjson, outputNdjson } from './json.js' export { formatIds, outputIds } from './ids.js' export { emitView, OUTPUT_MODES, resolveOutputMode } from './options.js' export type { ListViewOptions, OutputMode, ViewOptions } from './options.js' diff --git a/src/json.test.ts b/src/json.test.ts index b02f4d3..e0ec4d0 100644 --- a/src/json.test.ts +++ b/src/json.test.ts @@ -1,6 +1,6 @@ -import { describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' -import { formatJson, formatNdjson } from './json.js' +import { formatJson, formatNdjson, outputNdjson } from './json.js' describe('formatJson', () => { it('pretty-prints objects with 2-space indentation', () => { @@ -69,3 +69,34 @@ describe('formatNdjson', () => { expect(() => formatNdjson([Symbol('x')])).toThrow(/index 0/) }) }) + +describe('outputNdjson', () => { + afterEach(() => { + vi.restoreAllMocks() + }) + + it('writes NDJSON to stdout in one block', async () => { + const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true) + + await outputNdjson([{ a: 1 }, { a: 2 }]) + + expect(write).toHaveBeenCalledOnce() + expect(write).toHaveBeenCalledWith('{"a":1}\n{"a":2}\n') + }) + + it('writes nothing for an empty iterable', async () => { + const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true) + + await outputNdjson([]) + + expect(write).not.toHaveBeenCalled() + }) + + it('reports the index of a non-serializable item', async () => { + vi.spyOn(process.stdout, 'write').mockReturnValue(true) + + await expect(outputNdjson([1, undefined, 2])).rejects.toThrow( + /index 1.*not JSON-serializable/, + ) + }) +}) diff --git a/src/json.ts b/src/json.ts index d74ea88..4b579b6 100644 --- a/src/json.ts +++ b/src/json.ts @@ -1,3 +1,15 @@ +import { writeLines } from './stream.js' + +function stringifyNdjsonItem(item: unknown, index: number, source: string): string { + const line = JSON.stringify(item) + if (line === undefined) { + throw new TypeError( + `${source}: item at index ${index} is not JSON-serializable (got undefined, function, or symbol)`, + ) + } + return line +} + /** * Pretty-print a value as JSON with 2-space indentation. Matches the canonical * `--json` output style used across the Doist CLIs. @@ -25,15 +37,10 @@ export function formatJson(value: unknown): string { * silently emitting blank lines that would corrupt the output stream. */ export function formatNdjson(items: readonly unknown[]): string { - return items - .map((item, i) => { - const line = JSON.stringify(item) - if (line === undefined) { - throw new TypeError( - `formatNdjson: item at index ${i} is not JSON-serializable (got undefined, function, or symbol)`, - ) - } - return line - }) - .join('\n') + return items.map((item, index) => stringifyNdjsonItem(item, index, 'formatNdjson')).join('\n') +} + +/** Write newline-delimited JSON to stdout while respecting stream backpressure. */ +export async function outputNdjson(items: Iterable): Promise { + await writeLines(items, (item, index) => stringifyNdjsonItem(item, index, 'outputNdjson')) } diff --git a/src/options.ts b/src/options.ts index f57d63f..1ed5694 100644 --- a/src/options.ts +++ b/src/options.ts @@ -1,5 +1,5 @@ import { CliError } from './errors.js' -import { formatJson, formatNdjson } from './json.js' +import { formatJson, outputNdjson } from './json.js' /** Canonical output modes shared by Doist CLIs. */ export const OUTPUT_MODES = ['human', 'json', 'ndjson', 'ids-only'] as const @@ -53,17 +53,17 @@ export function resolveOutputMode(options: ListViewOptions): OutputMode { * human-mode strings (chalk colouring, conditional formatting) are never * built when machine output is requested. */ -export function emitView( +export async function emitView( view: ViewOptions, payload: Record, humanLines: () => ReadonlyArray, -): void { +): Promise { if (view.json) { console.log(formatJson(payload)) return } if (view.ndjson) { - console.log(formatNdjson([payload])) + await outputNdjson([payload]) return } for (const line of humanLines()) console.log(line) diff --git a/src/stream.test.ts b/src/stream.test.ts new file mode 100644 index 0000000..fe56e78 --- /dev/null +++ b/src/stream.test.ts @@ -0,0 +1,24 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { writeLines } from './stream.js' + +describe('writeLines', () => { + afterEach(() => { + vi.restoreAllMocks() + }) + + it('waits for stdout to drain before writing the next chunk', async () => { + const write = vi + .spyOn(process.stdout, 'write') + .mockReturnValueOnce(false) + .mockReturnValue(true) + const firstLine = 'a'.repeat(process.stdout.writableHighWaterMark) + + const output = writeLines([firstLine, 'b'], (line) => line) + + expect(write).toHaveBeenCalledOnce() + process.stdout.emit('drain') + await output + expect(write).toHaveBeenCalledTimes(2) + expect(write).toHaveBeenLastCalledWith('b\n') + }) +}) diff --git a/src/stream.ts b/src/stream.ts new file mode 100644 index 0000000..8a9aa84 --- /dev/null +++ b/src/stream.ts @@ -0,0 +1,42 @@ +import { once } from 'node:events' + +const FALLBACK_CHUNK_SIZE = 16 * 1024 + +export async function writeLines( + items: Iterable, + formatLine: (item: T, index: number) => string, +): Promise { + const chunkSize = process.stdout.writableHighWaterMark || FALLBACK_CHUNK_SIZE + let chunk = '' + let chunkBytes = 0 + let index = 0 + + const flush = async (): Promise => { + if (!chunk) return + const output = chunk + chunk = '' + chunkBytes = 0 + if (!process.stdout.write(output)) { + await once(process.stdout, 'drain') + } + } + + for (const item of items) { + const line = `${formatLine(item, index)}\n` + const lineBytes = Buffer.byteLength(line) + index += 1 + + if (chunk && chunkBytes + lineBytes > chunkSize) { + await flush() + } + + chunk += line + chunkBytes += lineBytes + + if (chunkBytes >= chunkSize) { + await flush() + } + } + + await flush() +}