eltro/lib/watch/index.mjs

import fs from 'fs'
import path from 'path'
import util from 'util'
import events from 'events'
import hasNativeRecursive from './has-native-recursive.mjs'
import * as is from './is.mjs'

const EVENT_UPDATE = 'update';
const EVENT_REMOVE = 'remove';

const SKIP_FLAG = Symbol('skip');

function hasDup(arr) {
  return arr.some(function(v, i, self) {
    return self.indexOf(v) !== i;
  });
}

function unique(arr) {
  return arr.filter(function(v, i, self) {
    return self.indexOf(v) === i;
  });
}

// One level flat
function flat1(arr) {
  return arr.reduce(function(acc, v) {
    return acc.concat(v);
  }, []);
}

function assertEncoding(encoding) {
  if (encoding && encoding !== 'buffer' && !Buffer.isEncoding(encoding)) {
    throw new Error('Unknown encoding: ' + encoding);
  }
}

function guard(fn) {
  if (is.func(fn)) {
    return function(arg, action) {
      if (fn(arg, false)) action();
    }
  }
  if (is.regExp(fn)) {
    return function(arg, action) {
      if (fn.test(arg)) action();
    }
  }
  return function(arg, action) {
    action();
  }
}

function composeMessage(names) {
  return names.map(function(n) {
    return is.exists(n)
      ? [EVENT_UPDATE, n]
      : [EVENT_REMOVE, n];
  });
}

function getMessages(cache) {
  var filtered = unique(cache);

  // Saving file from an editor? If so, assuming the
  // non-existed files in the cache are temporary files
  // generated by an editor and thus be filtered.
  var reg = /~$|^\.#|^##$/g;
  var hasSpecialChar = cache.some(function(c) {
    return reg.test(c);
  });

  if (hasSpecialChar) {
    var dup = hasDup(cache.map(function(c) {
      return c.replace(reg, '');
    }));
    if (dup) {
      filtered = filtered.filter(function(m) {
        return is.exists(m);
      });
    }
  }

  return composeMessage(filtered);
}

function debounce(info, fn) {
  var timer, cache = [];
  var encoding = info.options.encoding;
  var delay = info.options.delay;
  if (!is.number(delay)) {
    delay = 200;
  }
  function handle() {
    getMessages(cache).forEach(function(msg) {
      msg[1] = Buffer.from(msg[1]);
      if (encoding !== 'buffer') {
        msg[1] = msg[1].toString(encoding);
      }
      fn.apply(null, msg);
    });
    timer = null;
    cache = [];
  }
  return function(rawEvt, name) {
    cache.push(name);
    if (!timer) {
      timer = setTimeout(handle, delay);
    }
  }
}

function createDupsFilter() {
  var memo = {};
  return function(fn) {
    return function(evt, name) {
      memo[evt + name] = [evt, name];
      setTimeout(function() {
        Object.keys(memo).forEach(function(n) {
          fn.apply(null, memo[n]);
        });
        memo = {};
      });
    }
  }
}

function tryWatch(watcher, dir, opts) {
  try {
    return fs.watch(dir, opts);
  } catch (e) {
    process.nextTick(function() {
      watcher.emit('error', e);
    });
  }
}

function getSubDirectories(dir, fn, done = function() {}) {
  if (is.directory(dir)) {
    fs.readdir(dir, function(err, all) {
      if (err) {
        // don't throw permission errors.
        if (/^(EPERM|EACCES)$/.test(err.code)) {
          console.warn('Warning: Cannot access %s.', dir);
        } else {
          throw err;
        }
      }
      else {
        all.forEach(function(f) {
          var sdir = path.join(dir, f);
          if (is.directory(sdir)) fn(sdir);
        });
        done();
      }
    });
  } else {
    done();
  }
}

function semaphore(final) {
  var counter = 0;
  return function start() {
    counter++;
    return function stop() {
      counter--;
      if (counter === 0) final();
    };
  };
}

function nullCounter() {
  return function nullStop() {};
}

function shouldNotSkip(filePath, filter) {
  // watch it only if the filter is not function
  // or not being skipped explicitly.
  return !is.func(filter) || filter(filePath, SKIP_FLAG) !== SKIP_FLAG;
}

var deprecationWarning = util.deprecate(
  function() {},
  '(node-watch) First param in callback function\
  is replaced with event name since 0.5.0, use\
  `(evt, filename) => {}` if you want to get the filename'
);

function Watcher() {
  events.EventEmitter.call(this);
  this.watchers = {};
  this._isReady = false;
  this._isClosed = false;
}

util.inherits(Watcher, events.EventEmitter);

Watcher.prototype.expose = function() {
  var expose = {};
  var self = this;
  var methods = [
    'on', 'emit', 'once',
    'close', 'isClosed',
    'listeners', 'setMaxListeners', 'getMaxListeners',
    'getWatchedPaths'
  ];
  methods.forEach(function(name) {
    expose[name] = function() {
      return self[name].apply(self, arguments);
    }
  });
  expose.options = self.options
  return expose;
}

Watcher.prototype.isClosed = function() {
  return this._isClosed;
}

Watcher.prototype.close = function(fullPath) {
  var self = this;
  if (fullPath) {
    var watcher = this.watchers[fullPath];
    if (watcher && watcher.close) {
      watcher.close();
      delete self.watchers[fullPath];
    }
    getSubDirectories(fullPath, function(fpath) {
      self.close(fpath);
    });
  }
  else {
    Object.keys(self.watchers).forEach(function(fpath) {
      var watcher = self.watchers[fpath];
      if (watcher && watcher.close) {
        watcher.close();
      }
    });
    this.watchers = {};
  }
  // Do not close the Watcher unless all child watchers are closed.
  // https://github.com/yuanchuan/node-watch/issues/75
  if (is.emptyObject(self.watchers)) {
    // should emit once
    if (!this._isClosed) {
      this._isClosed = true;
      process.nextTick(emitClose, this);
    }
  }
}

Watcher.prototype.getWatchedPaths = function(fn) {
  if (is.func(fn)) {
    var self = this;
    if (self._isReady) {
      fn(Object.keys(self.watchers));
    } else {
      self.on('ready', function() {
        fn(Object.keys(self.watchers));
      });
    }
  }
}

function emitReady(self) {
  if (!self._isReady) {
    self._isReady = true;
    // do not call emit for 'ready' until after watch() has returned,
    // so that consumer can call on().
    process.nextTick(function () {
      self.emit('ready');
    });
  }
}

function emitClose(self) {
  self.emit('close');
}

Watcher.prototype.add = function(watcher, info) {
  var self = this;
  info = info || { fpath: '' };
  var watcherPath = path.resolve(info.fpath);
  this.watchers[watcherPath] = watcher;

  // Internal callback for handling fs.FSWatcher 'change' events
  var internalOnChange = function(rawEvt, rawName) {
    if (self.isClosed()) {
      return;
    }

    // normalise lack of name and convert to full path
    var name = rawName;
    if (is.nil(name)) {
      name = '';
    }
    name = path.join(info.fpath, name);

    if (info.options.recursive) {
      hasNativeRecursive(function(has) {
        if (!has) {
          var fullPath = path.resolve(name);
          // remove watcher on removal
          if (!is.exists(name)) {
            self.close(fullPath);
          }
          // watch new created directory
          else {
            var shouldWatch = is.directory(name)
              && !self.watchers[fullPath]
              && shouldNotSkip(name, info.options.filter);

            if (shouldWatch) {
              self.watchDirectory(name, info.options);
            }
          }
        }
      });
    }

    handlePublicEvents(rawEvt, name);
  };

  // Debounced based on the 'delay' option
  var handlePublicEvents = debounce(info, function (evt, name) {
    // watch single file
    if (info.compareName) {
      if (info.compareName(name)) {
        self.emit('change', evt, name);
      }
    }
    // watch directory
    else {
      var filterGuard = guard(info.options.filter);
      filterGuard(name, function() {
        if (self.flag) self.flag = '';
        else self.emit('change', evt, name);
      });
    }
  });

  watcher.on('error', function(err) {
    if (self.isClosed()) {
      return;
    }
    if (is.windows() && err.code === 'EPERM') {
      watcher.emit('change', EVENT_REMOVE, info.fpath && '');
      self.flag = 'windows-error';
      self.close(watcherPath);
    } else {
      self.emit('error', err);
    }
  });

  watcher.on('change', internalOnChange);
}

Watcher.prototype.watchFile = function(file, options, fn) {
  var parent = path.join(file, '../');
  var opts = Object.assign({}, options, {
    // no filter for single file
    filter: null,
    encoding: 'utf8'
  });

  // no need to watch recursively
  delete opts.recursive;

  var watcher = tryWatch(this, parent, opts);
  if (!watcher) {
    return;
  }

  this.add(watcher, {
    type: 'file',
    fpath: parent,
    options: Object.assign({}, opts, {
      encoding: options.encoding
    }),
    compareName: function(n) {
      return is.samePath(n, file);
    }
  });

  if (is.func(fn)) {
    if (fn.length === 1) deprecationWarning();
    this.on('change', fn);
  }
}

Watcher.prototype.updateDelay = function(delay) {
  console.log()
}

Watcher.prototype.watchDirectory = function(dir, options, fn, counter = nullCounter) {
  var self = this;
  var done = counter();
  hasNativeRecursive(function(has) {
    // always specify recursive
    options.recursive = !!options.recursive;
    // using utf8 internally
    var opts = Object.assign({}, options, {
      encoding: 'utf8'
    });
    if (!has) {
      delete opts.recursive;
    }

    // check if it's closed before calling watch.
    if (self._isClosed) {
      done();
      return self.close();
    }

    var watcher = tryWatch(self, dir, opts);
    if (!watcher) {
      done();
      return;
    }

    self.add(watcher, {
      type: 'dir',
      fpath: dir,
      options: options
    });

    if (is.func(fn)) {
      if (fn.length === 1) deprecationWarning();
      self.on('change', fn);
    }

    if (options.recursive && !has) {
      getSubDirectories(dir, function(d) {
        if (shouldNotSkip(d, options.filter)) {
          self.watchDirectory(d, options, null, counter);
        }
      }, counter());
    }

    done();
  });
}

function composeWatcher(watchers) {
  var watcher = new Watcher();
  var filterDups = createDupsFilter();
  var counter = watchers.length;

  watchers.forEach(function(w) {
    w.on('change', filterDups(function(evt, name) {
      watcher.emit('change', evt, name);
    }));
    w.on('error', function(err) {
      watcher.emit('error', err);
    });
    w.on('ready', function() {
      if (!(--counter)) {
        emitReady(watcher);
      }
    });
  });

  watcher.close = function() {
    watchers.forEach(function(w) {
      w.close();
    });
    process.nextTick(emitClose, watcher);
  }

  watcher.getWatchedPaths = function(fn) {
    if (is.func(fn)) {
      var promises = watchers.map(function(w) {
        return new Promise(function(resolve) {
          w.getWatchedPaths(resolve);
        });
      });
      Promise.all(promises).then(function(result) {
        var ret = unique(flat1(result));
        fn(ret);
      });
    }
  }

  return watcher.expose();
}

export default function watch(fpath, options, fn) {
  var watcher = new Watcher();

  if (is.buffer(fpath)) {
    fpath = fpath.toString();
  }

  if (!is.array(fpath) && !is.exists(fpath)) {
    process.nextTick(function() {
      watcher.emit('error',
        new Error(fpath + ' does not exist.')
      );
    });
  }

  if (is.string(options)) {
    throw new Error(`Invalid option, encoding as string is no longer supported. Use { encoding: "${options}" } instead.`)
  }

  if (is.func(options)) {
    fn = options;
    options = {};
  }

  if (arguments.length < 2) {
    options = {};
  }

  if (options.encoding) {
    assertEncoding(options.encoding);
  } else {
    options.encoding = 'utf8';
  }

  if (is.array(fpath)) {
    if (fpath.length === 1) {
      return watch(fpath[0], options, fn);
    }
    var filterDups = createDupsFilter();
    return composeWatcher(unique(fpath).map(function(f) {
      var w = watch(f, options);
      if (is.func(fn)) {
        w.on('change', filterDups(fn));
      }
      return w;
    }));
  }

  if (is.file(fpath)) {
    watcher.watchFile(fpath, options, fn);
    emitReady(watcher);
  }

  else if (is.directory(fpath)) {
    var counter = semaphore(function () {
      emitReady(watcher);
    });
    watcher.watchDirectory(fpath, options, fn, counter);
  }

  return watcher.expose();
}