ECMASCRIPT6 JavaScript multiply task module
- fixed Queue Module Continuous get() return undefined while has suspended put() task.
- bug fixed
- fixed
new Queue(size, fill, value)with uninitiated queue variable - Class Queue add
Queue.*push(value)method, so that Queue can act as a FIFO queue
- complete the test case and examples
- massive change to the api, make it more standard
- complete the API document
- complete Queue Class api document
$ npm install --save plug-task
TypeScript Code
import * as plugtask from 'plug-task';
function* subtask(id:number):plugtask.TaskIterator<number>
{
console.log('<subtask %d> started', id);
// use yield* to delegate a sub generator function
// will return the final result, and you can catch the exception from it
console.log('<subtask %d> start async_call()', id);
let result = yield* async_call(id);
console.log('<subtask %d> async_call() return: %s', id, result);
// use api sleep() to make current task sleep certain time
yield* plugtask.sleep(id * 1000);
console.log('<subtask %d> end', id);
return 100 + id;
}
// generator function can also use the plugtask api function
function* async_call(id:number):plugtask.TaskIterator<string>
{
yield* plugtask.sleep(id * 1000);
if (id % 2)
{
// task will exit when id is an odd number
yield* plugtask.exit(id, new Error('plugtask EXIT'));
}
return `call ${id} finish`;
}
// promise task for test the promise functionality
function* promise_task(mode:number):plugtask.TaskIterator<string>
{
yield* plugtask.sleep(1000);
if (mode % 2)
{
throw new Error(`promise exception ${mode}`);
}
else
{
return `promise value ${mode}`;
}
}
// main task generator function
function* main_task():plugtask.TaskIterator<void>
{
console.log('<main_task> start');
// create a sub task, you can name it a thread or a fiber
var task0 = plugtask.start(subtask(0));
// use api join() to wait the task instance end
// you can get the task instance return value from the result property
yield* plugtask.join(task0);
console.log('<main_task> task0 end with result:', task0.result);
// task instance can exit with the api exit(code, error?)
// join() will return the exit_code that provided by exit()
// task1 will exit at async_call() with and Error('exit with error')
// if task code throw an uncaught exception, task will exit with code -1 and
// the exception as the error
var task1 = plugtask.start(subtask(1));
let error_code = yield* plugtask.join(task1);
console.error(
`<main_task> task1 exited: %s [code: %d]`,
task1.error.toString(), error_code
);
console.log('---- join() api test end ----\n');
// api start() will return a Task class object, and it is a promise like object
// so you can use the promise method to handle the task return value or catch exception
var task2 = plugtask.start(promise_task(0));
task2.then((value) => {
console.log(`<main_task> task2.then() value: ${value}`);
});
var task3 = plugtask.start(promise_task(1));
task3.catch((err) => {
console.error('<main_task> task3.catch() got exception:', err);
});
// like other promise, callback function will run in asynchronous
console.log('---- main_task will not wait promise callback done ----');
yield* plugtask.sleep(1500);
console.log('---- promise like api test end ----\n');
// in task generator you can yield a promise
try {
// yield will wait and return the promise result
var result = yield new Promise<string>((resolve) => {
setTimeout(() => { resolve('promise resolved'); }, 1000);
});
console.log('<main_task> yield promise and get the result: %s', result);
// if the promise rejected, will throw a exception of the reject resaon
let error = yield Promise.reject('promise rejected');
console.log('<main_task> should not run this code');
}
catch (err)
{
// you can try/catch the promise exception error
console.error('<main_task> yield promise throw error: ', err);
}
console.log('---- yield promise test end ----\n');
// as you know task object is a promise like object, you absolutely
// can yield the task instance object and get task return or catch the error
// but the best way is use yield* to delegate the generator directly
try {
var value = yield plugtask.start(promise_task(2));
console.log(`<main_task> yield task return value: ${value}`);
}
catch (err)
{
// modiy the promise_task param value to even number to get throw an error
console.log(`<main_task> yield task throw an error: `, err);
}
console.log('---- yield Task object test end ----\n');
// sync multiple tasks return
var tasks = [
// task will return a normal result
plugtask.start(subtask(2)),
// task will exit with an error
plugtask.start(subtask(3)),
// task throw an exception and exit
plugtask.start(promise_task(3)),
];
// wait all tasks end and return the exit code array
let codes = yield* plugtask.join_all(tasks);
console.log('<main_task> tasks exit code is: ', codes);
console.log('<main_task> finish');
}
// handle the task uncaughtException, just hide it
plugtask.setExceptionHaldler(()=>{});
// start to run the first main task
plugtask.start(main_task());JavaScript Code
"use strict";
const plugtask = require('plug-task');
function* subtask(id) {
console.log('<subtask %d> started', id);
console.log('<subtask %d> start async_call()', id);
let result = yield* async_call(id);
console.log('<subtask %d> async_call() return: %s', id, result);
yield* plugtask.sleep(id * 1000);
console.log('<subtask %d> end', id);
return 100 + id;
}
function* async_call(id) {
yield* plugtask.sleep(id * 1000);
if (id % 2) {
yield* plugtask.exit(id, new Error('plugtask EXIT'));
}
return `call ${id} finish`;
}
function* promise_task(mode) {
yield* plugtask.sleep(1000);
if (mode % 2) {
throw new Error(`promise exception ${mode}`);
}
else {
return `promise value ${mode}`;
}
}
function* main_task() {
console.log('<main_task> start');
var task0 = plugtask.start(subtask(0));
yield* plugtask.join(task0);
console.log('<main_task> task0 end with result:', task0.result);
var task1 = plugtask.start(subtask(1));
let error_code = yield* plugtask.join(task1);
console.error(`<main_task> task1 exited: %s [code: %d]`, task1.error.toString(), error_code);
console.log('---- join() api test end ----\n');
var task2 = plugtask.start(promise_task(0));
task2.then((value) => {
console.log(`<main_task> task2.then() value: ${value}`);
});
var task3 = plugtask.start(promise_task(1));
task3.catch((err) => {
console.error('<main_task> task3.catch() got exception:', err);
});
console.log('---- main_task will not wait promise callback done ----');
yield* plugtask.sleep(1500);
console.log('---- promise like api test end ----\n');
try {
var result = yield new Promise((resolve) => {
setTimeout(() => { resolve('promise resolved'); }, 1000);
});
console.log('<main_task> yield promise and get the result: %s', result);
let error = yield Promise.reject('promise rejected');
console.log('<main_task> should not run this code');
}
catch (err) {
console.error('<main_task> yield promise throw error: ', err);
}
console.log('---- yield promise test end ----\n');
try {
var value = yield plugtask.start(promise_task(2));
console.log(`<main_task> yield task return value: ${value}`);
}
catch (err) {
console.log(`<main_task> yield task throw an error: `, err);
}
console.log('---- yield Task object test end ----\n');
var tasks = [
plugtask.start(subtask(2)),
plugtask.start(subtask(3)),
plugtask.start(promise_task(3)),
];
let codes = yield* plugtask.join_all(tasks);
console.log('<main_task> tasks exit code is: ', codes);
console.log('<main_task> finish');
}
plugtask.setExceptionHandler(() => { });
plugtask.start(main_task());<main_task> start
<subtask 0> started
<subtask 0> start async_call()
<subtask 0> async_call() return: call 0 finish
<subtask 0> end
<main_task> task0 end with result: 100
<subtask 1> started
<subtask 1> start async_call()
<main_task> task1 exited: Error: plugtask EXIT [code: 1]
---- join() api test end ----
---- main_task will not wait promise callback done ----
<main_task> task2.then() value: promise value 0
<main_task> task3.catch() got exception: [Error: promise exception 1]
---- promise like api test end ----
<main_task> yield promise and get the result: promise resolved
<main_task> yield promise throw error: promise rejected
---- yield promise test end ----
<main_task> yield task return value: promise value 2
---- yield Task object test end ----
<subtask 2> started
<subtask 2> start async_call()
<subtask 3> started
<subtask 3> start async_call()
<subtask 2> async_call() return: call 2 finish
<subtask 2> end
<main_task> tasks exit code is: [ 0, 3, -1 ]
<main_task> finish
plug-task support yield anything you want, but only handle Promise object,
plug-task will suspend current task and wait for the resolve/reject of the Promise
object, then continue task with the value of the Promise object. otherwise will
return back the value what you yield.
when you yield something out, then the plug-task core will take control the
program running, plug-task can handle the async call, or switch the running task,
make sure every task will has chance to be execute.
use yield * to delegate another generator, and return the generator's return
value.
create and run a task
| name | type | comment |
|---|---|---|
| fn* | generator |
task code that control the task program doing |
| task_name | string |
task name string, can make debug message more readable |
| RETURN | Task |
a promise like Task Class object, can use it to do some task synchronization |
return current task id
| name | type | comment |
|---|---|---|
| RETURN | number |
current task id |
exit current task, and return a error_code and error object
| name | type | comment |
|---|---|---|
| code | number |
exit code set to task.code |
| error | any |
anything set to the task.error |
wait the task end, and return the task exit code
| name | type | comment |
|---|---|---|
| task | Task |
task instance to wait |
| RETURN | number |
task exit code |
wait all tasks end, and return an array of tasks exit code
| name | type | comment |
|---|---|---|
| tasks | Task[] |
array of tasks instance object |
| RETURN | number[] |
array of tasks exit code |
current task sleep certain time
| name | type | comment |
|---|---|---|
| time | number |
millisecond to sleep |
suspend current task instance, a suspend task can wakeup by calling
task.resume_success() or task.resume_error()
suspend current task instance, and return task.resume_success(value)
param value.
set task private data name as key with value, task private data is
independent in each task instance, you can has same name but different
value in different task. useful for store global data by task.
| name | type | comment |
|---|---|---|
| key | string |
name string to be set |
| value | any |
value to set to name as key |
get a value name key from task private data object, like thread variable
if key not found in current task instance, plug-task will try to find from
parent task of current task.
| name | type | comment |
|---|---|---|
| key | string |
name string to be get |
| default_value | any |
default value return if no key value found |
| RETURN | any |
value of key or default_value |
set plug-task core configuration, will return the last configuration object. if not supply the options param, will juet return current configuration object.
| name | type | comment |
|---|---|---|
| options | TaskConfig |
configuration object |
| RETURN | TaskConfig |
last/current configuration object |
TaskConfig
| property | type | comment |
|---|---|---|
| switch_time | number |
task max continue run time by second, 0 turn off task time switching |
set a callback handler function to process the uncaughtException of task code,
if non ExceptionHandler is set, plug-task core will print the error through
console.error() as default.
| name | type | comment |
|---|---|---|
| handler | function |
callback function with type function(error){} |
| RETURN | function |
return the last handler callback function |
Convert a fn(...args, success_callback(data), error_callback(err)) function
to a TaskIterator generator function.
| name | type | comment |
|---|---|---|
| fn | function |
function accept two callback function |
| RETURN | generator |
arguments as fn but without callbacks, return success callback data, and throw the error_callback error |
Convert a thunk function fn(...args, callback(error, data)) function to a
TaskIterator generator function.
| name | type | comment |
|---|---|---|
| fn | function |
thunk function |
| RETURN | generator |
arguments as fn but without callback, if callback with error will throw out, and return the callback's data |
Convert a custom callback function to a TaskIterator generator function.
use a custom callback2thunk function to convert the function callback
to a thunk callback standard.
| name | type | comment |
|---|---|---|
| fn | function |
function that use custom callback format function |
| callback2thunk | function |
first param accept a thunk_callback format function, rest params as the fn's callback function arguments |
| RETURN | generator |
arguments as fn but without callback, if callback with error will throw out, and return the callback's data |
NodeJS fs.write example
"use strict";
var fs = require('fs');
var plugtask = require('plug-task');
var fs_write = plugtask.custom(fs.write, function(thunk_cb, err, written, buffer){
// covert written and buffer into an array data
thunk_cb(err, [written, buffer]);
});
function* main(){
let fp = fs.openSync('test.txt', 'a');
try {
// destruct the return value
let [ written, buffer ] = yield* fs_write(fp, 'test string\n');
console.log('%d bytes written to file', written);
}
catch (err)
{
console.log('write file error');
}
}
plugtask.start(main());A synchronization Queue base on plug-task.
you can use Queue to sync task's produce data and consuming data simply
Create a value queue object.
| name | type | comment |
|---|---|---|
| size | number |
Queue size, how many item could hold in queue cache. 0 mean no limit |
| values | array |
init queue cache items with this array values |
| fill | number |
how many item fill into the cache at init |
| value | any |
value that fill into the cache list |
| RETURN | Queue |
Queue object |
get a queue item from queue's tail, if queue is empty, will suspend current task and wait
| name | type | comment |
|---|---|---|
| RETURN | any |
queue item |
put a item into queue head, if queue has size limit and queue is full, will suspend current task and wait
| name | type | comment |
|---|---|---|
| value | any |
item value need to put into queue |
| RETURN | number |
return current queue items count |
push a item into queue tail, if queue has size limit and queue is full, will suspend current task and wait
| name | type | comment |
|---|---|---|
| value | any |
item value need to push into queue |
| RETURN | number |
return current queue items count |
get a queue item like get, but if queue is empty, method will return null without wait
| name | type | comment |
|---|---|---|
| RETURN | any |
queue item or null mean no queue empty |
TypeScript Code
import * as plugtask from 'plug-task';
var queue = new plugtask.Queue<number>(2);
function* producer():plugtask.TaskIterator<void>
{
var id = 1;
while (true)
{
console.log('put [ %d ] into queue', id);
yield* queue.put(id++);
}
}
function* consumer(id:number):plugtask.TaskIterator<void>
{
var time = 3;
while (time --> 0)
{
let num = yield* queue.get();
console.log('queue<%d> get [ %d ] from queue', id, num);
yield* plugtask.sleep(1000);
}
}
// create one producer task, produce numbers as soon as possible
plugtask.start(producer(), 'producer');
// create two consumer task, get 1 number per second, get 3 number at most
plugtask.start(consumer(1), 'consumer_1');
plugtask.start(consumer(2), 'consumer_2');JavaScript Code
"use strict";
const plugtask = require('plug-task');
var queue = new plugtask.Queue(2);
function* producer() {
var id = 1;
while (true) {
console.log('put [ %d ] into queue', id);
yield* queue.put(id++);
}
}
function* consumer(id) {
var time = 3;
while (time-- > 0) {
let num = yield* queue.get();
console.log('queue<%d> get [ %d ] from queue', id, num);
yield* plugtask.sleep(1000);
}
}
plugtask.start(producer(), 'producer');
plugtask.start(consumer(1), 'consumer_1');
plugtask.start(consumer(2), 'consumer_2');producer task will put 2 number into queue, and queue will be fulled, producer will suspend, two consumer task start get number from queue, after a consumer get a number, producer will wakeup and put a new number.
put [ 1 ] into queue
put [ 2 ] into queue
put [ 3 ] into queue
queue<1> get [ 1 ] from queue
put [ 4 ] into queue
queue<2> get [ 2 ] from queue
put [ 5 ] into queue
queue<1> get [ 3 ] from queue
put [ 6 ] into queue
queue<2> get [ 4 ] from queue
put [ 7 ] into queue
queue<1> get [ 5 ] from queue
put [ 8 ] into queue
queue<2> get [ 6 ] from queue
put [ 9 ] into queue
TypeScript Code
import * as plugtask from 'plug-task';
// create a keeper queue with init item as a key
let keeper = new plugtask.Queue<string>(0, ['key']);
function* visitor(name:string):plugtask.TaskIterator<void>
{
for (let time=1; time<=3; time++)
{
// before access the resource, get the keeper's key first
let key = yield* keeper.get();
// do a try/finally to make sure we wouldn't lose the key
try {
console.log('%s accessing the resource %d time', name, time);
// simulate do some async job with the resource
yield* plugtask.sleep(1000);
}
finally
{
// at last, return the key to the keeper
console.log('%s return the key %d time', name, time);
yield* keeper.put(key);
}
}
}
// create two visitor task
plugtask.start(visitor('jack'), 'visitor_jask');
plugtask.start(visitor('rose'), 'visitor_rose');JavaScript Code
"use strict";
const plugtask = require('plug-task');
let keeper = new plugtask.Queue(0, ['key']);
function* visitor(name) {
for (let time = 1; time <= 3; time++) {
let key = yield* keeper.get();
try {
console.log('%s accessing the resource %d time', name, time);
yield* plugtask.sleep(1000);
}
finally {
console.log('%s return the key %d time', name, time);
yield* keeper.put(key);
}
}
}
plugtask.start(visitor('jack'), 'visitor_jask');
plugtask.start(visitor('rose'), 'visitor_rose');jack and rose access the resource one by one
jack accessing the resource 1 time
jack return the key 1 time
rose accessing the resource 1 time
rose return the key 1 time
jack accessing the resource 2 time
jack return the key 2 time
rose accessing the resource 2 time
rose return the key 2 time
jack accessing the resource 3 time
jack return the key 3 time
rose accessing the resource 3 time
rose return the key 3 time