Stream of prime numbers in vanilla Node.JS

eratostream.js

// SICP 3.5.2: https://sarabander.github.io/sicp/html/3_002e5.xhtml#g_t3_002e5_002e2
// Node v0.10.31

var stream = require('stream');


var getPrimes = function (count) {
  integers(2)
    .pipe(primes())
    .pipe(take(count))
    .pipe(stringize())
    .pipe(process.stdout);
};


var objectStreamOptions = {
  objectMode: true,
  // No buffering for infinite streams.
  highWaterMark: 0
};


var integers = (function () {
  var prototype = Object.create(stream.Readable.prototype);

  prototype._read = function () {
    this.push(this.value++);
  };

  return function (start) {
    var obj = Object.create(prototype);
    stream.Readable.call(obj, objectStreamOptions);

    Object.defineProperty(obj, 'value', {
      value: start || 0,
      writable: true
    });

    return obj;
  };
}());


/**
 * Push values to the destination stream as they come.
 */
var loopback = (function () {
  var prototype = Object.create(stream.Writable.prototype);

  prototype._write = function (number, _, next) {
    this.dest.push(number);
    next();
  };

  return function (dest) {
    var obj = Object.create(prototype);
    stream.Writable.call(obj, objectStreamOptions);

    Object.defineProperty(obj, 'dest', {
      value: dest
    });

    return obj;
  };
}());


/**
 * Part of primes() pipeline.
 */
var noMultiplesOf = (function () {
  var prototype = Object.create(stream.Transform.prototype);

  prototype._transform = function (number, _, next) {
    if (number % this.base) {
      this.push(number);
    }
    next();
    this.pull();
  };

  return function (base) {
    var obj = Object.create(prototype);
    stream.Transform.call(obj, objectStreamOptions);

    Object.defineProperty(obj, 'base', {
      value: base
    });

    return obj;
  };
}());


var primes = (function () {
  var prototype = Object.create(stream.Transform.prototype);

  prototype._transform = function (number, _, next) {
    if (this.pipeline) {
      this.pipeline.pull = next;
      this.pipeline.write(number);
    }
    else if (this.first == null) {
      this.first = number;
      this.push(number);
      next();
    }
    else {
      (this.pipeline = noMultiplesOf(this.first))
           .pipe(primes())
           .pipe(this.loopback);
      this._transform(number, _, next);
    }
  };

  function primes() {
    var obj = Object.create(prototype);
    stream.Transform.call(obj, objectStreamOptions);

    Object.defineProperty(obj, 'loopback', {
      value: loopback(obj)
    });

    return obj;
  }

  return primes;
}());


/**
 * Cut the stream if it gets too long.
 */
var take = (function () {
  var prototype = Object.create(stream.Transform.prototype);

  prototype._transform = function (chunk, encoding, next) {
    if (this.count > 0) {
      this.output.push(chunk);
      this.count -= 1;
    }
    else {
      // Fill the buffer so that the flow stops eventually.
      this.push(chunk);
      if (!this.count) {
        // Interrupt the reader.
        this.output.push(null);
        this.count = -1;
      }
    }
    next();
  };

  // Other methods and events need to be proxied too, actually.
  prototype.pipe = function (dest, options) {
    return this.output.pipe(dest, options);
  };

  return function (count) {
    var obj = Object.create(prototype);
    stream.Transform.call(obj, objectStreamOptions);

    Object.defineProperties(obj, {
      count: {
        value: count,
        writable: true
      },
      output: {
        value: stream.PassThrough(objectStreamOptions)
      }
    });

    return obj;
  };
}());


/**
 * [1, 2, 3, 4] -> "1, 2, 3, 4, ..."
 */
var stringize = (function () {
  var prototype = Object.create(stream.Transform.prototype);

  prototype._transform = function (chunk, encoding, next) {
    this.push(chunk + ', ');
    next();
  };

  prototype._flush = function (next) {
    this.push('...\n');
    next();
  };

  return function () {
    var obj = Object.create(prototype);
    stream.Transform.call(obj, { highWaterMark: 0 });
    obj._writableState.objectMode = true;
    obj._readableState.objectMode = false;
    obj._readableState.encoding = 'ascii';

    return obj;
  };
}());


// Adjust process.maxTickDepth for larger values.
getPrimes(100);