"use strict";"use strict";A Task is a cancelable variant of a promise.
Like a promise, the observable is a proxy for the result of some work.
The interface is largely the same, but a observable can only have one
observer.
For example, calling then a second time will throw an error.
Instead, if a task has multiple observers, you can sacrifice cancelability
by coercing it to a promise, or use fork before observing it.
If every fork is cancelled, the cancelation will propagate back to the
original job.
The price of cancelability is a less robust system and more book keeping. A system that makes a great deal of use of tasks allows information to flow from any observable to any related task, even if distantly related. The cancelation of one task can propagate throughout an entire system of tasks, both forward and backward between consumers and producers. In exchange the system gains the ability to either free or avoid consuming resources proactively.
var asap = require("asap");
var WeakMap = require("weak-map");The consumer side of a task should receive the task’s observable. This object provides the ability to register exactly one observer for the result of the task, and the ability to cancel the task with an error.
function Task(setup, thisp) {
var deferred = Task.defer();
var handler = handlers.get(deferred.out);
handler.cancel = setup.call(thisp, deferred.in.return, deferred.in.throw);
return deferred.out;
}
/*
TODO Task.prototype = Object.create(Observable);
Such that it is possible to create parallel signaling for status and estimated
time to completion, or other arbitrary signals from the resolver to the
observable.
*/The done method registers an observer for any combination of completion or
failure with the given methods and optional context object.
The done method does not return a new task and does not capture errors
thrown by the observer methods.
Task.prototype.done = function (onreturn, onthrow, thisp) {
var self = this;
var handler = Task_getHandler(self);
handler.done(onreturn, onthrow, thisp);
};The then method registers an observer for any combination of completion or
failure, and creates a new task that will be completed with the result of
either the completion or failure handler.
Task.prototype.then = function (onreturn, onthrow, thisp) {TODO accept status and estimated time to completion arguments in arbitrary order.
var handler = Task_getHandler(this);
var task = Task.defer(this.cancel, this);
var _onreturn, _onthrow;
if (typeof onreturn === "function") {
_onreturn = function (value) {
try {
task.in.return(onreturn.call(thisp, value));
} catch (error) {
task.in.throw(error);
}
};
}
if (typeof onthrow === "function") {
_onthrow = function (error) {
try {
task.in.return(onthrow.call(thisp, error));
} catch (error) {
task.in.throw(error);
}
};
}
this.done(_onreturn, _onthrow);
return task.out;
};The spread method fills a temporary need to be able to spread an array
into the arguments of the completion handler of a then observer.
ECMAScript 6 introduces the ability to spread arguments into an array in the
signature of the method.
Task.prototype.spread = function (onreturn, onthrow, thisp) {
return this.then(function (args) {
return onreturn.apply(thisp, args);
}, onthrow, thisp);
};The catch method registers an error observer on a task and returns a new
task to be completed with the result of the observer.
The observer may return another task or thenable to transfer responsibility
to complete this task to another stage of the process.
Task.prototype.catch = function (onthrow, thisp) {
return this.then(null, onthrow, thisp);
};The finally method registers an observer for when the task either
completes or fails and returns a new task to perform some further work but
forward the original value or error otherwise.
Task.prototype.finally = function (onsettle, thisp) {
return this.then(function (value) {
return onsettle.call(thisp).then(function Task_finally_value() {
return value;
})
}, function (error) {
return onsettle.call(thisp).then(function Task_finally_error() {
throw error;
});
});
};The get method creates a task that will get a property of the completion
object for this task.
Task.prototype.get = function (key) {
return task.then(function (object) {
return object[key];
});
};The call method creates a task that will call the function that is the
completion value of this task with the given spread arguments.
Task.prototype.call = function (thisp /*, ...args*/) {
var args = [];
for (var index = 1; index < arguments.length; index++) {
args[index - 1] = arguments[index];
}
return task.then(function (callable) {
return callable.apply(thisp, args);
});
};The invoke method creates a task that will invoke a property of the
completion object for this task.
Task.prototype.invoke = function (name /*, ...args*/) {
var args = [];
for (var index = 1; index < arguments.length; index++) {
args[index - 1] = arguments[index];
}
return task.then(function (object) {
return object[name].apply(object, args);
});
};The thenReturn method registers an observer for the completion of this
task and returns a task that will be completed with the given value when
this task is completed.
Task.prototype.thenReturn = function (value) {
return this.then(function () {
return value;
});
};The thenReturn method registers an observer for the completion of this
task and returns a task that will fail with the given error when this task
is completed.
Task.prototype.thenThrow = function (error) {
return this.then(function () {
return error;
});
};Effects cancelation from the consumer side.
Task.prototype.throw = function (error) {
var handler = Task_getHandler(this);
if (handler.cancel) {
handler.throw(error);
}
};A task can only be observed once, but it can be forked.
The fork method returns a new task that will observe the same completion
or failure of this task.
Hereafter, this task and all forked tasks must all be cancelled for this
task’s canceller to propagate.
Task.prototype.fork = function () {The fork method works by fiddling with the handler of this task. First, we extract this task’s handler and make it the new parent for two child tasks.
var parentHandler = Task_getHandler(this);
parentHandler.done(function (value) {
left.in.return(value);
right.in.return(value);
}, function (error) {
left.in.throw(error);
right.in.throw(error);
});
/* TODO estimated time to completion forwarding */
/* TODO use a signal operator to propagate cancellation */
var leftCanceled = false, rightCanceled = false;
var left = Task.defer(function (error) {
if (leftCanceled) {
return;
}
leftCanceled = true;
if (rightCanceled) {
parentHandler.throw(error);
}
});
var right = Task.defer(function (error) {
if (rightCanceled) {
return;
}
rightCanceled = true;
if (leftCanceled) {
parentHandler.throw(error);
}
});We replace our own handler with the left child
handlers.set(this, Task_getHandler(left.out));And return the task with the right child handler
return right.out;
};The delay method of a task adds a delay of some miliseconds after the task
completes.
Cancelling the delayed task will cancel either the delay or the delayed
task.
Task.prototype.delay = function (ms) {
var self = this;
var task = Task.defer(function cancelDelayedTask() {
self.throw();
clearTimeout(handle);
});
var result = Task.defer();
var handle = setTimeout(function taskDelayed() {
task.in.return(result.out);
}, ms);
this.done(function (value) {
result.in.return(value);
}, function (error) {
task.in.throw(error);
});
return task.out;
};The timeout method will automatically cancel a task if it takes longer
than a given delay in miliseconds.
Task.prototype.timeout = function (ms, message) {
var self = this;
var task = Task.defer(function cancelTimeoutTask() {
this.throw();
clearTimeout(handle);
}, this);
var handle = setTimeout(function Task_timeout() {
self.throw();
task.in.throw(new Error(message || "Timed out after " + ms + "ms"));
}, ms);
this.done(function Task_timeoutValue(value) {
clearTimeout(handle);
task.in.return(value);
}, function Task_timeoutError(error) {
clearTimeout(handle);
task.in.throw(error);
});
return task.out;
};The producer side of a task should get a reference to a task’s resolver. The object provides the capability to settle the task with a completion value or a failure error.
function Completer(handler) {The task resolver implicitly binds its return and throw methods so these can be passed as free functions.
this.return = this.return.bind(this);
this.throw = this.throw.bind(this);
}The return method sets the tasks state to “fulfilled” (in the words of
promises) or “completed” (in the vernacular of tasks), with a given value.
If the corresponding observer was registered already, this will inform
the observer as soon as possible.
If the corresponding observer gets registered later, it will receive the
result as soon as possible thereafter.
Completer.prototype.return = function (value) {
var handler = Task_getHandler(this);
handler.become(Task.return(value));
};The throw method sets the tasks state to “rejected” (a term borrowed from
promises) or “failed” (the corresponding task jargon), with the given error.
Again, if the corresponding observer was registered already, this will
inform the observer as soon as possible.
If the corresponding observer gets registered later, it will receive the
result as soon as possible thereafter.
Completer.prototype.throw = function (error) {
var handler = Task_getHandler(this);
handler.become(Task.throw(error));
};The task constructor creates a resolver “in” and an observer “out” pair with some shared internal state. Particularly, since tasks can be canceled, the task constructor accepts a reference to the cancellation method and optionally the instance that hosts it.
module.exports = Task;
Task.defer = function (cancel, thisp) { // TODO estimate, label
var handler = new TaskHandler(); // TODO polymorph constructors
var input = Object.create(Completer.prototype);
var output = Object.create(Task.prototype);
Completer_bind(input);
handlers.set(input, handler);
handlers.set(output, handler);
handler.cancel = cancel;
handler.cancelThisp = thisp;
return {in: input, out: output};
}
function Completer_bind(completer) {
completer.return = completer.return.bind(completer);
completer.throw = completer.throw.bind(completer);
};The isTask utility method allows us to identify a task that was
constructed by this library.
This library does not attempt to make it provably impossible to trick the
Task.
Task.isTask = isTask;
function isTask(object) {
return (
Object(object) === object &&
!!handlers.get(object) &&
object instanceof Task
);
};The isThenable method is used internally to identify other singular
asynchronous duck types, including promises, which can be coerced into
tasks.
function isThenable(object) {
return Object(object) === object && typeof object.then === "function";
}The return function lifts a value into a task that has already completed
with a value.
Task.return = function (value) {
if (isTask(value)) {
return value;
} else if (isThenable(value)) {
/* TODO implement thenable coercion */
throw new Error("Thenables not yet implemented");
} else {
var handler = new TaskHandler();
handler.state = "fulfilled";
handler.value = value;
var task = Object.create(Task.prototype);
handlers.set(task, handler);
return task;
}
};The throw function lifts an error into a task that has already failed with
that error.
Task.throw = function (error) {
var handler = new TaskHandler();
handler.state = "rejected";
handler.error = error;
var task = Object.create(Task.prototype);
handlers.set(task, handler);
return task;
};The all function accepts an array of tasks, or values that can be coerced
into tasks, and produces a task that when completed will produce an array of
the individual completion values.
Task.all = function Task_all(tasks) {If the task is cancelled, or if any individual task fails, all of the outstanding individual tasks will be cancelled.
function cancelAll(error) {
for (var otherIndex = 0; otherIndex < tasks.length; otherIndex++) {Note that throwing an error upstream consitutes talking back to the producer. This is a reminder that tasks are a cooperation between a single consumer and a single producer and that information flows both ways and in fact allows information to propagate laterally by passing up one stream and down another.
tasks[otherIndex].throw(error);
}
result.in.throw(error);
}The number of outstanding tasks, tracked to determine when all tasks are completed.
var remaining = tasks.length;
var result = Task.defer(cancelAll);
var results = Array(tasks.length);
/* TODO estimated time to completion, label signals */
var estimates = [];
var estimate = -Infinity;
var setEstimate;
var estimates = tasks.map(function Task_all_each(task, index) {
task = tasks[index] = Task.return(task); // Coerce values to tasks
task.done(function Task_all_anyReturn(value) {
results[index] = value;
if (--remaining === 0) {
result.in.return(results);
}
}, cancelAll);
});
return result.out;
};The any method accepts an array of tasks, or value coercable to tasks, and
returns a task that will receive the value from the first task that
completes with a value.
After one succeeds, all remaining tasks will be cancelled.
If one of the tasks fails, it will be ignored.
If all tasks fail, this task will fail with the last error.
Task.any = function (tasks) {
/* TODO */
};The any method accepts an array of tasks, or value coercable to tasks, and
returns a task that will receive the value or error of the first task that
either completes or fails.
Afterward, all remaining tasks will be cancelled.
Task.race = function (tasks) {
/* TODO */
};The delay method accepts a duration of time in miliseconds and returns a
task that will complete with the given value after that amount of time has
elapsed.
Task.delay = function (ms, value) {
return Task.return(value).delay(ms);
};The resolver and observable side of a task share a hidden internal record with their shared state. Handlers are an alternative to using closures.
var handlers = new WeakMap();
function TaskHandler() {When a task is resolved, it “becomes” a different task and its
observable, if any, must be forwarded to the new task handler.
In the become method, we also adjust the “handlers” table so any
subsequent request for this handler jumps to the end of the “became”
chain.
this.became = null;Tasks may be created with a corresponding canceler.
this.cancel = null;
this.cancelThisp = null;Tasks may be “pending”, “fulfilled” with a value, or “rejected” with an error
this.state = "pending";
this.value = null;
this.error = null;A task may only be observed once. Any future attempt to observe a task will throw an error.
this.observed = false;Since a task can only be observed once, we only need to track one handler for fulfillment with a value or rejection with an error. A promise keeps an array of handlers to forward messages to. These handlers can be forgotten once a task settles since thereafter the observer would be informed immediately.
this.onreturn = null;
this.onthrow = null;The object to use as this in the context of onreturn and onthrow.
this.thisp = null;
}Since a task handler can become another task handler, this utility method will look up the end of the chain of “became” properties and rewrite the handler look up table so we never have to walk the same length of chain again.
function Task_getHandler(task) {
var handler = handlers.get(task);
while (handler && handler.became) {
handler = handler.became;
}
handlers.set(task, handler);
return handler;
}The done method is kernel for subscribing to a task observer.
If the task has already completed or failed, this will also arrange for the
observer to be notified as soon as possible.
TaskHandler.prototype.done = function (onreturn, onthrow, thisp) {
if (this.observed) {
throw new Error("Can't observe a task multiple times. Use fork");
}
this.observed = true;
this.onreturn = onreturn;
this.onthrow = onthrow;
this.thisp = thisp;If we are observing a task after it completed or failed, we dispatch the result immediately.
if (this.state !== "pending") {Instead of passing a callable closure, we pass ourself to avoid allocating another object. The task handler serves as a psuedo-function by implementing “call”.
asap(this);
}We handle the case of observing before completion or failure in the
become method.
};Above, we pass the task handler to asap.
The event dispatcher treats functions and callable objects alike.
This method will get called if this task has settled into a “fulfilled” or
“rejected” state so we can call the appropriate handler.
TaskHandler.prototype.call = function () {
if (this.state === "fulfilled") {
if (this.onreturn) {
this.onreturn.call(this.thisp, this.value);
}
} else if (this.state === "rejected") {
if (this.onthrow) {
this.onthrow.call(this.thisp, this.error);
} else {
throw this.error;
}
}We release the handlers so they can be potentially garbage collected.
this.onreturn = null;
this.onthrow = null;
this.thisp = null;
};The become method is the kernel of the task resolver.
TaskHandler.prototype.become = function (task) {
var handler = Task_getHandler(task);A task can only be resolved once. Subsequent resolutions are ignored. Ignoring, rather than throwing an error, greatly simplifies a great number of cases, like racing tasks and cancelling tasks, where handling an error would be unnecessary and inconvenient.
if (this.state !== "pending") {
return;
}The became property gets used by the internal handler getter to
rewrite the handler table and shorten chains.
this.became = handler;Once a task completes or fails, we no longer need to retain the canceler.
this.cancel = null;
this.cancelThisp = null;If an observer subscribed before it completed or failed, we forward the
resolution.
If an observer subscribes later, we take care of that case in done.
if (this.observed) {
handler.done(this.onreturn, this.onthrow, this.thisp);
}
};The throw method is used by the promise observer to cancel the task from
the consumer side.
TaskHandler.prototype.throw = function (error) {
if (this.cancel) {
this.cancel.call(this.cancelThisp);
}
this.become(Task.throw(error || new Error("Consumer canceled task")));
};