node.js line number stream. https://2ality.com/2019/11/nodejs-streams-async-iteration.html, https://github.com/deanhume/streams

README.md

node.js line number stream.

https://2ality.com/2019/11/nodejs-streams-async-iteration.html, https://github.com/deanhume/streams

createSplitStream.js

function createSplitStream(splitter) {
  let buffer = '';

  return new stream.Transform({
      transform(chunk, encoding, next) {
        buffer += chunk;
        const parts = buffer.split(splitter);
        parts.slice(0, -1).forEach(part => this.push(part));
        buffer = parts[parts.length - 1];
        next();
      },
      flush(next) {
        if (buffer) next(null, buffer);
      }
  });
}

enumerate.js

const os = require('os');
const fs = require('fs');
const stream = require('stream');

const [ , , inputFile, outputFile ] = process.argv;

class SplitStream extends stream.Transform {
  constructor(options) {
    super(options);
    this.tail = '';
  }

  _transform(chunk, encoding, next) {
    const pieces = (this.tail + chunk).split(os.EOL);
    if (pieces.length === 1) {
      this.tail += chunk;
      return next();
    }
    this.tail = pieces.pop();
    pieces.forEach(this._pushPiece, this);
    next();
  }

  _flush(next) {
    this.tail
      .split(os.EOL)
      .forEach(this._pushPiece, this);
    next();
  }

  _pushPiece(piece) {
    piece = piece.toString();
    // если передать пустую строку в push, то она игнорируется, поэтому добавляем пробел
    piece = piece.length === 0 ? piece + ' ' : piece;
    this.push(piece);
  }
};

class LineNumberStream extends stream.Transform {
  constructor(options) {
    super(options);
    this.count = 0;
  }

  _transform(chunk, encoding, next) {
    const line = `${++this.count}. ${chunk}`;
    next(null, line);
  }
}

class EndLineStream extends stream.Transform {
  constructor(options) {
    super(options);
  }

  _transform(chunk, encoding, next) {
    next(null, `${chunk}${os.EOL}`);
  }
}

class TrimLineStream extends stream.Transform {
  constructor(options) {
    super(options);
  }

  _transform(chunk, encoding, next) {
    next(null, chunk.toString().trim());
  }
}

stream.pipeline(
  fs.createReadStream(inputFile),
  new SplitStream(),
  new LineNumberStream(),
  new TrimLineStream(),
  new EndLineStream(),
  outputFile ? fs.createWriteStream(outputFile) : process.stdout,
  console.log
);

generators.js

'use strict'

import os from 'node:os'
import fs from 'node:fs'
import { pipeline } from 'node:stream/promises'

pipeline(
  fs.createReadStream('index.html'),
  async function* (source) {
    source.setEncoding('utf-8')
    yield* source
  },
  async function* (source) {
    let buffer = ''
    let splitter = os.EOL

    try {
      for await (const chunk of source) {
        buffer += chunk
        const parts = buffer.split(splitter)
        buffer = parts.pop()
        for (const part of parts) {
          yield part
        }
      }
    } finally {
      if (buffer) {
        return buffer
      }
    }
  },
  async function* (source) {
    let lineCounter = 0
    while (true) {
      const { value, done } = await source.next()
      lineCounter += 1
      const tail = done ? '' : os.EOL
      const linerNumber = lineCounter.toString().padEnd(4, ' ')
      const lineContent = value ?? ''
      yield linerNumber + ' ' + lineContent + tail

      if (done) {
        break
      }
    }
  },
  fs.createWriteStream('index.lines.html')
)
  .then(() => console.log('end'))
  .catch((error) => console.error('error: ', error.message))