# paced-work-stream

> Node.js stream working at constant pace and concurrent

Latest version **1.0.3** (published 2018-11-27) · MIT license · 0 weekly downloads

## Install

```sh
npm install paced-work-stream
pnpm add paced-work-stream
yarn add paced-work-stream
bun add paced-work-stream
```

## 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 | 2018-11-27 |
| First published | 2016-09-11 |
| Weekly downloads | 0 |
| License | MIT |
| TypeScript types | none |
| Module format | CommonJS |
| Node | >=6 |
| Dependencies | 0 |
| Unpacked size | 11.8 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| GitHub stars | 1 |
| Author | Toshimitsu Takahashi |
| Maintainers | tilfin |
| Keywords | pace, work, process, stream, concurrent, constant |

## Links

- npm: https://www.npmjs.com/package/paced-work-stream
- Repository: https://github.com/tilfin/paced-work-stream
- Homepage: https://github.com/tilfin/paced-work-stream#readme
- Issues: https://github.com/tilfin/paced-work-stream/issues
- npm.io page: https://npm.io/package/paced-work-stream

## Alternatives

- [@sveltejs/kit](https://npm.io/package/@sveltejs/kit.md) — 2.2M weekly downloads
- [@tangle-network/brand](https://npm.io/package/@tangle-network/brand.md) — 10.0K weekly downloads
- [seneca](https://npm.io/package/seneca.md) — 7.4K weekly downloads
- [@bsb/base](https://npm.io/package/@bsb/base.md) — 7.2K weekly downloads
- [@modern-js/plugin-swc](https://npm.io/package/@modern-js/plugin-swc.md) — 3.4K weekly downloads

## Recent versions

- 1.0.3 (latest) — 2018-11-27
- 1.0.2 — 2018-04-11
- 1.0.1 — 2017-03-10
- 1.0.0 — 2017-03-10
- 0.10.0 — 2017-03-10
- 0.9.4 — 2017-03-10
- 0.9.3 — 2016-10-01
- 0.9.2 — 2016-09-20
- 0.9.1 — 2016-09-13
- 0.9.0 — 2016-09-12
- 0.8.1 — 2016-09-12
- 0.8.0 — 2016-09-11
- 0.1.0 — 2016-09-11

## README

PacedWorkStream
===============

[![NPM Version][npm-image]][npm-url]
[![Node](https://img.shields.io/node/v/paced-work-stream.svg)]()
[![Build Status](https://travis-ci.org/tilfin/paced-work-stream.svg?branch=master)](https://travis-ci.org/tilfin/paced-work-stream)
[![Coverage Status](https://coveralls.io/repos/github/tilfin/paced-work-stream/badge.svg?branch=master)](https://coveralls.io/github/tilfin/paced-work-stream?branch=master)
[![dependencies Status](https://david-dm.org/tilfin/paced-work-stream/status.svg)](https://david-dm.org/tilfin/paced-work-stream)

Node.js transform stream working at constant pace and concurrent for object mode

## Features

* Work time at once can be specified. (workMS option)
* Concurrent workers can be specified. (concurrency option)
* Fires `done` event after when all workers have finished asynchrous -processes
* Counting tag system to call `this.countTag(<tag>)` in `_workPromise`, you can get summarized results `tagCounts` grouped by tag.
* Node.js 6.10 or later

## Targets

* API client that needs to handle the rate-limit
* DB client that needs to handle the read/write capacity units like AWS DynamoDB

## Install

```
$ npm install -save paced-work-stream
```

## How to Use

### Create a PacedWorkStream

**new PacedWorkStream(options, workPromise)**

* `options` `<Object>`
  * `concurrency` is the number of concurrent processes.
  * `workMS` is milliseconds of work time at once that contains process-time and wait-time.
  * `delay` is enable to start concurrent process in order delay for a time that divided workMS by concurrency, default is false. workPromise must return functions wrap each promise. Refer to the following figure for detailed operation pattern.
  * `highWaterMark` is maximum object buffer size. If you use flow mode, you should set it at least concurrency.
* `workPromise` is `function(item):` must return a _Promise_ processing the _item_ or a _Function_ that returns a _Promise_.

![Delay Figure](https://github.com/tilfin/paced-work-stream/wiki/delay-figure.png)

### Create subclass that extends PacedWorkStream

* `super(options)` must be called in the constructor.
* `_workPromise` method must be overrided and return a _Promise_ processing the _item_ or a _Function_ that returns a _Promise_.

```javascript
class MyWorkStream extends PacedWorkStream {
  constructor(options) {
    super(options);
  }
  _workPromise(item) {
    return () => {
      this.countTag(item.tag);
      return Promise.resolve(item.value);
    };
  }
}
```

## Examples

```javascript
const es = require('event-stream');
const devnull = require('dev-null');
const PacedWorkStream = require('paced-work-stream');

const pwStream = new PacedWorkStream({
    concurrency: 2,
    workMS: 1000,
    highWaterMark: 5
  }, function(item) {
    console.log(new Date().toISOString(), 'Begin', item);

    return new Promise((resolve, reject) => {
        setTimeout(() => {
          this.countTag('workDone');
          console.log(new Date().toISOString(), 'End', item);
          resolve();
        }, 600); // workMS contains the time.
      })
  })
  .on('done', function() {
    console.log(this.tagCounts);
  }).on('error', (err) => {
    console.error(err);
  });

const reader = es.readArray([11, 12, 21, 22, 31])
reader.pipe(pwStream).pipe(devnull({ objectMode: true }));
```

* Pay attention to handling `done` event to get last `tagCounts` because workers haven't processed items on `finish` event.
* If stream need not output, the stream must pipe dev-null.

### Console output

```
$ node example.js
2016-09-11T03:17:50.000Z Begin 11
2016-09-11T03:17:50.003Z Begin 12
2016-09-11T03:17:50.605Z End 11
2016-09-11T03:17:50.605Z End 12
2016-09-11T03:17:51.009Z Begin 21
2016-09-11T03:17:51.009Z Begin 22
2016-09-11T03:17:51.606Z End 21
2016-09-11T03:17:51.606Z End 22
2016-09-11T03:17:52.004Z Begin 31
2016-09-11T03:17:52.607Z End 31
{ workDone: 5 }
```

### Using with Promised Lifestream

[Promised Lifestream](https://github.com/tilfin/promised-lifestream) is useful for stream pipeline. The following example gets the same result as above.

```javascript
'use strict';

const es = require('event-stream');
const PromisedLife = require('promised-lifestream');

const PacedWorkStream = require('paced-work-stream');

const pacedWorker = new PacedWorkStream({
    concurrency: 2,
    workMS: 1000,
    highWaterMark: 5
  }, function(item) {
    console.log(new Date().toISOString(), 'Begin', item);

    return new Promise((resolve, reject) => {
        setTimeout(() => {
          this.countTag('workDone');
          console.log(new Date().toISOString(), 'End', item);
          resolve();
        }, 600); // workMS contains the time.
      })
  })

PromisedLife([
  es.readArray([11, 12, 21, 22, 31]),
  pacedWorker
])
.then(() => {
  console.log(pacedWorker.tagCounts);
})
.catch(err => {
  console.error(err);
});
```


## License

  [MIT](LICENSE)

[npm-image]: https://img.shields.io/npm/v/paced-work-stream.svg
[npm-url]: https://npmjs.org/package/paced-work-stream

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