Trickle

.gitignore

node_modules

README.md

trickle

Through stream for Node to slow incoming data to specific intervals.

For example, the following script will pipe itself to process.stdout, limited to one chunk per second.

var es = require('event-stream')
  , trickle = require('trickle')
  , fs = require('fs')

var stream = trickle({
  interval: 1000
})

fs.createReadStream(__filename, {
  encoding: 'utf8'
}).pipe(es.split(/\s+/g))
  .pipe(stream)
  .pipe(es.join('\n'))
  .pipe(process.stdout)

If you're looking to throttle data by bytes per second, check out throttle or brake.

Of course, this stream buffers data. If you want to discard input to avoid the stream filling up, use the limit option: this limits the total amount of chunks that the stream will buffer.

By default, the stream will ignore any new data if the buffer is full. Using the mru flag, the stream will instead remove the oldest chunk and add the new one to the end of the queue.

// Emits the most recent recieved chunk
// every second.
trickle({
      mru: true
    , limit: 1
    , interval: 1000
})

// Stores up to 50 chunks, discarding any
// after that. Flushes one chunk every
// five seconds.
trickle({
      limit: 50
    , interval: 5000
})

Parameters

• interval: The interval period, in milliseconds. Defaults to 50.
• limit: Maximum amount of chunks to buffer at once. Omit this field to buffer content endlessly.
• mru: Remove old chunks to make space for new ones. Disabled by default.
• flush: Amount of chunks to flush each interval. Defaults to 1.

example.js

/**
 * This script should write itself to
 * stdout, three lines at a time.
 */
var es = require('event-stream')
  , trickle = require('./index')
  , fs = require('fs')

var stream = trickle({
    interval: 2500
  , flush: 3
})

fs.createReadStream(__filename, {
  encoding: 'utf8'
}).pipe(es.split(/\n/g))
  .pipe(stream)
  .pipe(es.join('\n'))
  .pipe(process.stdout)

index.js

var through = require('through');

module.exports = function TrickleStream(options) {
    var options = options || {}
      , queued = []
      , finishing = false
      , interval = false
      , stream

    options.interval = options.interval || 50;
    options.flush = options.flush || 1;
    options.limit = options.limit || 0;
    options.mru = options.mru || false;

    interval = setInterval(function() {
        var next, i

        for (i = 0; i < options.flush; i += 1) {
            next = queued.splice(0, 1)[0];
            if (!next) break;
            if (!i) stream.resume();
            stream.emit('data', next);
        }
        if (finishing && !queued.length) {
            stream.emit('end');
            clearInterval(interval);
        }
    }, options.interval);

    stream = through(function write(data) {
        this.pause();
        if (!options.limit || queued.length < options.limit) {
            queued.push(data);
        } else
        if (options.mru) {
            queued.shift();
            queued.push(data);
        }
    }, function end() {
        finishing = true;
    });

    return stream;
};

package.json

{
  "name": "trickle",
  "version": "0.0.2",
  "description": "Slows incoming stream data to specific intervals",
  "main": "index.js",
  "repository": {
    "type": "git",
    "url": "git://gist.github.com/3740721.git"
  },
  "keywords": [
    "slow",
    "stream",
    "throttle",
    "trickle",
    "periodic",
    "delay"
  ],
  "author": "Hugh Kennedy",
  "license": "BSD",
  "dependencies": {
    "through": "~1.1.0"
  }
}