labeledpipe
![Gitter](https://badges.gitter.im/factset/labeledpipe.svg)
Lazypipe with optional labels.
Like lazypipe, labeledpipe creates an immutable lazily initialized pipeline.
Unlike lazypipe it allows pipeline stages to be labeled. You can then use those
labels to control where pipeline stages are added, and even remove previously
added stages.
Table of Contents
Installation
To install labeledpipe
for use in your project please run the following command:
yarn add labeledpipe
If you are using the npm
package manager:
npm install labeledpipe
Usage
Basic
In this example we use labeledpipe exactly as we would use lazypipe.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe(reportStage, 'A')
.pipe(reportStage, 'B')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
B: Some data
Labeling stages
To label a stage of the pipeline, we just prefix arguments to .pipe
with the
stages label:
var pipeline = labeledpipe()
.pipe('stage-label', reportStage, 'A')
;
We can now use the stage labels to change the point at which .pipe
inserts the
next pipeline stage. labeledpipe refers to this point as the cursor.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('stage-B', reportStage, 'B')
.before('stage-B')
.pipe('stage-C', reportStage, 'C')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
C: Some data
B: Some data
Cursor Movement
In addition to .before
, labeledpipe allows several other cursor positioning
commands. Here is a complete list:
.before(label)
.after(label)
.first()
.last()
.beginningOf(label)
.endOf(label)
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('stage-B', reportStage, 'B')
.pipe('stage-C', reportStage, 'C')
.before('stage-B')
.pipe(reportStage, 'A/B')
.after('stage-B')
.pipe(reportStage, 'B/C')
.first()
.pipe(reportStage, 'Start')
.last()
.pipe(reportStage, 'Finish')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
Start: Some data
A: Some data
A/B: Some data
B: Some data
B/C: Some data
C: Some data
Finish: Some data
Removing Stages
Labeledpipe also lets you remove labeled stages that were previously added to
the pipeline. This allows you to use most of a pipeline created by another
project
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('stage-B', reportStage, 'B')
.pipe('stage-C', reportStage, 'C')
.remove('stage-B')
.pipe(reportStage, 'D')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
C: Some data
D: Some data
Nested Pipelines
Like lazypipe, labeledpipe also lets you nest pipelines. This allows common
pipeline to be written once and used in multiple pipelines.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var common = labeledpipe()
.pipe(reportStage, 'A')
.pipe(reportStage, 'B')
.pipe(reportStage, 'C')
;
var pipeline = labeledpipe()
.pipe(common)
.pipe(reportStage, 'D')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
B: Some data
C: Some data
D: Some data
Unlike lazypipe however, you can also label the common pipelines and use the
cursor positioning commands to position relative to both the nested pipeline
itself, and the stages in the nested pipeline:
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var common = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('stage-B', reportStage, 'B')
.pipe('stage-C', reportStage, 'C')
;
var pipeline = labeledpipe()
.pipe('common-stage', common)
.before('common-stage')
.pipe(reportStage, 'before-common')
.beginningOf('common-stage')
.pipe(reportStage, 'beginning-of-common')
.endOf('common-stage')
.pipe(reportStage, 'end-of-common')
.after('common-stage')
.pipe(reportStage, 'after-common')
.after('stage-B')
.pipe(reportStage, 'inside-common')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
before-common: Some data
beginning-of-common: Some data
A: Some data
B: Some data
inside-common: Some data
C: Some data
end-of-common: Some data
after-common: Some data
Pseudo Stages
Labeledpipe lets you add labeled stages that do not contain a transform
stream. These are useful if you need to provide well know extension points.
Theses pseudo stages act like nested pipelines. As such you can use the
nested pipeline movement operators:
.beginningOf(label)
.endOf(label)
These operators move the cursor to just after the beginning or just before the
end of the pseudo stage.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('extend-here')
.pipe('stage-B', reportStage, 'B')
.endOf('extend-here')
.pipe('stage-Y', reportStage, 'Y')
.beginningOf('extend-here')
.pipe('stage-X', reportStage, 'X')
.endOf('extend-here')
.pipe('stage-Z', reportStage, 'Z')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
X: Some data
Y: Some data
Z: Some data
B: Some data
Mixing with lazypipe
Labeledpipe is designed to work seamlessly with lazypipe. Lazypipes can be used
as stages in a labeledpipe:
var labeledpipe = require('labeledpipe');
var lazypipe = require('lazypipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var lazyPipeline = lazypipe()
.pipe(reportStage, 'A')
.pipe(reportStage, 'B')
.pipe(reportStage, 'C')
;
var pipeline = labeledpipe()
.pipe('lazy', lazyPipeline)
.before('lazy')
.pipe(reportStage, 'before-lazy')
.after('lazy')
.pipe(reportStage, 'after-lazy')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
before-lazy: Some data
A: Some data
B: Some data
C: Some data
after-lazy: Some data
Similarly, labeledpipes can be used a stages in a lazypipe:
var labeledpipe = require('labeledpipe');
var lazypipe = require('lazypipe');
var through = require('through2');
function reportStage (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
done(null, obj);
});
}
var labeledPipeline = labeledpipe()
.pipe('stage-A', reportStage, 'A')
.pipe('stage-B', reportStage, 'B')
.pipe('stage-C', reportStage, 'C')
;
var pipeline = lazypipe()
.pipe(reportStage, 'before-labeled')
.pipe(labeledPipeline)
.pipe(reportStage, 'after-labeled')
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
before-labeled: Some data
A: Some data
B: Some data
C: Some data
after-labeled: Some data
Events
The labeledpipe object exports all of the chainable event emitter methods:
- addListener
- on
- once
- removeListener
- removeAllListeners
- setMaxListeners
The allows events handlers to be added to a pipeline as if the pipeline was
being constructed immediately.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function emitEvent (name) {
return through.obj(function (obj, enc, done) {
console.log(name + ':', obj);
this.emit(name, name);
done(null, obj);
});
}
var pipeline = labeledpipe()
.pipe(emitEvent, 'A')
.on('A', console.log.bind(console, 'Event:'))
.pipe(emitEvent, 'B')
.on('B', console.log.bind(console, 'Event:'))
;
var stream = pipeline();
stream.write('Some data');
stream.end();
Output:
A: Some data
Event: A
B: Some data
Event: B
Errors
Error events are a special case in node already. This treatment continues in
labeledpipe. By default, error events on pipeline's streams "bubble up" and are
re-emitted on the pipeline stream. For example:
var labeledpipe = require('labeledpipe');
var through = require('through2');
function returnError (name) {
return through.obj(function (obj, enc, done) {
done(new Error(name));
});
}
var stream = labeledpipe()
.pipe(returnError, 'A')
()
;
stream.on('error', function (error) {
console.log('Stream Hander:', error.message);
});
stream.write('my data');
stream.end();
Output:
Stream Hander: A
However, if you add an error handler to a pipeline stage, error event from that stage will no longer bubble up to the
stream.
var labeledpipe = require('labeledpipe');
var through = require('through2');
function returnError (name) {
return through.obj(function (obj, enc, done) {
done(new Error(name));
});
}
var stream = labeledpipe()
.pipe(returnError, 'A')
.on('error', function (error) {
console.log('Pipeline Hander:', error.message);
})
()
;
stream.on('error', function (error) {
console.log('Stream Hander:', error.message);
});
stream.write('my data');
stream.end();
Output:
Pipeline Hander: A
The same rules apply to sub pipelines
var labeledpipe = require('labeledpipe');
var through = require('through2');
function returnError (name) {
return through.obj(function (obj, enc, done) {
done(new Error(name));
});
}
var stream = labeledpipe()
.pipe(labeledpipe()
.pipe(returnError, 'A')
.on('error', function (error) {
console.log('Sub-pipeline Hander:', error.message);
this.push('More data');
})
.pipe(returnError, 'B')
)
.on('error', function (error) {
console.log('Pipeline Hander:', error.message);
this.push('More data');
})
.pipe(returnError, 'C')
()
;
stream.on('error', function (error) {
console.log('Stream Hander:', error.message);
});
stream.write('my data');
stream.end();
Output:
Sub-pipeline Hander: A
Pipeline Hander: B
Stream Hander: C
Debugging
To assist users of labeledpipe
with debugging the behavior of this module we use the debug utility package to print information to the console. To enable debug message printing, the environment variable DEBUG
, which is the variable used by the debug
package, must be set to a value configured by the package containing the debug messages to be printed.
To print debug messages on a unix system set the environment variable DEBUG
with the name of this package prior to executing a tool that invokes this module:
DEBUG=labeledpipe [CONSUMING TOOL]
On the Windows command line you may do:
set DEBUG=labeledpipe
[CONSUMING TOOL]
Node Support Policy
We only support Long-Term Support versions of Node.
We specifically limit our support to LTS versions of Node, not because this package won't work on other versions, but because we have a limited amount of time, and supporting LTS offers the greatest return on that investment.
It's possible this package will work correctly on newer versions of Node. It may even be possible to use this package on older versions of Node, though that's more unlikely as we'll make every effort to take advantage of features available in the oldest LTS version we support.
As each Node LTS version reaches its end-of-life we will remove that version from the node
engines
property of our package's package.json
file. Removing a Node version is considered a breaking change and will entail the publishing of a new major version of this package. We will not accept any requests to support an end-of-life version of Node. Any merge requests or issues supporting an end-of-life version of Node will be closed.
We will accept code that allows this package to run on newer, non-LTS, versions of Node. Furthermore, we will attempt to ensure our own changes work on the latest version of Node. To help in that commitment, our continuous integration setup runs against all LTS versions of Node in addition the most recent Node release; called current.
JavaScript package managers should allow you to install this package with any version of Node, with, at most, a warning if your version of Node does not fall within the range specified by our node
engines
property. If you encounter issues installing this package, please report the issue to your package manager.
Contributing
Please read our contributing guide to see how you may contribute to this project.
Copyright
Copyright 2018 FactSet Research Systems Inc
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.