<commit_before># Mutator I/O

Mutator I/O is a tiny library to handle data transformations. It uses [RxJS](https://github.com/Reactive-Extensions/RxJS) to compose streams of data from a source (inputStream) to a destination (outputStream)

## Pipes
The concept of a "pipe" in Mutator I/O is simply an inputStream piped into an outputStream

![Input output](../master/doc/assets/input-output.png?raw=true)

```typescript
import {
  MutatorIO,
  inputStreams,
  outputStreams
} from 'mutator-io'

const myPipe = {
  name: 'myPipeName',
  in: new inputStreams.Mqtt({
    url: 'mqtt://localhost:1883',
    topics: ['mytopic']
  }),
  out: new outputStreams.DynamoDB(),
}

const mutator = new MutatorIO([myPipe])
mutator.start()
```

## Default Input/Output streams
  - ### Input
    - MQTT
  - ### Output
    - DynamoDB
    - Pass-through

## Adding transformations
Once you create a pipe, you have the possiblity to append data transformations to manipulate / aggregate / mutate the incoming data as you please (before it gets to the outputStream in the pipe)

![Input transform output](../master/doc/assets/input-transform-output.png?raw=true)

```typescript
// Transform can be a simple function
// or a function returning an Observable or a Promise
mutator.transform('myPipeName', (msg) => msg + 'something')
mutator.transform('myPipeName', (msg) => Rx.Observable.from([msg + 'something']))
mutator.transform('myPipeName', (msg) => Promise.resolve(msg + 'something else delayed'))

// Transformers can implement Typescript interfaces provided by output Streams
// E.G. this transform operation transforms the incoming payload from myPipeName
// into a DynamoDB delete query, treating payload as the key of the item to delete
mutator.transform('myPipeName', (msg): outputStreams.Dynamo.DynamoDBMessage => {
  operation: outputStreams.DynamoDB.Operations.DELETE,
    params: {
      TableName: 'processor_test',
      Key: {
        id: msg.payload,
      }
  }
})
```

### TODO
  - ~~Add documentation on the existing pre-defined streams (Mqtt, passThrough, DynamoDB)~~
  - ~~Add documentation on how to construct an inputStream~~
  - ~~Add documentation on how to construct an outputStream~~
  - Create further documentation leveraging typescript
  - Add better error handling
  - ~~Add tests coverage~~
<commit_msg>Adding links to README :link:<commit_after># Mutator I/O

Mutator I/O is a tiny library to handle data [transformations](doc/transform-streams/transform-streams.md). It uses [RxJS](https://github.com/Reactive-Extensions/RxJS) to compose streams of data from a source (inputStream) to a destination (outputStream)

## Pipes
The concept of a "pipe" in Mutator I/O is simply an [input stream](doc/input-streams/input-streams.md) piped into an [output stream](doc/output-streams/output-streams.md)

![Input output](../master/doc/assets/input-output.png?raw=true)

```typescript
import {
  MutatorIO,
  inputStreams,
  outputStreams
} from 'mutator-io'

const myPipe = {
  name: 'myPipeName',
  in: new inputStreams.Mqtt({
    url: 'mqtt://localhost:1883',
    topics: ['mytopic']
  }),
  out: new outputStreams.DynamoDB(),
}

const mutator = new MutatorIO([myPipe])
mutator.start()
```

## Default Input/Transform/Output streams
  - ### Input
    - [MQTT](doc/input-streams/mqtt.md)
  - ### Transform
    - [Pass-through](doc/transform-streams/pass-through.md)
  - ### Output
    - [DynamoDB](doc/output-streams/dynamodb.md)
    - [Pass-through](doc/output-streams/pass-through.md)

## Adding transformations
Once you create a pipe, you have the possiblity to append data transformations to manipulate / aggregate / mutate the incoming data as you please (before it gets to the outputStream in the pipe)

![Input transform output](../master/doc/assets/input-transform-output.png?raw=true)

```typescript
// Transform can be a simple function
// or a function returning an Observable or a Promise
mutator.transform('myPipeName', (msg) => msg + 'something')
mutator.transform('myPipeName', (msg) => Rx.Observable.from([msg + 'something']))
mutator.transform('myPipeName', (msg) => Promise.resolve(msg + 'something else delayed'))

// Transformers can implement Typescript interfaces provided by output Streams
// E.G. this transform operation transforms the incoming payload from myPipeName
// into a DynamoDB delete query, treating payload as the key of the item to delete
mutator.transform('myPipeName', (msg): outputStreams.Dynamo.DynamoDBMessage => {
  operation: outputStreams.DynamoDB.Operations.DELETE,
    params: {
      TableName: 'processor_test',
      Key: {
        id: msg.payload,
      }
  }
})
```

### TODO
  - ~~Add documentation on the existing pre-defined streams (Mqtt, passThrough, DynamoDB)~~
  - ~~Add documentation on how to construct an inputStream~~
  - ~~Add documentation on how to construct an outputStream~~
  - Create further documentation leveraging typescript
  - Add better error handling
  - ~~Add tests coverage~~

{'fraction_non_alphanumeric': 0.12008865605480556, 'fraction_numerical': 0.001611928269192021, 'mean_word_length': 4.208814270724029, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5542485', 'n_tokens_mistral': 1528, 'n_tokens_neox': 1502, 'n_words': 474}