Files F120753301

index.js
No OneTemporary
Actions

Subscribers

None

File Metadata

Created: Sun, Jul 6, 19:17

index.js
View Options

	//filter will reemit the data if cb(err,pass) pass is truthy

	// reduce is more tricky
	// maybe we want to group the reductions or emit progress updates occasionally
	// the most basic reduce just emits one 'data' event after it has recieved 'end'


	var Stream = require('stream').Stream


	//create an event stream and apply function to each .write
	//emitting each response as data
	//unless it's an empty callback

	module.exports = function (mapper, opts) {

	var stream = new Stream()
	, inputs = 0
	, outputs = 0
	, ended = false
	, paused = false
	, destroyed = false
	, lastWritten = 0
	, inNext = false

	opts = opts \|\| {};
	var errorEventName = opts.failures ? 'failure' : 'error';

	// Items that are not ready to be written yet (because they would come out of
	// order) get stuck in a queue for later.
	var writeQueue = {}

	stream.writable = true
	stream.readable = true

	function queueData (data, number) {
	var nextToWrite = lastWritten + 1

	if (number === nextToWrite) {
	// If it's next, and its not undefined write it
	if (data !== undefined) {
	stream.emit.apply(stream, ['data', data])
	}
	lastWritten ++
	nextToWrite ++
	} else {
	// Otherwise queue it for later.
	writeQueue[number] = data
	}

	// If the next value is in the queue, write it
	if (writeQueue.hasOwnProperty(nextToWrite)) {
	var dataToWrite = writeQueue[nextToWrite]
	delete writeQueue[nextToWrite]
	return queueData(dataToWrite, nextToWrite)
	}

	outputs ++
	if(inputs === outputs) {
	if(paused) paused = false, stream.emit('drain') //written all the incoming events
	if(ended) end()
	}
	}

	function next (err, data, number) {
	if(destroyed) return
	inNext = true

	if (!err \|\| opts.failures) {
	queueData(data, number)
	}

	if (err) {
	stream.emit.apply(stream, [ errorEventName, err ]);
	}

	inNext = false;
	}

	// Wrap the mapper function by calling its callback with the order number of
	// the item in the stream.
	function wrappedMapper (input, number, callback) {
	return mapper.call(null, input, function(err, data){
	callback(err, data, number)
	})
	}

	stream.write = function (data) {
	if(ended) throw new Error('map stream is not writable')
	inNext = false
	inputs ++

	try {
	//catch sync errors and handle them like async errors
	var written = wrappedMapper(data, inputs, next)
	paused = (written === false)
	return !paused
	} catch (err) {
	//if the callback has been called syncronously, and the error
	//has occured in an listener, throw it again.
	if(inNext)
	throw err
	next(err)
	return !paused
	}
	}

	function end (data) {
	//if end was called with args, write it,
	ended = true //write will emit 'end' if ended is true
	stream.writable = false
	if(data !== undefined) {
	return queueData(data, inputs)
	} else if (inputs == outputs) { //wait for processing
	stream.readable = false, stream.emit('end'), stream.destroy()
	}
	}

	stream.end = function (data) {
	if(ended) return
	end(data)
	}

	stream.destroy = function () {
	ended = destroyed = true
	stream.writable = stream.readable = paused = false
	process.nextTick(function () {
	stream.emit('close')
	})
	}
	stream.pause = function () {
	paused = true
	}

	stream.resume = function () {
	paused = false
	}

	return stream
	}

Event Timeline