'use strict'; var Cell = require('./cell'), Pledge = require('./pledge'); var Pipeline = function(sessions) { this._cells = sessions.map(function(session) { return new Cell(session) }); this._stopped = { incoming: false, outgoing: false }; }; Pipeline.prototype.processIncomingMessage = function(message, callback, context) { if (this._stopped.incoming) return; this._loop('incoming', this._cells.length - 1, -1, -1, message, callback, context); }; Pipeline.prototype.processOutgoingMessage = function(message, callback, context) { if (this._stopped.outgoing) return; this._loop('outgoing', 0, this._cells.length, 1, message, callback, context); }; Pipeline.prototype.close = function(callback, context) { this._stopped = { incoming: true, outgoing: true }; var closed = this._cells.map(function(a) { return a.close() }); if (callback) Pledge.all(closed).then(function() { callback.call(context) }); }; Pipeline.prototype._loop = function(direction, start, end, step, message, callback, context) { var cells = this._cells, n = cells.length, self = this; while (n--) cells[n].pending(direction); var pipe = function(index, error, msg) { if (index === end) return callback.call(context, error, msg); cells[index][direction](error, msg, function(err, m) { if (err) self._stopped[direction] = true; pipe(index + step, err, m); }); }; pipe(start, null, message); }; module.exports = Pipeline;