From 7b8c67e941d68c6866e9627e0ced0a18cef8d8ae Mon Sep 17 00:00:00 2001
From: oleg <oleg.korobenko@gmail.com>
Date: Thu, 9 Apr 2015 21:55:29 +0300
Subject: [PATCH] add distributor with minimal testing

---
 lib/distributor.js  | 83 +++++++++++++++++++++++++++++++++++++++++++++
 test/distributor.js | 70 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 153 insertions(+)
 create mode 100644 lib/distributor.js
 create mode 100644 test/distributor.js

diff --git a/lib/distributor.js b/lib/distributor.js
new file mode 100644
index 0000000..e37ddfa
--- /dev/null
+++ b/lib/distributor.js
@@ -0,0 +1,83 @@
+'use strict';
+
+var	Steppy = require('twostep').Steppy,
+	_ = require('underscore'),
+	Node = require('./node').Node;
+
+
+function Distributor(params) {
+	var self = this;
+	// nodes to execute builds
+	self.nodes = _(params.nodes).map(function(nodeParams) {
+		return self._createNode(nodeParams);
+	});
+	// queued projects to build
+	self.queue = [];
+}
+
+exports.Distributor = Distributor;
+
+Distributor.prototype._createNode = function(nodeParams) {
+	return new Node(nodeParams);
+};
+
+Distributor.prototype._runNext = function(callback) {
+	var self = this;
+
+	Steppy(
+		function() {
+			var node;
+			var queueItemIndex = _(self.queue).findIndex(function(item) {
+				node = _(self.nodes).find(function(node) {
+					return node.hasFreeExecutor(item.project);
+				});
+				return node;
+			});
+
+			// quit if we have no suitable project
+			if (queueItemIndex) {
+				return callback();
+			}
+
+			this.pass(node);
+
+			var queueItem = self.queue[queueItemIndex];
+			this.pass(queueItemIndex, queueItem);
+
+			queueItem.build.status = 'in-progress';
+			self._updateBuild(queueItem.build, this.slot());
+		},
+		function(err, node, queueItemIndex, queueItem, build) {
+			self.queue.splice(queueItemIndex, 1);
+
+			var stepCallback = this.slot();
+			node.run(queueItem.project, build.params, function(err) {
+				build.status = err ? 'error' : 'done';
+				self._updateBuild(build, stepCallback);
+			});
+		},
+		callback
+	);
+};
+
+Distributor.prototype._updateBuild = function(build, callback) {
+	callback(null, build);
+};
+
+Distributor.prototype.run = function(project, params, callback) {
+	var self = this;
+	Steppy(
+		function() {
+			self._updateBuild({
+				project: project,
+				params: params,
+				status: 'waiting'
+			}, this.slot());
+		},
+		function(err, build) {
+			self.queue.push({project: project, build: build});
+			self._runNext(this.slot());
+		},
+		callback
+	);
+};
diff --git a/test/distributor.js b/test/distributor.js
new file mode 100644
index 0000000..6ef38f0
--- /dev/null
+++ b/test/distributor.js
@@ -0,0 +1,70 @@
+'use strict';
+
+var Distributor = require('../lib/distributor').Distributor,
+	Node = require('../lib/node').Node,
+	expect = require('expect.js');
+
+
+describe('Distributor', function() {
+	var distributor,
+		project1 = {name: 'project1'};
+
+	describe('with sucess project', function() {
+		var originalCreateNode,
+			originalUpdateBuild;
+
+		before(function() {
+			originalCreateNode = Distributor.prototype._createNode;
+			Distributor.prototype._createNode = function(params) {
+				var node = new Node(params);
+				node._createExecutor = function() {
+					return {run: function(params, callback) {
+						setTimeout(callback, 100);
+					}};
+				};
+				return node;
+			};
+
+			originalUpdateBuild = Distributor.prototype._updateBuild;
+
+			var updateBuildNumber = 1;
+			Distributor.prototype._updateBuild = function(build, callback) {
+				if (updateBuildNumber === 1) {
+					expect(distributor.queue).length(0);
+					expect(build.status).equal('waiting');
+				} else if (updateBuildNumber === 2) {
+					expect(distributor.queue).length(1);
+					expect(build.status).equal('in-progress');
+				} else if (updateBuildNumber === 3) {
+					expect(distributor.queue).length(0);
+					expect(build.status).equal('done');
+				} else {
+					throw new Error('Should never happend');
+				}
+				updateBuildNumber++;
+				callback(null, build)
+			};
+		});
+
+		it('instance should be created without errors', function() {
+			distributor = new Distributor({
+				nodes: [{type: 'local', maxExecutorsCount: 1}]
+			});
+		});
+
+		it('should run without errors', function() {
+			distributor.run(project1, {}, function(err) {
+				expect(err).not.ok();
+			});
+		});
+
+		it('wait for project done (should no errors)', function(done) {
+			setTimeout(done, 100);
+		});
+
+		after(function() {
+			Distributor.prototype._createNode = originalCreateNode;
+			Distributor.prototype._updateBuild = originalUpdateBuild;
+		});
+	});
+});