Global-sales/wai-server/core/emitter/queueProcessor.js

'use strict';

const { default: pRetry } = require('@fullstax/p-retry');

/**
 * Creates an async queue processor.
 * 不同 itemId 的数据并发处理
 * 每个 itemId 的数据有序地仅处理一个, 先进先出
 * @param {Function} processItemFn - The function to process each item in the queue.
 * @param {Object} [options] - Options for the queue processor.
 * @param {Object} [options.retryOptions] - Options for retrying failed items.
 * @param {number} [options.retryOptions.retries] - Number of retries for failed items.
 * @returns {Object} - An object with an enqueue method to add items tothe queue.
 */
async function createAsyncQueueProcessor(processItemFn, options = {}) {
  if (typeof processItemFn !== 'function') {
    throw new Error('processItemFn must be a function');
  }

  const { retryOptions = { retries: 0 } } = options;
  const queues = new Map();
  const processing = new Map();

  async function processQueue(itemId) {
    if (processing.get(itemId)) {
      return; // processing
    }

    processing.set(itemId, true);

    try {
      const queue = queues.get(itemId);
      if (!queue) return;

      while (queue.length > 0) {
        const item = queue.shift();
        try {
          if (retryOptions.retries > 0) {
            await pRetry(async () => {
              await processItemFn(item);
            }, retryOptions);
          } else {
            await processItemFn(item);
          }
        } catch (itemError) {
          console.error(`Error processing item ${JSON.stringify(item)}:\n`, itemError);
          // Handle error (e.g., retry, log, DLQ)
        }
      }
      queues.delete(itemId); // Clean up empty queue
    } finally {
      processing.set(itemId, false);
    }
  }

  function enqueue(itemId, itemData) {
    if (!queues.has(itemId)) {
      queues.set(itemId, []);
    }

    queues.get(itemId).push(itemData);
    processQueue(itemId);
  }

  return { enqueue };
}

module.exports = createAsyncQueueProcessor;