Pogostick streaming?

pogostick-stream.js

function noop() {}
var debug = noop;


function feedServer(stream, options, procHandler) {

	stream.on('data', function(buf) {
		var data = buf.toString();
		var msg = data.split('\n');
		var result = procHandler(msg);

		if(typeof result.then === 'function') {
			result.then(function(res) {
				if(!stream.write(res)) {
					stream.on('drain', function() {
						stream.write(res);
					});
				}
			});
		} else {
			if(!steam.write(res)) {
				stream.on('drain', function() {
					stream.write(res);
				});
			}
		}
	});
	
	if(options.on.error) {
		stream.on('error', function(err) {
			options.on.error(err);
		});
	}

	if(options.on.end) {
		stream.on('end', function() {
			options.on.end();
		});
	}

}

// turn streams into a request-response type of function.
function mkReqHandler(con) {
	// Doesn't inherit from any prototype, so safe to use for..in.
	listeners = Object.create(null);

	con.on('data', function(buf) {
		var data = buf.toString().split('\n');
		var rand = data[2];
		var stamp = data[1];
		var key = rand + '-' + stamp;
		
		if(listeners[key] !== undefined) {
			listeners[key](null, data);
			delete listeners[key];
		}
	});

	// End is called whenever there is a timeout or error. Skip handling it more than that for now.
	con.on('end', function() {
		var err = new Error('Connection was closed');
		// connection was ended, so toss an error to all listeners.
		for(var key in listeners) {
			listeners[key](err);
			delete listeners[key];
		}
	});

	return function(body, cb) {
		listeners[options.body.rand + '-' + options.body.stamp] = cb;
		if(!con.write(body.str)) {
			var onFree;
			onFree = function() {
				con.removeListener('drain', onFree);
				con.write(body.str);
			};
			con.on('drain', onFree);
		}
	};
}