map-async.js

var Transform = require("stream").Transform

module.exports = mapAsync

function mapAsync(write, end) {
  var stream = new Transform({ objectMode: true })
  stream._transform = function (value, _, cb) {
    write(value, cb)
  }
  stream._flush = function (cb) {
    if (end) end(cb)
    else cb()
  }
  return stream
}

uppercase.js

// to create a "through" stream use mapAsync
var stream = mapAsync(function (word, cb) {
  this.push(word.toUpperCase()); // put objects in the buffer
  cb(); // pull in the next value
});

// to consume stuff from stream use a Writable sink
// this handles buffering / backpressure etc
var sink = new Writable({ objectMode: true });
sink._write = function (chunk, _, cb) {
  console.log("got chunk", chunk);
  cb();
};

// write like normally do
stream.write("hello");
stream.write("world");

// notice sink does not miss chunk
stream.pipe(sink);