Pool

poll.js

var async = require('async');

module.exports.Pool = function (factory) {
	var private = {}
	var public = {}

	private.clients = {};
	private.queue = [];

	private.draining = false;

	factory.validate = factory.validate || function (client, cb) {
		setImmediate(cb, null);
	};

	factory.max = parseInt(factory.max, 10);
	factory.min = parseInt(factory.min, 10);
	if (!isNaN(factory.min) && factory.min > 0) {
		factory.min = Math.min(factory.min, factory.max);
	} else {
		factory.min == undefined;
	}

	factory.max = Math.max(isNaN(factory.max) ? 1 : factory.max, 1);

	private.validate = function (client, cb) {
		if (private.draining) return setImmediate(cb, {msg: 'pool is draining'});

		factory.validate(client, cb);
	}

	private.createResource = function (cb) {
		//if (private.draining) return setImmediate(cb, {msg: 'pool is draining'});
		var uid = (new Date()).getMilliseconds();
		private.clients[uid] = {status: 100};

		factory.create(function (err, client) {
			client.uid = uid;
			client.status = 201;

			private.clients[uid] = client;
			if (!err) {
				public.release(client);
			} else {
				delete private.clients[uid];
			}

			cb(err, client);
		});
	}

	public.release = function (client) {
		client.status = 200;
	}

	private.manager = function (cb) {
		if (private.draining) {
			return setImmediate(cb, {msg: 'pool is draining'});
		}

		async.whilst(
			function () {
				var max = Math.min(factory.max, private.queue.length);
				//var min =
				return ((private.queue.length || factory.min) && (Object.keys(private.clients).length < (factory.min || Math.min(factory.max, private.queue.length))))
			},
			function (cb) {
				private.createResource(cb);
			},
			function (err) {
				for (var uid in private.clients) {
					if (private.queue.length) {
						if (private.clients[uid].status != 200) continue;

						var callback = private.queue.pop();
						private.clients[uid].status = 202;
						setImmediate(function (client, callback) {
							private.validate(client, function (err) {
								callback(err, client);
							})
						}, private.clients[uid], callback);
					} else {
						break;
					}
				}
				setImmediate(cb, err);
			}
		);
	}

	public.acquire = function (cb) {
		if (private.draining) return setImmediate(cb, {msg: 'pool is draining'});

		private.queue.unshift(cb)
	}

	public.destroy = function (client, cb) {
		if (client.status == 200) {
			client.status = 410;
			factory.destroy(client, function () {
				delete private.clients[client.uid];
				setImmediate(cb, null);
			});
		} else {
			setImmediate(cb, {msg: 'client is busy'})
		}
	}

	public.drain = function (cb) {
		private.draining = true;

		async.whilst(
			function () {
				return (Object.keys(private.clients).length > 0)
			},
			function (cb) {
				var uid = Object.keys(private.clients)[0];

				public.destroy(private.clients[uid], function () {
					cb()
				});
			},
			function (err) {
				cb(err)
			}
		);
	}

	function infinite() {
		private.manager(function () {
			setTimeout(infinite, 100);
		})
	}

	infinite();

	return public;
}