jrf-pip v1.0.0

jrf-pip (parallel iteration processing)

jrf-pip - This is a package for parallel iterative processing of large arrays. Allows parallel processing asynchronously, or asynchronously in synchronous style. Processing implemented through Promise.all()

Example

const parallelProcessing = require('jrf-pip');

// Send a message to group users
async function sendMesToGroupUsers() {

  // Generate groups of 10,000 users.
  // 5000 female
  // 5000 males
  const groupUsers = generateUsers();

  // Filter Function
  // Skip Male Users
  // i.e. send message only to female
  const nextValueFn = ({value, index, arrayValues, iteration}) => {
    return value.sex === 'male';
  };

  // Processing Function
  // Send a message
  const processingFn = ({value, index, arrayValues, iteration}) => {
    const user = value;
    user.sendMes(iteration);
  };

  // Callback function. It will work after the completion of sending the message.
  const cb = (stackError) => console.log('Finish parallel iteration processing');

  // Starting parallel iterative processing
  // An array of arrayValues is specified for processing
  // filter function nextValueFn specified
  // the processing function of the processingFn array element is specified
  // a callback function is set that will be executed when parallel processing is completed
  // set the number of 1000 parallel processed array values through Promise.all()
  await parallelProcessing({
    arrayValues: groupUsers,
    nextValueFn,
    processingFn,
    cb,
    parallel: 1000
  });

  console.log('Before finish');

}

// Send a message to the users of the group in a synchronous style
async function sendMesToGroupUsersSyncStyle() {

  // Generate groups of 10,000 users.
  // 5000 female
  // 5000 males
  const groupUsers = generateUsers();

  // Filter Function
  // Skip Male Users
  // i.e. send message only to female
  const nextValueFn = ({value, index, arrayValues, iteration}) => {
    return value.sex === 'male';
  };

  // Processing Function
  // Send a message
  const processingFn = ({value, index, arrayValues, iteration}) => {
    const user = value;
    user.sendMes(iteration);
  };

  // Starting parallel iterative processing
  // An array of arrayValues is specified for processing
  // filter function nextValueFn specified
  // the processing function of the processingFn array element is specified
  // the awaitRes response wait parameter is specified i.e. synchronous execution
  // set the number of 2000 parallel values of the array through Promise.all()
  // time of cycleTimeout of an asynchronous pause between iterations is set
  const stackError = await parallelProcessing({
    arrayValues: groupUsers,
    nextValueFn,
    processingFn,
    awaitRes: true,
    parallel: 2000,
    cycleTimeout: 100
  });

  console.log('Finish parallel iteration processing in sync style');

}

function generateUsers(count = 10000) {

  const groupUsers = [];

  const even = n => !(n % 2);
  for (let i = 1; i <= count; i++) {

    const sex = even(i) ? 'male' : 'female';
    groupUsers.push({
      id: i,
      sex,
      sendMes(iteration) {
        console.log(`send mes to user id: ${this.id} sex: ${this.sex} iteration: ${iteration}`);
      }
    });

  }

  return groupUsers;

}

Promise.resolve()
  .then(sendMesToGroupUsers)
  .then(sendMesToGroupUsersSyncStyle)
  .catch(e => console.error(e));

Parameters

The parallelProcessing function at the input accepts the following parameters:

paramsForFunction

Parameters passed to the processingFn and nextValueFn function

stackError

An error stack returned with awaitRes = true, or passed to a function callback when awaitRes = false.

Array of objects: