node/test/parallel/test-fastutf8stream-reopen.js

'use strict';

const common = require('../common');
const tmpdir = require('../common/tmpdir');
const assert = require('node:assert');
const {
  open,
  openSync,
  readFile,
  renameSync,
  write,
} = require('node:fs');
const { Utf8Stream } = require('node:fs');
const { join } = require('node:path');
const { isMainThread } = require('node:worker_threads');

tmpdir.refresh();
let fileCounter = 0;

if (isMainThread) {
  process.umask(0o000);
}

function getTempFile() {
  return join(tmpdir.path, `fastutf8stream-${process.pid}-${Date.now()}-${fileCounter++}.log`);
}

runTests(false);
runTests(true);

// A write started by a 'ready' listener after reopen() must complete before
// 'drain' is emitted. Deferring the reopened file's write by one setImmediate
// makes the write land after the nextTick on which reopen() used to emit a
// premature 'drain'. Async mode only: sync mode writes before 'ready'.
{
  const dest = getTempFile();
  const after = dest + '-new';
  const stream = new Utf8Stream({
    dest,
    minLength: 0,
    sync: false,
    fs: {
      write(fd, buf, enc, cb) {
        if (stream.file === after) {
          setImmediate(() => write(fd, buf, enc, cb));
          return;
        }
        return write(fd, buf, enc, cb);
      },
    },
  });

  assert.ok(stream.write('hello world\n'));

  stream.once('drain', common.mustCall(() => {
    stream.reopen(after);

    stream.once('ready', common.mustCall(() => {
      assert.ok(stream.write('after reopen\n'));

      stream.once('drain', common.mustCall(() => {
        assert.strictEqual(stream.writing, false);
        readFile(after, 'utf8', common.mustSucceed((data) => {
          assert.strictEqual(data, 'after reopen\n');
          stream.end();
        }));
      }));
    }));
  }));
}

function runTests(sync) {

  {
    const dest = getTempFile();
    const stream = new Utf8Stream({ dest, sync });

    assert.ok(stream.write('hello world\n'));
    assert.ok(stream.write('something else\n'));

    const after = dest + '-moved';

    stream.once('drain', common.mustCall(() => {
      renameSync(dest, after);
      stream.reopen();

      stream.once('ready', common.mustCall(() => {
        // 'drain' can come from reopen() before this write completes. 'write'
        // is emitted synchronously in sync mode, so attach before writing.
        stream.once('write', common.mustCall(() => {
          readFile(after, 'utf8', common.mustSucceed((data) => {
            assert.strictEqual(data, 'hello world\nsomething else\n');
            readFile(dest, 'utf8', common.mustSucceed((data) => {
              assert.strictEqual(data, 'after reopen\n');
              stream.end();
            }));
          }));
        }));

        assert.ok(stream.write('after reopen\n'));
      }));
    }));
  }

  {
    const dest = getTempFile();
    const stream = new Utf8Stream({ dest, sync });

    assert.ok(stream.write('hello world\n'));
    assert.ok(stream.write('something else\n'));

    stream.reopen();
    stream.end();

    stream.on('close', common.mustCall());
  }

  {
    const dest = getTempFile();
    const stream = new Utf8Stream({ dest, minLength: 0, sync });

    assert.ok(stream.write('hello world\n'));
    assert.ok(stream.write('something else\n'));

    const after = dest + '-new';

    stream.once('drain', common.mustCall(() => {
      stream.reopen(after);
      assert.strictEqual(stream.file, after);

      stream.once('ready', common.mustCall(() => {
        // As above: 'drain' can come from reopen() before the write completes.
        stream.once('write', common.mustCall(() => {
          readFile(dest, 'utf8', common.mustSucceed((data) => {
            assert.strictEqual(data, 'hello world\nsomething else\n');
            readFile(after, 'utf8', common.mustSucceed((data) => {
              assert.strictEqual(data, 'after reopen\n');
              stream.end();
            }));
          }));
        }));

        assert.ok(stream.write('after reopen\n'));
      }));
    }));
  }

  {
    let throwOnNextOpen = false;
    const err = new Error('open error');
    const fsOverride = {};
    if (sync) {
      fsOverride.openSync = function(...args) {
        if (throwOnNextOpen) {
          throwOnNextOpen = false;
          throw err;
        }
        return openSync(...args);
      };
    } else {
      fsOverride.open = function(file, flags, mode, cb) {
        if (throwOnNextOpen) {
          throwOnNextOpen = false;
          process.nextTick(() => cb(err));
          return;
        }
        return open(file, flags, mode, cb);
      };
    }

    const dest = getTempFile();
    const stream = new Utf8Stream({
      dest,
      sync,
      fs: fsOverride,
    });

    assert.ok(stream.write('hello world\n'));
    assert.ok(stream.write('something else\n'));

    const after = dest + '-moved';
    stream.on('error', common.mustCall());

    stream.once('drain', common.mustCall(() => {
      renameSync(dest, after);
      throwOnNextOpen = true;
      if (sync) {
        assert.throws(() => stream.reopen(), err);
      } else {
        stream.reopen();
      }

      setTimeout(common.mustCall(() => {
        assert.ok(stream.write('after reopen\n'));

        stream.end();
        stream.on('finish', common.mustCall(() => {
          readFile(after, 'utf8', common.mustSucceed((data) => {
            assert.strictEqual(data, 'hello world\nsomething else\nafter reopen\n');
          }));
        }));
      }), 10);
    }));
  }

  {
    const dest = getTempFile();
    const stream = new Utf8Stream({ dest, sync });

    assert.ok(stream.write('hello world\n'));
    assert.ok(stream.write('something else\n'));

    const after = dest + '-moved';
    stream.once('drain', common.mustCall(() => {
      renameSync(dest, after);
      stream.reopen();

      stream.once('drain', common.mustCall(() => {
        assert.ok(stream.write('after reopen\n'));

        stream.once('drain', common.mustCall(() => {
          readFile(after, 'utf8', common.mustSucceed((data) => {
            assert.strictEqual(data, 'hello world\nsomething else\n');
            readFile(dest, 'utf8', common.mustSucceed((data) => {
              assert.strictEqual(data, 'after reopen\n');
              stream.end();
            }));
          }));
        }));
      }));
    }));
  }
}