Function attachNdjsonFramer

src/utils/ndjsonFramer.ts:27–100

(
  socket: Socket,
  onMessage: (msg: T) => void,
  parse: (text: string) => T = text => JSON.parse(text) as T,
  options: NdjsonFramerOptions = {},
)

 *                from slowOperations.
 */
export function attachNdjsonFramer<T = unknown>(
  socket: Socket,
  onMessage: (msg: T) => void,
  parse: (text: string) => T = text => JSON.parse(text) as T,
  options: NdjsonFramerOptions = {},
): void {
  let buffer = ''
  let bufferBytes = 0
  const maxFrameBytes = options.maxFrameBytes ?? Number.POSITIVE_INFINITY

  const rejectOversizedFrame = (bytes: number): void => {
    const error = new Error(
      `NDJSON frame exceeded ${maxFrameBytes} bytes (${bytes})`,
    )
    options.onFrameError?.(error)
    if (options.destroyOnFrameError ?? true) {
      socket.destroy(error)
    }
  }

  const rejectInvalidFrame = (error: unknown): void => {
    const frameError =
      error instanceof Error ? error : new Error('Invalid NDJSON frame')
    options.onInvalidFrame?.(frameError)
    if (options.destroyOnInvalidFrame ?? false) {
      socket.destroy(frameError)
    }
  }

  const emitLine = (line: string): void => {
    if (!line.trim()) return
    try {
      onMessage(parse(line))
    } catch (error) {
      rejectInvalidFrame(error)
    }
  }

  socket.on('data', (chunk: Buffer) => {
    let start = 0
    for (let index = 0; index < chunk.length; index++) {
      if (chunk[index] !== 0x0a) continue

      const segmentBytes = index - start
      if (
        Number.isFinite(maxFrameBytes) &&
        bufferBytes + segmentBytes > maxFrameBytes
      ) {
        rejectOversizedFrame(bufferBytes + segmentBytes)
        return
      }

      buffer += chunk.subarray(start, index).toString('utf8')
      emitLine(buffer)
      buffer = ''
      bufferBytes = 0
      start = index + 1
    }