Source: engine/universal/core/src/engine/processForwarding.js

const system = require('@proceed/system');
const { db, communication } = require('@proceed/distribution');

/**
 * Sends the process definition to the next machine
 *
 * @param {String} ip the ip of the other machine
 * @param {Number} port the port the engine is published on on the other machine
 * @param {String} definitionId the name of the file the process is supposed to be stored under
 * @param {String} bpmn the process definition
 */
async function forwardProcess(ip, port, definitionId, bpmn) {
  await system.network.sendData(ip, port, `/process/${definitionId}`, 'PUT', 'application/json', {
    bpmn,
  });
}

/**
 * Sends the necessary instance information and the signal to continue the process on the next machine
 *
 * @param {String} ip the ip of the next machine
 * @param {Number} port the port the engine is published on on the next machine
 * @param {String} definitionId the name of the file the process information is supposed to be stored under on the next machine
 * @param {String} instanceId id of the instance we want to continue
 * @param {Object} instanceInfo the complete instance information that exists at this point
 */
async function forwardInstance(ip, port, definitionId, instanceId, instanceInfo) {
  await system.network.sendData(
    ip,
    port,
    `/process/${definitionId}/instance/${instanceId}`,
    'PUT',
    'application/json',
    instanceInfo
  );
}

/**
 * Sends the html information for all user tasks to the next machine
 *
 * @param {String} ip the ip of the next machine
 * @param {Number} port the port the engine is published on on the next machine
 * @param {String} definitionId the name of the file the process information is supposed to be stored under on the next machine
 * @param {Boolean} imported indicates if we want to send the HTML information for imported processes or just the ones for the main process
 */
async function forwardHTML(ip, port, definitionId, imported) {
  const htmlFileNames = await db.getAllUserTasks(definitionId, imported);
  let endpoint;

  if (!imported) {
    endpoint = `/process/${definitionId}/user-tasks`;
  } else {
    endpoint = `/process/${definitionId}/imported/user-tasks`;
  }

  const htmlSendRequests = htmlFileNames.map(async (htmlFileName) => {
    const html = await db.getHTML(definitionId, htmlFileName, imported);

    await system.network.sendData(
      ip,
      port,
      `${endpoint}/${htmlFileName}`,
      'PUT',
      'application/json',
      {
        html,
      }
    );
  });

  await Promise.all(htmlSendRequests);
}

/**
 * Sends the process descriptions and user task data for all imported processes to the next machine
 *
 * @param {String} ip the ip of the next machine
 * @param {Number} port the port the engine is published on on the next machine
 * @param {String} definitionId the name of the file the process information is supposed to be stored under on the next machine
 */
async function forwardImports(ip, port, definitionId) {
  const importedProcesses = await db.getImportedProcesses(definitionId);

  const importSendRequests = Object.entries(importedProcesses).reduce(
    (curRequests, [importDefinitionId, importDefinitions]) => {
      const newRequests = [...curRequests];

      newRequests.push(
        system.network.sendData(
          ip,
          port,
          `/process/${definitionId}/imported/${importDefinitionId}`,
          'PUT',
          'application/json',
          { bpmn: importDefinitions }
        )
      );

      return newRequests;
    },
    []
  );

  await Promise.all(importSendRequests);

  await forwardHTML(ip, port, definitionId, true);
}

/**
 * Requests additional information about a machine
 *
 * @param {String} ip the ip of the machine
 * @param {Number} port the port the engine is running on
 */
async function getMachineInfo(ip, port) {
  const { body } = await system.network.sendRequest(ip, port, '/machine/id,name,hostname');

  return JSON.parse(body);
}

/**
 * Sends request to abort process instance to all machines in the local network
 *
 * @param {string} definitionId name of the file the process is stored in
 * @param {string} instanceId the id of the instance to abort
 */
async function abortInstanceOnNetwork(definitionId, instanceId) {
  const requests = communication.getAvailableMachines().map(async (machine) => {
    try {
      await system.network.sendData(
        machine.ip,
        machine.port,
        `/process/${definitionId}/instance/${instanceId}/instanceState`,
        'PUT',
        'application/json',
        { instanceState: 'aborted' }
      );
    } catch (err) {}
  });

  await Promise.all(requests);
}

/**
 * Sends request to stop process instance to all machines in the local network
 *
 * @param {string} definitionId name of the file the process is stored in
 * @param {string} instanceId the id of the instance to stop
 */
async function stopInstanceOnNetwork(definitionId, instanceId) {
  const requests = communication.getAvailableMachines().map(async (machine) => {
    try {
      await system.network.sendData(
        machine.ip,
        machine.port,
        `/process/${definitionId}/instance/${instanceId}/instanceState`,
        'PUT',
        'application/json',
        { instanceState: 'stopped' }
      );
    } catch (err) {}
  });

  await Promise.all(requests);
}

/**
 * @memberof module:@proceed/core
 *
 * Exposes functions to send necessary process information to the next machine
 */
module.exports = {
  forwardProcess,
  forwardInstance,
  forwardHTML,
  forwardImports,
  getMachineInfo,
  abortInstanceOnNetwork,
  stopInstanceOnNetwork,
};