42 lines
		
	
	
		
			885 B
		
	
	
	
		
			JavaScript
		
	
	
	
	
	
			
		
		
	
	
			42 lines
		
	
	
		
			885 B
		
	
	
	
		
			JavaScript
		
	
	
	
	
	
| 'use strict';
 | |
| 
 | |
| const { PassThrough } = require('stream');
 | |
| 
 | |
| module.exports = function (/*streams...*/) {
 | |
|   var sources = []
 | |
|   var output  = new PassThrough({objectMode: true})
 | |
| 
 | |
|   output.setMaxListeners(0)
 | |
| 
 | |
|   output.add = add
 | |
|   output.isEmpty = isEmpty
 | |
| 
 | |
|   output.on('unpipe', remove)
 | |
| 
 | |
|   Array.prototype.slice.call(arguments).forEach(add)
 | |
| 
 | |
|   return output
 | |
| 
 | |
|   function add (source) {
 | |
|     if (Array.isArray(source)) {
 | |
|       source.forEach(add)
 | |
|       return this
 | |
|     }
 | |
| 
 | |
|     sources.push(source);
 | |
|     source.once('end', remove.bind(null, source))
 | |
|     source.once('error', output.emit.bind(output, 'error'))
 | |
|     source.pipe(output, {end: false})
 | |
|     return this
 | |
|   }
 | |
| 
 | |
|   function isEmpty () {
 | |
|     return sources.length == 0;
 | |
|   }
 | |
| 
 | |
|   function remove (source) {
 | |
|     sources = sources.filter(function (it) { return it !== source })
 | |
|     if (!sources.length && output.readable) { output.end() }
 | |
|   }
 | |
| }
 |