index.js
function pipe(dest, options) {
    var settings = options || {}
    var onend = once(onended)

    emitter.on('data', ondata)
    emitter.on('error', onerror)
    emitter.on('end', cleanup)
    emitter.on('close', cleanup)

    // If the `end` option is not supplied, `dest.end()` will be called when the
    // `end` or `close` events are received.  Only `dest.end()` once.
    if (!dest._isStdio && settings.end !== false) {
      emitter.on('end', onend)
    }

    dest.on('error', onerror)
    dest.on('close', cleanup)

    dest.emit('pipe', emitter)

    return dest

    // End destination.
    function onended() {
      if (dest.end) {
        dest.end()
      }
    }

    // Handle data.
    function ondata(chunk) {
      if (dest.writable) {
        dest.write(chunk)
      }
    }

    // Clean listeners.
    function cleanup() {
      emitter.removeListener('data', ondata)
      emitter.removeListener('end', onend)
      emitter.removeListener('error', onerror)
      emitter.removeListener('end', cleanup)
      emitter.removeListener('close', cleanup)

      dest.removeListener('error', onerror)
      dest.removeListener('close', cleanup)
    }

    // Close dangling pipes and handle unheard errors.
    function onerror(err) {
      var handlers = emitter._events.error

      cleanup()

      // Cannot use `listenerCount` in node <= 0.12.
      if (!handlers || handlers.length === 0 || handlers === onerror) {
        throw err // Unhandled stream error in pipe.
      }
    }
  }
function pipe(dest, options) {
    var settings = options || {}
    var onend = once(onended)

    emitter.on('data', ondata)
    emitter.on('error', onerror)
    emitter.on('end', cleanup)
    emitter.on('close', cleanup)

    // If the `end` option is not supplied, `dest.end()` will be called when the
    // `end` or `close` events are received.  Only `dest.end()` once.
    if (!dest._isStdio && settings.end !== false) {
      emitter.on('end', onend)
    }

    dest.on('error', onerror)
    dest.on('close', cleanup)

    dest.emit('pipe', emitter)

    return dest

    // End destination.
    function onended() {
      if (dest.end) {
        dest.end()
      }
    }

    // Handle data.
    function ondata(chunk) {
      if (dest.writable) {
        dest.write(chunk)
      }
    }

    // Clean listeners.
    function cleanup() {
      emitter.removeListener('data', ondata)
      emitter.removeListener('end', onend)
      emitter.removeListener('error', onerror)
      emitter.removeListener('end', cleanup)
      emitter.removeListener('close', cleanup)

      dest.removeListener('error', onerror)
      dest.removeListener('close', cleanup)
    }

    // Close dangling pipes and handle unheard errors.
    function onerror(err) {
      var handlers = emitter._events.error

      cleanup()

      // Cannot use `listenerCount` in node <= 0.12.
      if (!handlers || handlers.length === 0 || handlers === onerror) {
        throw err // Unhandled stream error in pipe.
      }
    }
  }
function pipe ( dest , options ) { var settings = options || { } var onend = once ( onended ) emitter . on ( 'data' , ondata ) emitter . on ( 'error' , onerror ) emitter . on ( 'end' , cleanup ) emitter . on ( 'close' , cleanup ) // If the `end` option is not supplied, `dest.end()` will be called when the // `end` or `close` events are received.  Only `dest.end()` once. if ( ! dest . _isStdio && settings . end !== false ) { emitter . on ( 'end' , onend ) } dest . on ( 'error' , onerror ) dest . on ( 'close' , cleanup ) dest . emit ( 'pipe' , emitter ) return dest // End destination. function onended ( ) { if ( dest . end ) { dest . end ( ) } } // Handle data. function ondata ( chunk ) { if ( dest . writable ) { dest . write ( chunk ) } } // Clean listeners. function cleanup ( ) { emitter . removeListener ( 'data' , ondata ) emitter . removeListener ( 'end' , onend ) emitter . removeListener ( 'error' , onerror ) emitter . removeListener ( 'end' , cleanup ) emitter . removeListener ( 'close' , cleanup ) dest . removeListener ( 'error' , onerror ) dest . removeListener ( 'close' , cleanup ) } // Close dangling pipes and handle unheard errors. function onerror ( err ) { var handlers = emitter . _events . error cleanup ( ) // Cannot use `listenerCount` in node <= 0.12. if ( ! handlers || handlers . length === 0 || handlers === onerror ) { throw err // Unhandled stream error in pipe. } } }
Pipe the processor into a writable stream.  Basically `Stream#pipe`, but inlined and simplified to keep the bundled size down. See <https://github.com/nodejs/node/blob/master/lib/stream.js#L26>.
Pipe the processor into a writable stream . Basically Stream#pipe but inlined and simplified to keep the bundled size down . See <https : // github . com / nodejs / node / blob / master / lib / stream . js#L26 > .