How to use the readable-stream.Writable function in readable-stream

var file = {
    fieldName: fileStream.name,
    originalFilename: fileStream.filename,
    path: uploadPath(self.uploadDir, fileStream.filename),
    headers: fileStream.headers,
  };
  file.ws = fs.createWriteStream(file.path);
  self.openedFiles.push(file);
  fileStream.pipe(file.ws);
  var counter = new StreamCounter();
  fileStream.pipe(counter);
  var hashWorkaroundStream
    , hash = null;
  if (self.hash) {
    // workaround stream because https://github.com/joyent/node/issues/5216
    hashWorkaroundStream = stream.Writable();
    hash = crypto.createHash(self.hash);
    hashWorkaroundStream._write = function(buffer, encoding, callback) {
      hash.update(buffer);
      callback();
    };
    fileStream.pipe(hashWorkaroundStream);
  }
  file.ws.on('error', function(err) {
    if (!self.error) self.handleError(err);
  });
  file.ws.on('close', function() {
    if (hash) file.hash = hash.digest('hex');
    file.size = counter.bytes;
    self.emit('file', fileStream.name, file);
    endFlush(self);
  });

var gulpBundleAssets = function (options) {
  options = options || {};

  var writable = new readableStream.Writable({objectMode: true});
  var readable = through.obj(function (file, enc, cb) { // noop
    this.push(file);
    cb();
  });

  writable._write = function _write(file, encoding, done) {

    var config;

    if (file.isNull()) {
      this.push(file);
      return done();
    }

    if (file.isStream()) {
      this.emit('error', new gutil.PluginError('gulp-bundle-assets', 'Streaming not supported'));

fieldName: fileStream.name,
    originalFilename: fileStream.filename,
    path: uploadPath(self.uploadDir, fileStream.filename),
    headers: fileStream.headers,
  };
  file.ws = fs.createWriteStream(file.path);
  self.openedFiles.push(file);
  fileStream.pipe(file.ws);
  var counter = new StreamCounter();
  var seenBytes = 0;
  fileStream.pipe(counter);
  var hashWorkaroundStream
    , hash = null;
  if (self.hash) {
    // workaround stream because https://github.com/joyent/node/issues/5216
    hashWorkaroundStream = stream.Writable();
    hash = crypto.createHash(self.hash);
    hashWorkaroundStream._write = function(buffer, encoding, callback) {
      hash.update(buffer);
      callback();
    };
    fileStream.pipe(hashWorkaroundStream);
  }
  counter.on('progress', function() {
    var deltaBytes = counter.bytes - seenBytes;
    seenBytes += deltaBytes;
    self.totalFileSize += deltaBytes;
    if (self.totalFileSize > self.maxFilesSize) {
      if (hashWorkaroundStream) fileStream.unpipe(hashWorkaroundStream);
      fileStream.unpipe(counter);
      fileStream.unpipe(file.ws);
      self.handleError(new Error("maxFilesSize " + self.maxFilesSize + " exceeded"));

#!/usr/bin/env node

var stream = require("readable-stream");

var duplexer2 = require("./");

var writable = new stream.Writable({objectMode: true}),
    readable = new stream.Readable({objectMode: true});

writable._write = function _write(input, encoding, done) {
  if (readable.push(input)) {
    return done();
  } else {
    readable.once("drain", done);
  }
};

readable._read = function _read(n) {
  // no-op
};

// simulate the readable thing closing after a bit
writable.once("finish", function() {

*/

'use strict';
var getUserMedia = require('get-user-media-promise');
var MicrophoneStream = require('microphone-stream');
var RecognizeStream = require('./recognize-stream.js');
var L16 = require('./webaudio-l16-stream.js');
var FormatStream = require('./format-stream.js');
var assign = require('object.assign/polyfill')();
var WritableElementStream = require('./writable-element-stream');
var { Writable } = require('readable-stream');
var ResultStream = require('./result-stream');
var SpeakerStream = require('./speaker-stream');

var preservedMicStream;
var bitBucket = new Writable({
  write: function(chunk, encoding, callback) {
    // when the keepMicrophone option is enabled, unused audio data is sent here so that it isn't buffered by other streams.
    callback();
  },
  objectMode: true, // can still accept strings/buffers
  decodeStrings: false
});

/**
 * @module watson-speech/speech-to-text/recognize-microphone
 */

/**
 * Create and return a RecognizeStream sourcing audio from the user's microphone
 *
 * @param {Object} options - Also passed to {RecognizeStream}, and {FormatStream} when applicable

#!/usr/bin/env node

var stream = require("readable-stream");

var duplexer2 = require("./");

var writable = new stream.Writable({objectMode: true}),
    readable = new stream.Readable({objectMode: true});

writable._write = function _write(input, encoding, done) {
  if (readable.push(input)) {
    return done();
  } else {
    readable.once("drain", done);
  }
};

readable._read = function _read(n) {
  // no-op
};

// simulate the readable thing closing after a bit
writable.once("finish", function() {

function ws (fn) {
  var s = Writable();
  var buf = [];
  s._write = function(chunk, _, next) {
    buf.push(chunk);
    next();
  }
  s.on('finish', function() {
    fn(buf.join(''));
  });
  return s;
}

MqttClient.prototype._setupStream = function () {
  var connectPacket,
    that = this,
    writable = new Writable(),
    parser = mqttPacket.parser(this.options),
    completeParse = null,
    packets = [];

  this._clearReconnect();

  this.stream = this.streamBuilder(this);

  parser.on('packet', function (packet) {
    packets.push(packet);
  });

  function process () {
    var packet = packets.shift(),
      done = completeParse;
    if (packet) {

function multisend (args) {
  var client = mqtt.connect(args);
  var sender = new Writable({
    objectMode: true
  })
  sender._write = function (line, enc, cb) {
    client.publish(args.topic, line.trim(), args, cb);
  }

  client.on('connect', function () {
    pump(process.stdin, split2(), sender, function (err) {
      client.end()
      if (err) {
        throw err
      }
    })
  })
}