@paybase/pool

@paybase/pool

async/await pooling mechanism built on top of Communicating Sequential Processes.

Installation

$ npm install --save @paybase/pool

or

$ yarn add @paybase/pool

Example Usage

Pooling can simplify many complex problems, for example:

Network I/O parallelisation

By defining a process as something which is akin to a mutex lock, we can limit request parallelisation to the size of the pool.

const assert = require('assert');
const fetch = require('node-fetch');
const createPool = require('@paybase/pool');

const { run, close } = createPool({
  poolSize: 2,
  createProcess: () => Symbol('ticket'),
  handler: (_, query) => {
    console.log(`🚀  running request with query: ${query}`);
    return fetch(`https://postman-echo.com/get?q=${query}`)
      .then(res => res.json())
      .then(res => assert.equal(res.args.q, query));
      .then(_ => console.log(`👌  request completed successfully`));
  }
});

(async () => {
  const queries = Array.from({ length: 20 }, (_, i) => run(`${++i}`));
  await Promise.all(queries);
  close();
})();

Child process pooling

For spawning multiple child processes and spreading usage between the pool.

const assert = require('assert');
const { spawn } = require('child_process');
const createPool = require('@paybase/pool');

const { run, close } = createPool({
  poolSize: 10,
  createProcess: () => {
    const p = spawn('cat', [ '-' ]);
    p.stdin.setEncoding('utf-8');
    p.stdout.setEncoding('utf-8'); 
    return p;
  },
  handler: (p, input) =>
    new Promise(resolve => {
      p.stdout.once('data', d => {
        assert(d, input);
        console.log(`👌  received data: ${d.trim()} from pid: ${p.pid}`);
        resolve(d);
      });
      console.log(`🚀  sending data: ${input.trim()} to pid: ${p.pid}`);
      p.stdin.write(input);
    }),   
});

(async () => {
  const inputs = Array.from({ length: 100 }, (_, i) => run(`${++i}\n`));
  await Promise.all(inputs);
  close();
})();

API

This library exposes a single factory method for creating pools.

createPool({ poolSize = 5, createProcess, handler }) -> Pool

The pool factory takes an options object containing 3 properties:

• poolSize - defaults to 5, determines the size of the pool
• createProcess - defines a process factory function which can return anything
• handler(process, input) -> Promise - defines a function which handles a unit of work. The handler must return a Promise and receives a process (as defined by the process factory) and the input from a call to run on the pool

A returned Pool exposes 2 methods:

• Pool.run(input) -> Promise - The interface defined to run against the pool of processes, supplied input can be of any type as the handler supplied at pool creation defines how the input interacts which the underlying process
• Pool.close -> Promise - A mechanism for destroying the pool when it is no longer needed

Contributions

Contributions are welcomed and appreciated!

1. Fork this repository.
2. Make your changes, documenting your new code with comments.
3. Submit a pull request with a sane commit message.

Feel free to get in touch if you have any questions.

License

Please see the LICENSE file for more information.