Initial commit

Author: Balassa Márton

.gitignore

@@ -0,0 +1,2 @@
+node_modules
+typings

.vscode/launch.json

@@ -0,0 +1,57 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Launch",
+            "type": "node",
+            "request": "launch",
+            "program": "${workspaceRoot}/app.js",
+            "stopOnEntry": false,
+            "args": [],
+            "cwd": "${workspaceRoot}",
+            "preLaunchTask": null,
+            "runtimeExecutable": null,
+            "runtimeArgs": [
+                "--nolazy"
+            ],
+            "env": {
+                "NODE_ENV": "development"
+            },
+            "console": "internalConsole",
+            "sourceMaps": false,
+            "outDir": null
+        },
+        {
+            "name": "Attach",
+            "type": "node",
+            "request": "attach",
+            "port": 5858,
+            "address": "localhost",
+            "restart": false,
+            "sourceMaps": false,
+            "outDir": null,
+            "localRoot": "${workspaceRoot}",
+            "remoteRoot": null
+        },
+        {
+            "name": "Attach to Process",
+            "type": "node",
+            "request": "attach",
+            "processId": "${command.PickProcess}",
+            "port": 5858,
+            "sourceMaps": false,
+            "outDir": null
+        },
+        {
+            "name": "Run mocha",
+            "type": "node",
+            "request": "launch",
+            "program": "${workspaceRoot}/node_modules/mocha/bin/_mocha",
+            "stopOnEntry": false,
+            "args": ["test/**/*.js", "--no-timeouts"],
+            "cwd": "${workspaceRoot}",
+            "runtimeExecutable": null,
+            "env": { "NODE_ENV": "testing"}
+        }
+    ]
+}

.vscode/tasks.json

@@ -0,0 +1,10 @@
+{
+    // See https://go.microsoft.com/fwlink/?LinkId=733558
+    // for the documentation about the tasks.json format
+    "version": "0.1.0",
+    "command": "tsc",
+    "isShellCommand": true,
+    "args": ["-p", "."],
+    "showOutput": "silent",
+    "problemMatcher": "$tsc"
+}

package.json

@@ -0,0 +1,12 @@
+{
+  "name": "sequential-task-queue",
+  "version": "1.0.0",
+  "description": "Sequential task queue for node and the browser",
+  "author": {
+    "name": "BalassaMarton"
+  },
+  "devDependencies": {
+    "mocha": "^3.0.2",
+    "sinon": "^1.17.5"
+  }
+}

src/.gitignore

@@ -0,0 +1 @@
+*.js

src/sequential-task-queue.ts

@@ -0,0 +1,197 @@
+export interface ITaskQueueOptions {
+    name?: string;    
+    errorHandler?: (e: any) => any;
+    timeout?: number;
+}
+
+export interface ICancellationToken {
+    cancelled?: boolean;
+    reason?: string;
+}
+
+export var cancellationTokenReasons = {
+    timeout: "timeout",
+    cancel: "cancel"
+}
+
+export class TimeoutError extends Error {
+}
+
+export class SequentialTaskQueue {
+
+    private _errorHandler: (e: any) => void;
+    private _queue: ITaskEntry[] = [];
+    private _isClosed: boolean = false;
+    private _waiters: Function[] = [];
+    private _timeout: number;
+    private _currentTask: ITaskEntry;
+
+    static defaultTimeout = 0;
+
+    name: string;
+
+    get isClosed() {
+        return this._isClosed;
+    }
+
+    constructor(options?: ITaskQueueOptions) {
+        if (!options)
+            options = {};
+        if (options.timeout === undefined || options.timeout === null)
+            options.timeout = SequentialTaskQueue.defaultTimeout;
+        this._timeout = options.timeout;
+        this.name = options.name;
+        this._errorHandler = options.errorHandler;
+    }
+
+    push(task: Function, timeout?: number): ICancellationToken {
+        if (this._isClosed)
+            throw new Error(`${this.name || "SequentialTaskQueue"} has been previously closed`);
+        var entry: ITaskEntry = {
+            fn: task,
+            timeout: timeout === undefined ? this._timeout : timeout,
+            cancellationToken: new CancellationToken()
+        };
+        this._queue.push(entry);
+        this.schedule(() => this.next());
+        return entry.cancellationToken;
+    }
+
+    cancel(): PromiseLike<any> {
+        if (this._currentTask) {
+            if (this._currentTask.timeoutHandle)
+                clearTimeout(this._currentTask.timeoutHandle);
+            this._currentTask.cancellationToken.cancel(cancellationTokenReasons.cancel);
+        }
+        this._queue.splice(0);
+        return this.wait();
+    }
+
+    close(cancel?: boolean): PromiseLike<any> {
+        if (this._isClosed)
+            return this.wait();
+        this._isClosed = true;
+        if (cancel)
+            return this.cancel();
+        return this.wait();
+    }
+
+    wait(): PromiseLike<any> {
+        if (!this._currentTask && this._queue.length === 0)
+            return Promise.resolve();
+        return new Promise(resolve => {
+            this._waiters.push(resolve);
+        });
+    }
+
+    onError(handler: (e: any) => void) {
+        this._errorHandler = handler;
+    }
+
+    private next() {
+        if (!this._currentTask) {
+            let task = this._currentTask = this._queue.shift();
+            if (task) {
+                try {
+                    if (task.cancellationToken.cancelled) {
+                        this.schedule(() => this.next());
+                        return;
+                    }
+                    if (task.timeout) {
+                        task.timeoutHandle = setTimeout(() => {
+                            task.timeoutHandle = undefined;
+                            task.cancellationToken.cancel(cancellationTokenReasons.timeout);
+                            this.handleError(new TimeoutError());
+                            if (this._currentTask === task)
+                                this.doneTask();
+                        }, task.timeout);
+                    }
+                    let res = task.fn(task.cancellationToken);
+                    if (res && isPromise(res)) {
+                        res.then(() => {
+                                if (this._currentTask === task)
+                                    this.doneTask();
+                            },
+                            err => {
+                                this.handleError(err);
+                                if (this._currentTask === task)
+                                    this.doneTask();
+                            });
+                    } else
+                        this.doneTask();
+
+                } catch (e) {
+                    this.handleError(e);
+                    this.doneTask();
+                }
+            } else {
+                // queue is empty, call waiters
+                this.callWaiters(); 
+            }
+        }
+    }
+
+    private doneTask() {
+        if (this._currentTask.timeoutHandle)
+            clearTimeout(this._currentTask.timeoutHandle);
+        this._currentTask = undefined;
+        if (!this._queue.length)
+            this.callWaiters();
+        else
+            this.schedule(() => this.next());
+    }
+
+    private callWaiters() {
+        let waiters = this._waiters.splice(0);
+        waiters.forEach(waiter => waiter());
+    }
+
+    private schedule(fn: Function) {
+        setTimeout(fn, 0);
+    }
+
+    private handleError(e: any) {
+        try {
+            if (typeof this._errorHandler === "function")
+                this._errorHandler(e);
+        } catch (ie) {
+            // suppress errors thrown in error handler
+        }
+    }
+}
+
+class CancellationToken implements ICancellationToken {
+    cancelled: boolean = false;
+    reason: string = null;
+
+    cancel(reason: string) {
+        this.cancelled = true;
+        this.reason = reason;
+    }
+}
+
+interface ITaskEntry {
+    fn: Function;
+    timeout: number;
+    timeoutHandle?: any;
+    cancellationToken?: CancellationToken;
+}
+
+function isPromise(obj: any): obj is PromiseLike<any> {
+    return (obj && typeof obj.then === "function");
+}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

test/.gitignore

@@ -0,0 +1 @@
+*.js