node-plan

Execute a complicated dependency graph of tasks with smooth progress events.

Usage

Create tasks

Here are some examples:

• waveform
• callback
• s3-upload
• s3-download
• thumbnail
• transcode

Execute a plan

In this example, we will download a song from s3, generate a waveform image and preview audio, and upload each generated thing to s3, all the while reporting smooth and accurate processing progress to the end user.

1var Plan = require('plan')
2  , WaveformTask = require('plan-waveform')
3  , TranscodeTask = require('plan-transcode')
4  , UploadS3Task = require('plan-s3-upload')
5  , DownloadS3Task = require('plan-s3-download')
6
7var downloadTask = Plan.createTask(DownloadS3Task, "download", {
8  s3Key: '...',
9  s3Bucket: '...',
10  s3Secret: '...',
11})
12var waveformTask = Plan.createTask(WaveformTask, "waveform", {
13  width: 1000,
14  height: 200,
15});
16var previewTask = Plan.createTask(TranscodeTask, "preview", {
17  format: 'mp3'
18});
19var uploadWaveformTask = Plan.createTask(UploadS3Task, "upload-waveform", {
20  url: "/{uuid}/waveform{ext}"
21  s3Key: '...',
22  s3Bucket: '...',
23  s3Secret: '...',
24})
25var uploadPreviewTask = Plan.createTask(UploadS3Task, "upload-preview", {
26  s3Key: '...',
27  s3Bucket: '...',
28  s3Secret: '...',
29})
30
31// planId is used to index progress statistics. Same with the 2nd parameter
32// to `Plan.createTask` above. Next time we run the same planId, node-plan
33// uses the gathered stats to inform the progress events, so that they will
34// be much more accurate and smooth.
35var planId = "process-audio";
36
37var plan = new Plan(planId);
38plan.addTask(uploadPreviewTask);
39plan.addDependency(uploadPreviewTask, previewTask);
40plan.addDependency(previewTask, downloadTask);
41plan.addTask(uploadWaveformTask);
42plan.addDependency(uploadWaveformTask, waveformTask);
43plan.addDependency(waveformTask, downloadTask);
44plan.on('error', function(err, task) {
45  console.log("task", task.name, "error", err);
46});
47plan.on('progress', function(amountDone, amountTotal) {
48  console.log("progress", amountDone, amountTotal);
49});
50plan.on('update', function(task) {
51  console.log("update", task.exports);
52});
53plan.on('end', function(context) {
54  console.log("done", context);
55});
56var context = {
57  s3Url: '/the/file/to/download',
58  makeTemp: require('temp').path
59};
60plan.start(context);

What would I use this for?

Creating a service such as TransloadIt.

Mediablast is an open source service such as this running on node-plan.

Documentation

Creating a Task Definition

1var TaskDefinition = {};

start

1TaskDefinition.start = function(done) {
2  ...
3};

start is your main entry point. When you are finished processing, call done. In this scope, this points to the task instance.

If your task encounters an error, call done with the error object as the first parameter.

context

context acts as your input as well as your output. Sometimes it makes sense to delete the parameter that you are using; sometimes it does not.

Access context via this.context in the start function.

exports

exports is output that is tied to the task instance and is not passed to the next task in the dependency graph.

There are some special fields on exports that you should be wary of:

Fields you should write to:

• exports.amountTotal - as soon as you find out how long executing this task is going to take, set amountTotal. If the task is unable to emit progress, node-plan will guess based on previous statistics.
• exports.amountDone - this number will change based on progress whereas amountTotal should not.

Whenever you update amountDone or amountTotal, you should emit a progress event: this.emit('progress'). Don't worry about emitting an update event for this case.

Fields you should not write to:

• exports.startDate - the date the task instance started
• exports.endDate - the date the task instance completed
• exports.state - one of ['queued', 'skipped', 'processing', 'complete']
  • queued - this task has not yet been started
  • skipped - this task has been skipped, because one or more of its dependencies emitted an error, and ignoreDependencyErrors is not set to true.
  • processing - this task is currently in progress
  • complete - this task has completed, possibly unsuccessfully.

You are free to add as many other exports fields as you wish.

Whenever you change something in exports, you should emit a update event: this.emit('update').

Access exports via this.exports in the start function.

options

options are per-task-instance configuration values. It is the third parameter of Plan.createTask(definition, name, options).

Access options via this.options in the start function.

CPU Bound Tasks

If your task spawns a CPU-intensive process and waits for it to complete, you should mark your task as cpuBound:

1TaskDefinition.cpuBound = true;

node-plan pools all cpuBound tasks, defaulting to a worker count equal to the number of CPU cores on your machine.

Executing a Plan

new Plan(planId)

planId is used for the progress heuristics. when you're building a similar plan, use the same planId and progress events will use past statistics for accuracy and smoothness.

Plan.setWorkerCap(count)

Set this to limit the number of simultaneous CPU-bound tasks.

Plan.createTask(definition, name, options)

• definition - task definition described above
• name - a string, used to store statistics data. If you're doing a similar task, use the same name.
• options - an object which is passed to the task instance to configure it. In addition to the options which the task definition recognizes, all tasks have these additional built-in options:
  • ignoreDependencyErrors - if set to true, the task will execute even if one or more of its dependencies did not suceed. default false.

Plan.prototype.addTask(task)

This adds a root node to the dependency graph. If you imagine a tree, where the plan instance itself is the root node, addTask adds a task to the root node.

Plan.prototype.addDependency(targetTask, dependencyTask)

This is how you specify dependencies. dependencyTask becomes a leaf node of targetTask.

Plan.prototype.start(context)

Executes the plan. context is cloned and passed to all leaf nodes. Task instances modify context and pass a clone to the next task in the tree.

Plan 'end' event (context)

The plan has completed executing successfully. context is a merged result object of the tasks that were added with addTask.

Plan 'error' event (err)

One or more tasks returned an error. err is the error object.

Plan 'progress' event (amountDone, amountTotal)

Tells you how far along the execution of the plan is.

Plan 'update' event

Occurs when a task instance's exports have updated.