# @xfx/utils

> ## Concept

Latest version **1.0.3** (published 2021-03-05) · MIT license · 0 weekly downloads

## Install

```sh
npm install @xfx/utils
pnpm add @xfx/utils
yarn add @xfx/utils
bun add @xfx/utils
```

## Health

**Score 15/100 (F)** — status: abandoned.

Positive: no vulnerabilities.

Warnings: low downloads; no types; no esm support.

Negative: abandoned; low maintenance score.

## Facts

| | |
|---|---|
| Version | 1.0.3 |
| Published | 2021-03-05 |
| First published | 2021-03-03 |
| Weekly downloads | 0 |
| License | MIT |
| TypeScript types | none |
| Module format | CommonJS |
| Dependencies | 3 |
| Unpacked size | 101.6 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| GitHub stars | 4 |
| Author | im4LF |
| Maintainers | andrey_vorobiev, him4lf |

## Links

- npm: https://www.npmjs.com/package/@xfx/utils
- Repository: https://github.com/im4LF/flox-utils
- Homepage: https://github.com/im4LF/flox-utils#readme
- Issues: https://github.com/im4LF/flox-utils/issues
- npm.io page: https://npm.io/package/@xfx/utils

## Dependencies (3)

- [pino](https://npm.io/package/pino.md) ^6.11.1
- [deepmerge](https://npm.io/package/deepmerge.md) ^4.2.2
- [pino-pretty](https://npm.io/package/pino-pretty.md) ^4.6.0

## Recent versions

- 1.0.3 (latest) — 2021-03-05
- 1.0.2 — 2021-03-05
- 1.0.1 — 2021-03-03
- 1.0.0 — 2021-03-03

## README

# Flows simple library

## Concept

`Flow` - flow consists of `processes` and `connections`

`Process` - process is an atomic operation containing `in ports` and `out ports`. Each process is based on `Node`

`Node` - defenition of behavior for processes. Node can have `props`, `input ports`, `out ports`

`Connection` - connections between ports of processes 

## Flow example

```bash
# main loop
# each process is defined as name_of_process(path/to/Node)
ticks(@xfx/time/Ticks) -> locker(@xfx/etc/Locks) -> tasks(@xfx/pg/ArrayQuery) -> dispatch(./Dummy)

# when task is started notify wait process about it
dispatch START -> wait(@xfx/etc/Wait)

# when task is done notify wait process about it
dispatch DONE -> DONE wait

# when wait process knows that all tasks are started, then it pushes message to ALLDONE port
wait ALLDONE -> NEXT tasks

# when done
tasks DONE -> UNLOCK locker
```

Each process can be initialized with args in flow definition, for example:

```bash
ticks(@xfx/time/Ticks, {"interval": 3000}) -> tasks(@xfx/pg/ArrayQuery, {"sql":"select * from some.table where p = $1", "values":[ 123 ]})
```

Or args can be defined outside 

```js
const data = {
    ticks: {
        interval: 5000
    },
    tasks: {
        dsn: dsn1,
        sql: `
            select 
                *
            from some.table
            where 
                p = $1
            limit $2
        `,
        values: message => [ 123, 12 ],
        shiftBy: 2,
        out: 'task'
    }
}
```

And passed to flow builder

```js

build(flow_def, data, (err, flow) => {

})

```

## Node examples

### Ticks node

props:
- `interval`

inports:
- `start`
- `stop`

outports:
- `out`

```js
const Ticks = {
    // props that must be initialized 
    props: {
        interval: { default: 1000 }
    },
    // in ports named with underscore at start
    _start(message, done) {
        let interval = typeof this.props.interval === 'function' ? this.props.interval(message) : this.props.interval
        this.send('out', { t: Date.now() }, done)
        this.__timer = setInterval(_ => this.send('out', { t: Date.now() }), interval)
    },
    _stop(message, done) {
        if (this.__timer) clearInterval(this.__timer)
        done()
    },
    // out ports named with underscore at end
    out_: 1
}
```

### Locks node

```js
const Locks = {
    _in(message, done) {
        if (this.locked) {
            this.logger.info({ locked: this.locked }, this.name + '_in')
            this.send('pass', message, done)
        }
        else {
            this.locked = true
            this.send('out', message, done)
        }
    },
    out_: 1,
    pass_: 1,
    _unlock(message, done) {
        this.logger.info(message, this.name + '_reset')
        this.locked = false
        done()
    }
}
```

### ArrayQuery node

props:
- `dsn` - connection string
- `sql` - string or function generator for sql
- `values` - array of function generator for values
- `shiftBy` - count of parallel shifting in out port
- `out` - datakey for each row pushed into message

inports:
- `in`
- `next`

outports:
- `out`
- `errors`
- `done`

```js
const ArrayQuery = {
    props: {
        dsn: { required: true },
        sql: [String,Function],
        values: [Array,Function],
        shiftBy: { type: Number, default: 1 },
        out: { default: 'item' }
    },
    out_: 1,
    errors_: 1,
    done_: 1,
    // each node can have init function
    init(done) {
        this.__pool = getPool(this.props.dsn)
        done(null, this)
    },
    _in(message, done) {
        
        let sql = typeof this.props.sql === 'function' ? this.props.sql(message) : this.props.sql
        let values = typeof this.props.values === 'function' ? this.props.values(message) : this.props.values

        this.logger.info({ dsn: this.props.dsn, sql, values }, this._name)

        this.__pool.query(sql, values, (err, res) => {
            if (err) {
                this.logger.error(err)
                message.error = err
                this.send('errors', message, done)
            }
            else {
                this.__items = res.rows
                this._next(message, done)
            }
        })
    },
    _next(message, done) {
        let buf = this.__items.splice(0, this.props.shiftBy)
        if (!buf.length) {
            this.send('done', message)
        }
        else {
            buf.forEach(item => {
                this.send('out', { ...message, [ this.props.out ]: item })
            })
        }
        done()
    }
}
```

## Experimental nodes definition

Define node as a function with some useful input arguments, for example:
- `env` - environment of flow, can contain `logger`

```js
const Locks = (args, env) => ({
    _in(message, done) {
        if (this.locked) {
            env.logger.info({ locked: this.locked }, this.name + '_in')
            this.send('pass', message, done)
        }
        else {
            this.locked = true
            this.send('out', message, done)
        }
    },
    out_: 1,
    pass_: 1,
    _unlock(message, done) {
        env.logger.info(message, this.name + '_reset')
        this.locked = false
        done()
    }
})

const Parallel = (args, env) => {

    let res = {
        _in(message, done) {
            // do parallel on N outs 
        }
    }

    for (let i = 0; i < args.parallel; i++) res[ 'out' + i + '_' ] = 1

    return res
}
```

## Gramma

```bash
# comment
process_name1(path/to/Node) OUT_PORT_NAME -> IN_PORT_NAME process_name2(path/to/Node, { "a": 1, "b": 2 })

process_name3(path/to/Node)

process_name2 -> process_name3
```

---
_Source: https://npm.io/package/@xfx/utils · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._