Module to allow a Sequelize select query to be converted into a stream.

createSequelizeStream.js

"use strict";

var Readable = require('stream').Readable;

/**
 * Create a stream from batching a Sequelize select query.
 *
 * @param {Function} query - Function that returns a promise.
 * @param {object}   options
 * @param {number}   options.limit  - The limit for the number of rows to return in each "batch"
 * @param {number}   options.offset - The row offset to start at.
 * @returns {stream.Readable}
 * @throws if options.limit or options.offset is not supplied.
 */
function createSequelizeStream(query, options) {
	var stream = new Readable({ objectMode: true });

	if (options.limit == void 0) {
		throw new Error('createSequelizeStream: <options.limit> required.');
	}

	if (options.offset == void 0) {
		throw new Error('createSequelizeStream: <options.offset> required.');
	}

	stream._read = function () {
		var self = this;

		self.pause();

		query(options)
			.then(function (rows) {
				if (rows.length === 0) {
					self.push(null)
				}
				else {
					rows.forEach(function (row) {
						self.push(row);
					});
					options.offset += options.limit;
				}

				self.resume();
			})
			.catch(function (err) {
				options.offset += options.limit;
				self.resume();
				self.emit('error', err);
			});
	};

	return stream;
}

module.exports = createSequelizeStream;