Source stream leaked then piped

[CrabDude]

When piping multiple source streams to a single through stream for the purposes of combining them, the source streams never close. The result is an "EventEmitter memory leak" for the close, drain and finish events. You can verify by adding the following code to through:

  stream.on('pipe', function(src) {
    console.log('on pipe')
    src.on('end', function() {
      console.log('on end')
    })
    src.on('finish', function() {
      console.log('on finish')
    })
    src.on('close', function() {
      console.log('on close')
    })
  })

Here's the code to reproduce:

var through = require('through')
var http = require('http')
var fs = require('fs')
var path = require('path')
var throughStream = through(null, function(){}) // Don't close the stream
throughStream.pipe(fs.createWriteStream(path.join(__dirname, 'server.log')))

http.createServer(function(req, res) {
  req.pipe(throughStream)
  res.end('hello world')
}).listen(8001)

I'm assuming this is an issues with the the destination (through) stream staying open, but I'm not sure what specifically about pipe is causing it or how to get around it ATM.

[jcrugzz]

@CrabDude process.stdout is a bad example since its a writable stream that never ends. Try it with something else to confirm the behavior you are seeing.

[CrabDude]

I suppose this doesn't matter. I didn't realize .pipe supports this already:

source.pipe(dest, {end: false})

FWIW, looks like _stream_readable.pipe adds a close listener on the destination stream for cleanup purposes. Further, since I wasn't passing end: true, dest.end() is called instead of cleanup directly. In through, autoDestroy is irrelevant since end checks stream.readable before destroying. Since destination is still readable it doesn't destroy and thus never closes.

This seems like a leaky abstraction and extremely backwards on core's part since the source stream leaks dependent on the destination stream implementation.

If you want this functionality with through, you have 2 options: Option 1: Use pipe options

let throughStream = through(null, ()=>{})
src.pipe(throughStream), {end: false})

Option 2: Add the following to through

stream.on('pipe', src => {
  // Unfortunately there's not a better way to do this
  src.removeAllListeners('end')
  src.once('end', ()=> src.unpipe(stream))
})

So, ya, just use pipeOptions I suppose.

[CrabDude]

@jcrugzz The example / issue is more about multiple source streams than the destination stream. I've updated the example to eliminate the confusion you mention, but the issue still exists.

[dominictarr]

@CrabDude this is not really a case that node streams (classic or new) handles very well. I mostly use pull-streams which have much more well defined behavior around closing and errors, and are much simpler.

Probably the simplest way to make this work with node streams is to use setMaxListeners https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n

[CrabDude]

Thanks. I'll checkout pull-streams. The end option does solve this at least from the source perspective, which shifts the conversation from "is it possible" to "would it be better to be able to specify this in the destination stream".

On Thursday, April 30, 2015, Dominic Tarr notifications@github.com wrote:

@CrabDude https://github.com/CrabDude this is not really a case that node streams (classic or new) handles very well. I mostly use pull-streams https://github.com/dominictarr/pull-stream which have much more well defined behavior around closing and errors, and are much simpler.

Probably the simplest way to make this work with node streams is to use setMaxListeners https://nodejs.org/api/events.html#events_emitter_setmaxlisteners_n

— Reply to this email directly or view it on GitHub https://github.com/dominictarr/through/issues/33#issuecomment-97738268.

Adam Crabtree Noderiety PO Box 3823 Los Altos, CA 94024 Telephone: _(_469) 777-8415 https://mail.google.com/ Email: dude@noderiety.com

This e-mail message is intended for the use of the addressee(s) only and may contain information that is privileged-confidential and exempt from disclosure under applicable law. If the reader of the message is not the intended recipient or agent responsible for delivering the message to the intended recipient, you are hereby notified that any review, dissemination, distribution or copying of this communication is strictly prohibited. If you receive this communication in error, please notify us immediately by reply e-mail and delete the original message. Thank you.