Make pipeline of streams easy : Easy Streams

It's just a wrapper to build Stream transformers with functional style. It's like the koa / expressjs middlewares !

Example

const ezs = require('ezs')

ezs.use(require('ezs-basics'));

process.stdin.resume();
process.stdin.setEncoding('utf8');
process.stdin
  .pipe(ezs('split', { delimiter: "\n" }))
  .pipe(ezs('counter'))
  .pipe(ezs((input, output) => output.send(input.toString()));
  .pipe(process.stdout);

Installation

With npm:

$ npm install ezs

Concepts

Scope

Each statement function have its own scope and can access to few methods :

• this.isLast()
• this.isFirst()
• this.getIndex()
• this.getParam(name, defaultValue)

Output Object

Output object is an object with few methods :

• output.write(something)
• output.end()
• output.send(something)
• output.close()
• output.stop(withAnError)

With a sync statement, you can break the pipe with throw but with an async statement, you should use stop(with An Error) instead of throw.

CLI

The command line interface is used to create run pipelines described in .ini files.

It is also possible to launch a web server allowing remote execution of .ini files

$ ezs -h                                                                                                                                                                   ven. 15 mars 2019 16:15:20 CET
Usage: ezs [options] [<file>|<directory>] [<file2> <file3> ...]

Options:
  --help         Affiche de l'aide                                     [booléen]
  --version      Affiche le numéro de version                          [booléen]
  --verbose, -v  Enable debug mode with DEBUG=ezs      [booléen] [défaut: false]
  --daemon, -d   Launch daemon on a directory containing commands script
                                                           [chaine de caractère]
  --server, -s   Server to dispach commands                [chaine de caractère]
  --env, -e      Execute commands with environement variables as input
                                                       [booléen] [défaut: false]

for more information, find our manual at https://github.com/touv/node-ezs

API Documentation

ezs(statement : Mixed, [params : Object]) : Stream

Converts a transform stream with existing function or adhoc function.

	const ezs = require('ezs'),
	let transformer = ezs(function(input, output) {
		output.send(input.toString())
	})

ezs.use(module: Function) : None

Adding bundle of statements. see the avaible modules here : https://www.npmjs.com/browse/keyword/ezs

	import ezs from 'ezs';
	import basics from 'ezs-basics';
	import files from 'ezs-files';

	ezs.use(basics);
	ezs.use(files);

ezs.catch(func : Function)

catch Error in NodeJS pipeline

    // Example #1
    process.stdin
        .pipe(ezs('truncate', { length: 100 }))
        .pipe(ezs((d, f) => f.send(new Error('Badaboum')))))
        .pipe(ezs.catch(e => e)) // catch errors in chunks and throw a error, which breaking the pipeline
        .on('error', console.error)

    // Example #2
    process.stdin
        .pipe(ezs('truncate', { length: 100 }))
        .pipe(ezs((d, f) => f.send(new Error('Badaboum')))))
        .pipe(ezs.catch(e => console.error('Warning:', e))) // catch errors in chunks to display them without breaking the pipeline

ezs.toBuffer(options : Object)

get chunk of in NodeJS pipeline and send Buffer of the chunk

    process.stdin
        .pipe(ezs('replace', { path: 'id', value: 'xxxxx' }))
        .pipe(ezs('dump'))
        .pipe(ezs.toBuffer())
        .pipe(process.stdout);

Statements

Table of Contents

• assign
  • Parameters
• concat
  • Parameters
• debug
  • Parameters
• delegate
  • Parameters
• dispatch
  • Parameters
• dump
  • Parameters
• env
  • Parameters
• extract
  • Parameters
• group
  • Parameters
• keep
  • Parameters
• pack
  • Parameters
• replace
  • Parameters
• shift
  • Parameters
• shuffle
  • Parameters
• tracer
  • Parameters
• transit
  • Parameters
• truncate
  • Parameters
• ungroup
  • Parameters
• unpack
  • Parameters
• validate
  • Parameters

assign

Take Object and add new field

Parameters

• data
• feed
• path String? path of the new field
• value String? value of the new field

Returns Object

concat

Take all String, concat them and thow just one

Parameters

• data
• feed
• beginWith String? Add value at the begin
• joinWith String? use value to join 2 chunk
• endWith String? Add value at the end

Returns String

debug

Take Object , print it and throw the same object

Parameters

• data
• feed
• level String console level : log or error (optional, default log)
• text String text before the dump (optional, default valueOf)
• path String? path of field to print

Returns Object

delegate

Takes an Object delegate processing to an external pipeline

Parameters

• data
• feed
• file String? the external pipeline is descrbied in a file
• script String? the external pipeline is descrbied in a sting of characters
• commands String? the external pipeline is descrbied in object

Returns Object

dispatch

Takes an Object dispatch processing to an external pipeline in one or more servers

Parameters

• data
• feed
• server String? servers to dispatch data
• file String? the external pipeline is descrbied in a file
• script String? the external pipeline is descrbied in a sting of characters
• commands String? the external pipeline is descrbied in object

Returns Object

dump

Take all Object and genereta a json array

Parameters

• data
• feed
• indent String indent JSON (optional, default false)

Returns String

env

Take Object and send the same object but in the meantime, it is possible to add new environment field

Parameters

• data
• feed
• path String? path of the new field
• value String? value of the new field

Returns Object

extract

Take Object and throw each value of fields

Parameters

• data
• feed
• path String? path of field to extract

Returns Object

group

Take all chunk, and throw array of chunks

Parameters

• data
• feed
• size Number? Size of each partition

Returns String

keep

Take Object and throw the same object but keep only spefici fields

Parameters

• data
• feed
• path String? path of field to keep

Returns Object

pack

Take all Object, throw encoded String

Parameters

• data
• feed

Returns String

replace

Take Object and create a new object with some fields

Parameters

• data
• feed
• path String? path of the new field
• value String? value of the new field

Returns Object

shift

Take the first Object and close the feed

Parameters

• data
• feed

Returns Object

shuffle

Take Object, shuffle data of the whole object or only some fields specified by path

Parameters

• data
• feed
• path String? path of field to shuffle

Returns Object

tracer

Take Object, print a character and throw the same object

Parameters

• data
• feed
• print String character to print at each object (optional, default .)
• last String character to print at last call (optional, default .)
• first String character to print at first call (optional, default .)

Returns Object

transit

Take Object and throw the same object

Parameters

• data
• feed

Returns Object

truncate

Takes all the chunks, and closes the feed when the total length is equal to the parameter

Parameters

• data
• feed
• length Number? Length of the feed

Returns Mixed

ungroup

Take all chunk, and throw each item of chunks

Parameters

• data
• feed

Returns String

unpack

Take String and throw Object builded by JSON.parse on each line

Parameters

• data
• feed

Returns String

validate

• See: laravel validtor rules

Take Object and throw the same object if all rules passes

Parameters

• data
• feed
• path String? path of the field
• rule String? rule to validate the field

Returns Object