{"size": 3989, "ext": "js", "max_stars_repo_path": "src/subscription-handler.js", "max_stars_repo_name": "MagnusAk78/mongodb-message-store", "max_stars_repo_stars_event_min_datetime": "2021-05-14T12:33:56.000Z", "max_stars_repo_stars_event_max_datetime": "2021-09-06T04:29:24.000Z", "max_issues_repo_path": "src/subscription-handler.js", "max_issues_repo_name": "MagnusAk78/mongodb-message-store", "max_issues_repo_issues_event_min_datetime": null, "max_issues_repo_issues_event_max_datetime": null, "max_forks_repo_path": "src/subscription-handler.js", "max_forks_repo_name": "MagnusAk78/mongodb-message-store", "max_forks_repo_forks_event_min_datetime": null, "max_forks_repo_forks_event_max_datetime": null}
{"max_stars_count": 4, "max_issues_count": null, "max_forks_count": null, "avg_line_length": 30.4503816794, "max_line_length": 116, "alphanum_fraction": 0.6432689897}
const uuid = require('uuid/v4');

/**
 * The subscription handler exposes the createSubscription function.
 * @param {*} reader    reader
 * @param {*} writer    writer
 */
const createSubscriptionHandler = function (reader, writer) {
  // Internal function, reads subsribers position from database.
  async function readPosition(subscriberStreamName) {
    const lastPositionMessage = await reader.readLastMessage(subscriberStreamName);
    return lastPositionMessage ? lastPositionMessage.data.position : 0;
  }

  // Internal function, writes subsribers position to database.
  function writePosition(subscriberStreamName, position) {
    const positionMessage = {
      id: uuid(),
      type: 'ReadPosition',
      data: { position },
    };

    return writer.write(subscriberStreamName, positionMessage);
  }

  /**
   * Creates a subscription to a stream. A subscription must be stared with the start function
   * and stopped with the stop function.
   * @param {String} streamName                             Stream name
   * @param {Object<EventType, function (event)>} handlers  The handlers that should handle incomming events
   * @param {String} subscriberId                           Subscriber ID
   * @param {Number} maxMessagesPerTick                     How many messages should maximum be handled tick
   * @param {Number} positionUpdateInterval                 How often should the read position be stored to database
   * @param {Number} pollIntervalMs                         How often should subscription poll for new messages
   */
  function createSubscription(
    streamName,
    handlers,
    subscriberId,
    maxMessagesPerTick = 100,
    positionUpdateInterval = 100,
    pollIntervalMs = 100
  ) {
    const subscriberStreamName = ['subscriberPosition-', subscriberId].join().toString();

    let currentGlobalPosition = 0;
    let messagesSinceLastPositionWrite = 0;
    let keepGoing = true;

    // Internal function, updates the global position of the subscription.
    function updateGlobalPosition(globalPosition) {
      currentGlobalPosition = globalPosition;
      messagesSinceLastPositionWrite += 1;

      if (messagesSinceLastPositionWrite === positionUpdateInterval) {
        messagesSinceLastPositionWrite = 0;

        writePosition(subscriberStreamName, currentPosition);
      }
    }

    // Internal function, send message to corresponding handlers.
    function handleMessage(message) {
      const handler = handlers[message.type] || handlers.$any;

      return handler ? handler(message) : Promise.resolve(true);
    }

    // Internal function, read messages and handle them.
    async function tick() {
      let messagesHandled = 0;
      try {
        const messages = await reader.read(streamName, currentGlobalPosition + 1, maxMessagesPerTick);
        messages.forEach(async (message) => {
          try {
            await handleMessage(message);
            await updateGlobalPosition(message.globalPosition);
            messagesHandled += 1;
          } catch (err) {
            console.error('error handling message => %j', message);

            throw err;
          }
        });
      } catch (err) {
        console.error('Subscription failed, Error => %j', err);
        stop();
      }
      return messagesHandled;
    }

    // Internal function, poll for new messages.
    async function poll() {
      currentPosition = await readPosition(subscriberStreamName);

      while (keepGoing) {
        const nrOfMessages = await tick();

        if (nrOfMessages === 0) {
          await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
        }
      }
    }

    /**
     * Start the subscription.
     */
    function start() {
      return poll();
    }

    /**
     * Stop the subscription.
     */
    function stop() {
      keepGoing = false;
    }

    return {
      start,
      stop,
    };
  }

  return {
    createSubscription,
  };
};

module.exports = createSubscriptionHandler;
