@ezs/core

core

Ce plugin propose une série d'instructions natives. Elles sont constamment disponibles car chargées automatiquement.

installation

npm install @ezs/core

détails

Plusieurs instructions permettent de créer des sous flux (sub pipeline), à partir d'un fichier d’instructions ou d'instructions imbriquées. Si elles s'utilisent toutes de la même manière (avec les mêmes paramètres) centaines peuvent apparaître comme similaires mais leur fonctionnement est différent :

• [delegate] : 1 sous flux pour tous les éléments
• [swing] : 1 sous flux pour tous les éléments filtrés selon une condition
• [spaw] : 1 sous flux par élément
• [loop] : 1 sous flux par élément
• [expand] : 1 sous flux pour N éléments (N = size), seul le champ sélectionné est envoyé dans le pipeline
• [combine] : 1 sous flux pour tous les éléments, seul le champ sélectionné est comparé avec le résultat du sous flux
• [singleton] : 1 sous flux pour le premier élément

usage

Table of Contents

• assign
• combine
• concat
• debug
• dedupe
• delegate
• dispatch
• dump
• env
• exchange
• expand
• extract
• fork
• group
• identify
• ignore
• keep
• loop
• map
• metrics
• overturn
• pack
• parallel
• pop
• remove
• replace
• shift
• shuffle
• singleton
• spawn
• swing
• throttle
• time
• tracer
• transit
• truncate
• ungroup
• unpack
• use
• validate

assign

Add a new field to an Object.

Input file:

[{
   "a": 1,
},
{
   "a": 2,
},
{
   "a": 3,
},
{
   "a": 4,
},
{
   "a": 5,
}]

Script:

[assign]
path = b.c
value = 'X'

Output:

[{
   "a": 1,
   "b": { "c": "X" },
},
{
   "a": 2,
   "b": { "c": "X" },
},
{
   "a": 3,
   "b": { "c": "X" },
},
{
   "a": 4,
   "b": { "c": "X" },
},
{
   "a": 5,
   "b": { "c": "X" },
}]

Parameters

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

Returns Object

combine

Takes an Object and substitute a field with the corresponding value found in a external pipeline the internal pipeline must produce a stream of special object (id, value)

[
          { year: 2000, dept: 54 },
          { year: 2001, dept: 55 },
          { year: 2003, dept: 54 },
]

Script:

[use]
plugin = analytics

[combine]
path = dept
file = ./departement.ini

Output:

 [
          { year: 2000, dept: { id: 54, value: 'Meurthe et moselle' } },
          { year: 2001, dept: { id: 55, value: 'Meuse' } },
          { year: 2003, dept: { id: 54, value: 'Meurthe et moselle' } },
 ]

Parameters

• path String? the path to substitute
• default String? value if no substitution (otherwise value stay unchanged)
• primer String Data to send to the external pipeline (optional, default n/a)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors
• cacheName String? Enable cache, with dedicated name

Returns Object

concat

Take all String, concat them and throw just one.

[
     "a",
     "b",
     "c"
]

Script:

[concat]
beginWith = <
joinWith = |
endWith = >

Output:

[
     "<a|b|c>"
]

Parameters

• 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 (with its number), and throw the same object.

with ezs debug enabled: every object will be stringify for printed and all others ezs debug traces will be print

with ezs debug disabled: every objects will be inspected (indented and colorized) and print on stderr (error level) or stdout (log level)

if ezs parameter is set, every object are not log (it's a global action)

Parameters

• level String console level : log or error or silent (optional, default error)
• text String text before the dump (optional, default valueOf)
• path String? path of field to print
• ezs Boolean? enable or disable ezs global debug traces

Returns Object

dedupe

Take Object, and check that the object identifier has not already been used previously

Parameters

• data
• feed
• path String path containing the object Identifier (optional, default uri)
• ignore Boolean Just ignore duplicate object (optional, default false)

Returns Object

delegate

Delegate processing to an external pipeline.

Note: works like spawn, but each chunk share the same external pipeline.

Parameters

• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

dispatch

Dispatch processing to an external pipeline on one or more servers.

Parameters

• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command

Returns Object

dump

Take all Objects and generate a JSON array

[
    { "a": 1 },
    { "a": 2 },
    { "a": 3 },
    { "a": 4 },
    { "a": 5 }
]

Script:

[dump]
indent = true

Output:

 [{
   "a": 1
  },
  {
   "a": 2
  },
  {
   "a": 3
  },
  {
   "a": 4
  },
  {
   "a": 5
  }
]

Parameters

• indent boolean indent JSON (optional, default false)

Returns String

env

Send the input object again, while adding new environment field(s) with the first Object of the feed.

Parameters

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

Returns Object

exchange

Take Object and throw a new item computed by the value= parameter (which replace the input one).

Input file:

[{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
},
{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
}]

Script:

[exchange]
value = omit('c')

Output:

[{
   "a": "abcdefg",
   "b": "1234567"
},
{
   "a": "abcdefg",
   "b": "1234567"
}]

Parameters

• value String? value to replace input object

Returns Object

expand

Takes an Object and substitute a field with the corresponding value found in a external pipeline the internal pipeline receive a special object { id, value } id is the item identifier & value is the item path value The internal pipeline can expand value with another

[
          { year: 2000, dept: 54 },
          { year: 2001, dept: 55 },
          { year: 2003, dept: 54 },
]

Script:

[use]
plugin = analytics

[expand]
path = dept
file = ./departement.ini

Output:

 [
          { year: 2000, dept: { id: 54, value: 'Meurthe et moselle' } },
          { year: 2001, dept: { id: 55, value: 'Meuse' } },
          { year: 2003, dept: { id: 54, value: 'Meurthe et moselle' } },
 ]

Parameters

• path String? the path to substitute
• size Number How many chunk for sending to the external pipeline (optional, default 1)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors
• cacheName String? Enable cache, with dedicated name
• token String? add token values in the subpipeline (optional)

Returns Object

extract

Take Object and throw each value of fields

Note: extract cannot throw undefined or null values

[{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
},
{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
}]

Script:

[extract]
path = a
path = b

Output:

[
   "abcdefg",
   "1234567",
   "abcdefg",
   "1234567"
}]

Parameters

• path String? path of field to extract

Returns Object

fork

fork the current pipeline

Note: but each chunk is sent to the same external pipeline.

Parameters

• standalone Boolean The current pipeline will be able to end without waiting for the end of the external pipeline (optional, default false)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

group

Take all chunks, and throw them grouped by length.

See also ungroup.

[
     "a",
     "b",
     "c",
     "d",
     "e",
     "f",
     "g",
     "h"
]

Script:

[group]
length = 3

Output:

[
     [ "a", "b", "c" ],
     [ "d", "e", "f" ],
     [ "g", "h" ]
]

Parameters

• length Number? Size of each partition

Returns String

identify

Take Object, and compute & add an identifier

Parameters

• data
• feed
• scheme String scheme to use (uid or sha) (optional, default uid)
• path String path containing the object Identifier (optional, default uri)

Returns String

ignore

Takes all the chunks, and ignore the firtst N chunk

Input file:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
},
{
   "a": 4
},
{
   "a": 5
}]

Script:

[ignore]
length = 3

Output:

[{
   "a": 4
},
{
   "a": 5
}]

Parameters

• length Number? Length of the feed to ignore

Returns any

keep

Throw input Object but keep only specific fields.

Input file:

[{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
},
{
   "a": "abcdefg",
   "b": "1234567",
   "c": "XXXXXXX"
}]

Script:

[keep]
path = a
path = b

Output:

[{
   "a": "abcdefg",
   "b": "1234567"
},
{
   "a": "abcdefg",
   "b": "1234567"
}]

Parameters

• path String? path of field to keep

Returns Object

loop

Loop on external pipeline, until test will be true

Note: works like delegate, but each chunk use its own external pipeline

Parameters

• test String? if test is true
• reverse Boolean to reverse the test (optional, default false)
• maxDepth Number to limit the number of loops (optional, default 100000)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in an object
• command String? the external pipeline is described in an URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

map

From an array field delegate processing of each items to an external pipeline

Note: works like delegate, but each chunk use its own external pipeline

Parameters

• path String? the path to substitute
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in an object
• command String? the external pipeline is described in an URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

metrics

• **See: ../server/knownPipeline.js **

Take Object, and throw the same object.

This statement will only be used if :

• EZS_METRICS is enabled
• ezs is running in server mode

WARNING: avoid setting bucket to "input" or "output", as these labels are used by ezs. If you do, you risk distorting the associated metrics.

Parameters

• pathName String to identify the script (optional, default auto)
• bucket String to identify the moment of measurement (optional, default unknow)

Returns Object

overturn

Takes an Object and substitute twice a field with the corresponding value found in a external pipeline the internal pipeline receive a special object { id, value, token } :

• id is the item identifier
• value is the item path value,
• token is an array containing stream id and an number (0 for first time, 1 for the second tme The internal pipeline can overturn value with another.

It's work like [expand] but the second call starts only when all the values of the stream have been sent once

[
          { year: 2000, dept: 'Meuse' },
          { year: 2001, dept: 'Moselle' },
          { year: 2003, dept: 'Vosges'},
]

Script #1:

[overturn]
path = dept

[overturn/assign]
path = value
value = get('value').split('').reverse().join('')

Output:

 [
          { year: 2000, dept: 'Meuse' },
          { year: 2001, dept: 'Moselle' },
          { year: 2003, dept: 'Vosges' },
 ]

Script #2:

[overturn]
path = dept

[overturn/drop]
  path = token.1
  if = 0

[overturn/assign]
path = value
value = get('value').split('').reverse().join('')

Output:

 [
          { year: 2000, dept: 'esueM' },
          { year: 2001, dept: 'ellesoM' },
          { year: 2003, dept: 'segsoV' },
 ]

Parameters

• path String? the path to overturn
• size Number How many chunk for sending to the external pipeline (optional, default 1)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command

Returns Object

pack

Take all Object, throw encoded String

Returns String

parallel

Takes an Object delegate processing to X internal pipelines

Parameters

• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

pop

• **See: shift **

Return the last Object and close the feed

Input file:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
},
{
   "a": 4
},
{
   "a": 5
}]

Script:

[shift]

Output:

[{
   "a": 5
}]

Returns Object

remove

Take Object and remove it from the feed if test is true Input file:

[{
   a: "a"
},
{
   a: 2
},
{
   a: "b"
},
{
   a: 4
},
{
   a: "c"
}]

Script:

[remove]
test = get('a).isInteger()
reverse = true

Output:

[
    {
       a: 2
    },
    {
       a: 4
    }
]

Parameters

• test String? if test is true
• reverse String reverse the test (optional, default false)

Returns Object

replace

Take Object and replace it with a new object with some fields.

See also exchange and assign.

Input file:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
},
{
   "a": 4
},
{
   "a": 5
}]

Script:

[replace]
path = b.c
value = 'X'

Output:

[{
   "b": { "c": "X" }
},
{
   "b": { "c": "X" }
},
{
   "b": { "c": "X" }
},
{
   "b": { "c": "X" }
},
{
   "b": { "c": "X" }
}]

Parameters

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

Returns Object

shift

Return the first Object and close the feed

Input file:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
},
{
   "a": 4
},
{
   "a": 5
}]

Script:

[shift]

Output:

[{
   "a": 1
}]

Returns Object

shuffle

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

Input file:

[{
   "a": "abcdefg",
   "b": "1234567"
},
{
   "a": "abcdefg",
   "b": "1234567"
}]

Script:

[shuffle]
path = a

Output:

[{
   "a": "cadbefg",
   "b": "1234567"
},
{
   "a": "dcaegbf",
   "b": "1234567"
}]

Parameters

• path String? path of field to shuffle

Returns Object

singleton

Takes only the first Object delegate processing to a external pipeline

Parameters

• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in a object
• command String? the external pipeline is described in a URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors

Returns Object

spawn

Delegate processing to an external pipeline, throw each chunk from the result.

Note: works like delegate, but each chunk use its own external pipeline

Parameters

• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in an object
• command String? the external pipeline is described in an URL-like command
• logger String? A dedicaded pipeline described in a file to trap or log errors
• cache String? Use a specific ezs statement to run commands (advanced)

Returns Object

swing

Delegate processing to an external pipeline under specifics conditions

Note: works like spawn, but each chunk shares the same external pipeline.

Parameters

• test String? if test is true
• reverse String reverse the test (optional, default false)
• file String? the external pipeline is described in a file
• script String? the external pipeline is described in a string of characters
• commands String? the external pipeline is described in an object
• command String? the external pipeline is described in an URL-like
• logger String? A dedicaded pipeline described in a file to trap or log errors command

Returns Object

throttle

Take Object and return the same object

[{
         { id: 'x', value: 2 },
         { id: 't', value: 2 },
}]

Script:

[use]
plugin = analytics

[throttle]
bySecond = 2

Output:

[
         { id: 'x', value: 2 },
         { id: 't', value: 2 },
]

Parameters

• bySecond Number Number of object by second (optional, default 1)

Returns Object

time

Measure the execution time of a script, on each chunk of input.

Parameters

• script string?

Examples

Input

[1]

Program

const script = `
[transit]
`;
from([1])
    .pipe(ezs('time', { script }))

Output

[{
  data: 1,
  time: 15 // milliseconds
}]

Returns object

tracer

Take Object, print a character and throw the same object. Useful to see the progress in the stream.

Parameters

• 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 again.

Input file:

[{
   "a": 1
},
{
   "a": 2
}]

Script:

[transit]

Output:

[{
   "a": 1
},
{
   "a": 2
}]

Returns Object

truncate

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

Input file:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
},
{
   "a": 4
},
{
   "a": 5
}]

Script:

[truncate]
length = 3

Output:

[{
   "a": 1
},
{
   "a": 2
},
{
   "a": 3
}]

Parameters

• length Number? Length of the feed

Returns any

ungroup

Take all chunks, and throw one item for every chunk.

See also group.

[
     [ "a", "b", "c" ],
     [ "d", "e", "f" ],
     [ "g", "h" ]
]

Script:

[ungroup]

Output:

[
     "a",
     "b",
     "c",
     "d",
     "e",
     "f",
     "g",
     "h"
]

Returns Array<any>

unpack

Take Strings or Buffers and throw Object builded by JSON.parse on each line.

Returns object

use

Take all String, concat them and throw just one.

Script:

[use]
plugin = basics
plugin = analytics

Parameters

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

Returns String

validate

From an Object, throw the same object if all rules pass

See

• Laravel validator rules
• https://github.com/skaterdav85/validatorjs#readme

Input file:

[{
   "a": 1,
   "b": "titi"
},
{
   "a": 2,
   "b": "toto"
},
{
   "a": false
},
]

Script:

[validate]
path = a
rule = required|number

path = a
rule = required|string

Output:

[{
   "a": 1,
   "b": "titi"
},
{
   "a": 2,
   "b": "toto"
}]

Parameters

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

Returns Object