Initial commit

.gitignore

@@ -0,0 +1 @@
+node_modules/

package.json

@@ -0,0 +1,11 @@
+{
+  "name": "stage",
+  "version": "1.0.0",
+  "description": "Async collection",
+  "main": "stage.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "Avril",
+  "license": "ISC"
+}

stage.js

@@ -0,0 +1,97 @@
+function Stage(from) {
+	var base = this;
+	this.array = from || [];
+	this.over = false;
+	this.acceptInvalid = false;
+	var next = {
+	waiters: [],
+	wait: function() {
+			return new Promise(resolve => next.waiters.push(function() {  resolve(); }));
+		},
+		signal: function() {
+			if(next.waiters.length>0) {
+				next.waiters.shift()();
+			}
+		}, 
+		flush: function() {
+			while(next.waiters.length>0) {
+				next.waiters.shift()();
+			}
+		}
+	};
+
+	this.take = async function() {
+		if(base.over && base.array.length<=0)
+			return undefined;
+		
+		if(base.array.length<=0)
+			await next.wait();
+
+
+		if(base.over && base.array.length<=0)
+			return undefined;
+	
+		return base.array.shift();
+		
+	};
+
+	this.poll = function() {
+		return base.array;
+	};
+
+	this.takeNB = function() {
+		if(base.array.length<=0)
+			return null;
+		return base.array.shift();
+	};
+
+	this.swallow0 = async function() {
+		var ar = [];
+		var token;
+		while(token = await base.take()) {
+			ar.push(token);
+		}
+		return ar;
+	};
+
+	this.swallow = function(timeout) {
+		if(!timeout) return base.swallow0();
+		else {
+			return new Promise(function(_resolve,_reject) {
+				var running=true;
+				var resolve = function(v) {
+					if(running) _resolve(v);
+					running=false;
+				};
+				var reject=  function(v) {
+					if(running) _reject(v);
+					running=false;
+				};
+				base.swallow0().then(resolve);
+				if(timeout>0)
+					setTimeout(()=> reject(new Error("swallow timeout reached ("+timeout+")")), timeout);
+			});
+		}
+	};
+
+	this.giveMany= function(values) {
+		for(let value of values)
+			base.give(value);
+	};
+
+	this.give = function(value) {
+		if(!value && !base.acceptInvalid)
+			return;
+		base.array.push(value);
+		next.signal();
+	};
+
+	this.commit = function() {
+		base.over = true;
+		next.flush();
+
+		return base;
+	};
+};
+
+module.exports = Stage;