streams-example

index.js

'use strict';

var util = require('util');
var stream = require('stream');
var Writable = stream.Writable || require('readable-stream').Writable;

function dbConnect(connectionString) {
  console.log('connectionString',
              connectionString);
  return {
    insert: function insert(data, cb) {
      console.log('data', data, 'inserted');
      return cb(null);
    }
  };
}
exports.dbConnect = dbConnect;

function DbWriteStream(options) {

  // use without 'new'
  if (!(this instanceof DbWriteStream)) {
    return new DbWriteStream(options);
  }

  // 'super' initialization
  options.objectMode = true;
  Writable.call(this, options);

  this.connection = options.connection;
}

util.inherits(DbWriteStream, Writable);

DbWriteStream.prototype._write = function write(chunk, enc, cb) {
  this.connection.insert(chunk, cb);
};
exports.DbWriteStream = DbWriteStream;

package.json

{
  "name": "streams-example",
  "version": "1.0.0",
  "description": "Example of usage of node.js streams",
  "main": "index.js",
  "scripts": {
    "test": "./node_modules/.bin/mocha test.js"
  },
  "repository": {
    "type": "git",
    "url": "git+ssh://git@gist.github.com/5f3eb075aa3e3d84ba75.git"
  },
  "keywords": [
    "node",
    "streams",
    "test"
  ],
  "author": "Thiago Sigrist",
  "license": "MIT",
  "bugs": {
    "url": "https://gist.github.com/5f3eb075aa3e3d84ba75"
  },
  "homepage": "https://gist.github.com/5f3eb075aa3e3d84ba75",
  "dependencies": {
    "chai": "^3.4.1",
    "mocha": "^2.3.4",
    "readable-stream": "^2.0.4",
    "sinon": "^1.17.2",
    "sinon-chai": "^2.8.0"
  }
}

test.js

'use strict';

var chai = require('chai');
var sinon = require('sinon');
var sinonChai = require('sinon-chai');
chai.use(sinonChai);

var expect = chai.expect;

var dbstreams = require('.');
var DbWriteStream = dbstreams.DbWriteStream;

describe('DbWriteStream', function() {

  var data = {
    foo: 1,
    bar: 2
  };

  it('should insert to DB', function() {
    var conn = {
      insert: sinon.spy()
    };
    var dbws = new DbWriteStream({
      connection: conn
    });

    dbws.write(data);

    expect(conn.insert).to.have.been.calledWith(data);
  });
});