gistfile1.js

var lineStream; <-- assume this is a readable stream of lines of text utf8 encoded

var through2 = require('through2')

var terminatingStream = through2(function(line, enc, cb) {
  query('INSERT INTO line_table line VALUE ($)', [line], function(err, res) {
    if (err) return cb(err);
    return cb(null, line);
  });
});


//pipe the output of terminatingStream to standard out
//so the data will start 'flowing' through the pipe
lineStream.pipe(terminatingStream).pipe(process.stdout);