[low-level-async-buffered-read.js] Low-level async buffered read from file with Node.js #nodejs #async #file

low-level-async-buffered-read.js

const fs = require("fs");
const EventEmitter = require("events").EventEmitter;

const read = (fileHandle, bufferSize, position, emitter) => {
  const buffer = Buffer.alloc(bufferSize);

  fs.read(
    fileHandle,
    buffer,
    0,
    buffer.byteLength,
    position,
    (err, bytesRead, buffer) => {
      if (err) {
        emitter.emit("error", err);
        return;
      }

      if (bytesRead === 0) {
        fs.close(fileHandle, err => {
          if (err) {
            emitter.emit("error", err);
            return;
          }

          emitter.emit("finish");
        });
        return;
      }

      emitter.emit("buffer", { buffer, bytesRead });

      position += bytesRead;

      // prevents stack from blowing up
      setImmediate(() => {
        read(fileHandle, bufferSize, position, emitter);
      });
    }
  );
};

const readFile = (path, bufferSize) => {
  const emitter = new EventEmitter();

  fs.open(path, "r", (err, fileHandle) => {
    if (err) {
      emitter.emit("error", err);
      return;
    }
    emitter.emit("open", fileHandle);

    let position = 0;
    read(fileHandle, bufferSize, 0, emitter);
  });

  return emitter;
};

// Usage:
const startTime = Date.now();
let readLength = 0;

readFile("./package.json", 1 * 1024 ** 2)
  .on("open", () => console.log("Opened!"))
  .on("buffer", ({ buffer, bytesRead }) => {
    readLength += bytesRead;
    process.stdout.clearLine();
    process.stdout.write(
      `\r${readLength / 1024 ** 2} MB - rss: ${process.memoryUsage().rss /
        1024 ** 2} MB`
    );
  })
  .on("error", console.error)
  .on("finish", () => {
    const endTime = Date.now();
    // avoid division by zero
    const deltaTime = (endTime - startTime || 1) / 1000;

    console.log("\nDone!");
    console.log(
      `Read speed: ${(readLength / deltaTime / 1024 ** 2).toFixed(2)} MB/s`
    );
  });