Initial commit

.gitignore

@@ -0,0 +1,2 @@
+node_modules
+yarn.lock

index.js

@@ -0,0 +1,26 @@
+"use strict";
+
+const propagateAbort = require("@promistream/propagate-abort");
+
+const createSequentialQueue = require("./sequential-queue");
+
+module.exports = function sequentialize() {
+	let withQueue = createSequentialQueue();
+
+	return {
+		_promistreamVersion: 0,
+		description: `sequentialize`,
+		// FIXME: We don't queue up aborts because once a downstream has encountered an error, it may have stopped trying to read, and we would deadlock. While the Aborted marker reads *do* get queued, the abort itself should probably be immediate. Need to make sure that this doesn't clash with any other part of the spec.
+		abort: propagateAbort,
+		peek: function peek(source) {
+			return withQueue(() => {
+				return source.peek();
+			});
+		},
+		read: function read(source) {
+			return withQueue(() => {
+				return source.read();
+			});
+		}
+	};
+};

package.json

@@ -0,0 +1,14 @@
+{
+  "name": "@promistream/sequentialize",
+  "version": "0.1.0",
+  "main": "index.js",
+  "repository": "http://git.cryto.net/promistream/sequentialize.git",
+  "author": "Sven Slootweg <admin@cryto.net>",
+  "license": "WTFPL OR CC0-1.0",
+  "dependencies": {
+    "@joepie91/unreachable": "^1.0.0",
+    "@promistream/propagate-abort": "^0.1.2",
+    "bluebird": "^3.5.4",
+    "p-defer": "^3.0.0"
+  }
+}

sequential-queue.js

@@ -0,0 +1,51 @@
+"use strict";
+
+// FIXME: Make this a stand-alone package; it should be useful as a general-purpose mechanism for sequentializing multiple asynchronous operations that may originate from multiple different callsites (like with read/peek/abort)
+
+const Promise = require("bluebird");
+const pDefer = require("p-defer");
+const unreachable = require("@joepie91/unreachable")("@promistream/sequentialize"); // FIXME: Change name when moved out into a stand-alone package
+
+module.exports = function createSequentialQueue() {
+	/* TODO: Does this need a more efficient FIFO queue implementation? */
+	let queue = [];
+	let processing = false;
+	
+	function nextItem() {
+		if (queue.length > 0) {
+			let item = queue.shift();
+			item();
+		} else {
+			unreachable("Tried to process an item from an empty queue");
+		}
+	}
+
+	function tryStart() {
+		if (processing === false) {
+			processing = true;
+			nextItem();
+		}
+	}
+
+	function markCompleted() {
+		if (queue.length > 0) {
+			nextItem();
+		} else {
+			processing = false;
+		}
+	}
+
+	return function withQueue(callback) {
+		let { resolve, promise } = pDefer();
+		queue.push(resolve);
+		tryStart();
+
+		return Promise.try(() => {
+			return promise;
+		}).then(() => {
+			return callback();
+		}).tap(() => {
+			markCompleted();
+		});
+	};
+}