A stream pipeline based message processing framework.
npm install billow
'use strict'
const { Billow, Flow, Droplet } = require('billow')
let billow = new Billow({ separator: '\r\n' })
let flowOne = new Flow({ events: ['error', 'dropletError'] })
let flowTwo = new Flow({ events: ['error', 'dropletError'] })
flowOne.on('error', console.error).on('dropletError', console.error)
flowTwo.on('error', console.error).on('dropletError', console.error)
flowOne.addDroplets([
new Droplet({
handler: async function (chunk, encoding) {
return await Promise.resolve(`${chunk.toString()}==`)
}
}),
new Droplet({
handler: function (chunk, encoding) {
console.log(chunk.toString())
}
})
])
flowTwo.addDroplets([
new Droplet({
handler: async function (chunk, encoding) {
return await Promise.resolve(`${chunk.toString()}~~`)
}
}),
new Droplet({
handler: function (chunk, encoding) {
console.log(chunk.toString())
}
})
])
billow.addFlow(flowOne).addFlow(flowTwo).write('billow!\r\nbillow!\r\n')
// billow!==
// billow!~~
// billow!==
// billow!~~
Billow
's workflow is based on Flow
which is the combination of Droplet
s. Each Droplet
is an implementation of Node.js transform stream, in which you can use async / await
function to handle the asynchronous data processing logic with happiness. And Flow
is the abstact pipeline of those Droplet
s, so the data transmission in Flow
is very fast and heap memory saved. Because Node.js transform stream will buffer the data when there is no downstream, so all data in Flow
s will be piped to a writable stream which name is blackHole
eventually to prevent the potential resident memory leak.
| => flow one => |
| ( droplet(1) => droplet(2) => ... ) |
| |
data => | => flow two => | => blackHole
| ( droplet(4) => droplet(3) => ... ) |
| |
| => other flows => |
- separator
Number
: The separator to separate the in coming buffer, if separator is set to benull
, thenbillow
will pass the buffer directly to theflows
, by default it is'\r\n'
. - highWaterMark
Number
: If provided, it will be passed to the insidestream.Writable
contructor.
return a new instance of Billow
.
- flow
Flow
: An instance offlow
.
Register a new flow
to billow
.
- events
Array<String>
: The events you want to listen emitted from thedroplet
s of thisflow
.
return a new instance of Flow
.
- droplets
Droplet | Array<Droplet>
: Thedroplet
s you want to add to theflow
.
Register new droplets
to flow
.
- handler
async function (chunk, encoding) {}
: The handler to process thechunk
, the value this function returns will be pass to the nextdroplet
in currentflow
. - highWaterMark
Number
: If provided, it will be passed to the insidestream.Transform
contructor.
return a new instance of Droplet
.