jrf-pip

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:

name	type	required	default	description
arrayValues	array	true		An array of values to process
processingFn	function	true		A synchronous or asynchronous function that processes an array element. The input accepts parameters paramsForFunction
nextValueFn	function	false		A synchronous or asynchronous function that performs filtering. If it returns true, then the current element of the array will not be processed. false - the current element of the array will be processed. The input accepts parameters paramsForFunction
parallel	number	false	1000	Number of elements to be processed in parallel per iteration
awaitRes	boolean	false	false	Expect an answer i.e. wait for completion to complete in synchronous style
cb	function	false		A synchronous or asynchronous callback function, will work if awaitRes = false. An error stack will be passed to the function stackError
cycleTimeout	number	false	200	Asynchronous pause between iterations (milliseconds), for awaitRes = true

paramsForFunction

Parameters passed to the processingFn and nextValueFn function

name	type	description
value	any	Array element to process
index	number	Array element index
arrayValues	array	Array
iteration	number	Iteration number

stackError

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

Array of objects:

name	type	description
value	any	Array element during processing of which an error occurred
index	number	Array element index
iteration	number	Iteration number
error	object	Error