sample of what a streaming orm might look like. Includes prefix and suffix support.

streaming-orm.js

//////////////////////////////////////////
// Dam
//////////////////////////////////////////
// Simple transformer stream

// Allows for custom iterator, prefix, and suffix

// Set readable and writable in constructor.
// Inherit from base stream class.
var Dam = function(options) {

	if (options.prefix) {
		this.prefix = options.prefix;
	}
	if (options.suffix) {
		this.suffix = options.suffix;
	}

	// this.readable = true;
	this.writable = true;


	// initialize on next tick
	if (this.prefix) {
		var self = this;
		process.nextTick(function() {
			// Write prefix to stream if specified
			self.emit('data', self.prefix);
		});
	}
};
require('util').inherits(Dam, require('stream'));

// Extract args to `write` and emit as `data` event.
Dam.prototype.write = function(model, iterator, cb) {
	var self = this;

	iterator(model, function(err, transformedModel) {

		// Write transformed model to stream
		self.emit('data', transformedModel);

		// Inform that we're finished
		// Pass error back in callback, if one exists
		return cb(err);
	});
};

// If err set, emit `error`, otherwise emit `end` event.
Dam.prototype.end = function(err) {
	if (err) this.emit('error', err);
	else {
		console.log(this.suffix, "!!!");
		if (this.suffix) {
			this.emit('data', this.suffix);
		}
		this.emit('end');
	}
};



//////////////////////////////////////////
// Mocked ORM
//////////////////////////////////////////
var User = {
	findAll: function() {
		return User.deferred;
	},

	my: {},

	deferred: {

		prefix: function(prefix) {
			User.my.prefix = prefix;
			return User.deferred;
		},

		suffix: function(suffix) {
			User.my.suffix = suffix;
			return User.deferred;
		},

		stream: function(iterator) {
			var dam = new Dam({
				prefix: User.my.prefix,
				suffix: User.my.suffix
			});

			// Pretend to load several models
			require('async').each([{
				loc: 'http://google.com/1'
			}, {
				loc: 'http://google.com/2'
			}, {
				loc: 'http://google.com/3'
			}, {
				loc: 'http://google.com/4'
			}, {
				loc: 'http://google.com/5'
			}],
			// Iterate over each model

			function(model, cb) {
				// Fake a wait from the server
				setTimeout(function() {
					dam.write(model, iterator, cb);
				}, Math.round(500 + Math.random() * 300));
			},
			// Complete the stream

			function(err) {
				dam.end(err);
			});

			return dam;
		}
	}
};

//////////////////////////////////////////
// Basic Usage
//////////////////////////////////////////
var http = require('http');
http.createServer(function(req, res) {
	if (req.url === '/favicon.ico') return;

	User.findAll({
		limit: 50
	}).stream(function(model, emit) {
		// Do iterative thing w/ data (could be an asynchronous thing!)
		var url = model.loc;
		emit(null, "<uri>" + url + "</uri>\n");
	}).pipe(res);

}).listen(1337, '127.0.0.1');


//////////////////////////////////////////
// Advanced Usage
//////////////////////////////////////////
var http = require('http');
http.createServer(function(req, res) {
	if (req.url === '/favicon.ico') return;

	User.findAll({
		limit: 50
	})
		.prefix('<code>')
		.suffix('</code>')
		.stream(function(model, emit) {
		emit(null, "<uri>" + model.loc + "</uri>\n");
	})
		.pipe(res);

}).listen(1338, '127.0.0.1');