sequencing.js
function sequenceMessages(batch, context) {
  const states = batch.states;
  const messages = batch.messages;

  if (messages.length < 2 || !context.streamProcessing.sequencingRequired) {
    return messages;
  }

  // First prepare the messages for sequencing by normalizing all of the messages' sequence numbers
  prepareMessagesForSequencing(messages, states, context);

  const firstMessagesToProcess = [];

  const sequencingPerKey = context.streamProcessing.sequencingPerKey;
  const comparator = sequencingPerKey ?
    (m1, m2) => compareSameKeyMessages(m1, m2, states, context):
    (m1, m2) => compareAnyKeyMessages(m1, m2, states, context);

  // Group all of the messages by the stringified versions of their keys (to avoid strict equal failures)
  const messagesByKeyString = sequencingPerKey ?
    groupBy(messages, msg => {
      const msgState = states.get(msg);
      const keys = msgState.keys;
      return keys && keys.length > 0 ? msgState.key : '?';
    }) : {'*': messages};

  // Sort all of the messages that share the same keys by their sequence numbers and then link them to each other via
  // their prevMessage and nextMessage state properties
  const keyStrings = Object.getOwnPropertyNames(messagesByKeyString);

  for (let i = 0; i < keyStrings.length; ++i) {
    const keyString = keyStrings[i];
    const msgs = messagesByKeyString[keyString];

    if (context.traceEnabled && msgs.length > 1) context.trace(`BEFORE sorting (${keyString}): ${stringify(msgs.map(m => states.get(m).seqNo))}`);

    msgs.sort(comparator);

    if (context.traceEnabled) context.trace(` AFTER sorting (${keyString}): ${stringify(msgs.map(m => states.get(m).seqNo))}`);

    let prevMessage = undefined;
    let prevMessageState = undefined;

    firstMessagesToProcess.push(msgs[0]);

    for (let m = 0; m < msgs.length; ++m) {
      const nextMessage = msgs[m];
      const nextMessageState = states.get(nextMessage);

      setPrevMessage(nextMessageState, prevMessage);

      if (prevMessageState) {
        setNextMessage(prevMessageState, nextMessage);
      }
      prevMessage = nextMessage;
      prevMessageState = nextMessageState;
    }
    if (prevMessageState) {
      setNextMessage(prevMessageState, undefined);
    }
  }

  if (context.debugEnabled) {
    const firstMessagesDetails = firstMessagesToProcess.map(m => `(${getMessageStateKeySeqNoId(states.get(m))})`).join(", ");
    context.debug(`Finished sequencing messages - found ${firstMessagesToProcess.length} first message${firstMessagesToProcess.length !== 1 ? 's' : ''} to process out of ${messages.length} message${messages.length !== 1 ? 's' : ''} - first [${firstMessagesDetails}]`);
  }

  return firstMessagesToProcess;
}
function sequenceMessages(batch, context) {
  const states = batch.states;
  const messages = batch.messages;

  if (messages.length < 2 || !context.streamProcessing.sequencingRequired) {
    return messages;
  }

  // First prepare the messages for sequencing by normalizing all of the messages' sequence numbers
  prepareMessagesForSequencing(messages, states, context);

  const firstMessagesToProcess = [];

  const sequencingPerKey = context.streamProcessing.sequencingPerKey;
  const comparator = sequencingPerKey ?
    (m1, m2) => compareSameKeyMessages(m1, m2, states, context):
    (m1, m2) => compareAnyKeyMessages(m1, m2, states, context);

  // Group all of the messages by the stringified versions of their keys (to avoid strict equal failures)
  const messagesByKeyString = sequencingPerKey ?
    groupBy(messages, msg => {
      const msgState = states.get(msg);
      const keys = msgState.keys;
      return keys && keys.length > 0 ? msgState.key : '?';
    }) : {'*': messages};

  // Sort all of the messages that share the same keys by their sequence numbers and then link them to each other via
  // their prevMessage and nextMessage state properties
  const keyStrings = Object.getOwnPropertyNames(messagesByKeyString);

  for (let i = 0; i < keyStrings.length; ++i) {
    const keyString = keyStrings[i];
    const msgs = messagesByKeyString[keyString];

    if (context.traceEnabled && msgs.length > 1) context.trace(`BEFORE sorting (${keyString}): ${stringify(msgs.map(m => states.get(m).seqNo))}`);

    msgs.sort(comparator);

    if (context.traceEnabled) context.trace(` AFTER sorting (${keyString}): ${stringify(msgs.map(m => states.get(m).seqNo))}`);

    let prevMessage = undefined;
    let prevMessageState = undefined;

    firstMessagesToProcess.push(msgs[0]);

    for (let m = 0; m < msgs.length; ++m) {
      const nextMessage = msgs[m];
      const nextMessageState = states.get(nextMessage);

      setPrevMessage(nextMessageState, prevMessage);

      if (prevMessageState) {
        setNextMessage(prevMessageState, nextMessage);
      }
      prevMessage = nextMessage;
      prevMessageState = nextMessageState;
    }
    if (prevMessageState) {
      setNextMessage(prevMessageState, undefined);
    }
  }

  if (context.debugEnabled) {
    const firstMessagesDetails = firstMessagesToProcess.map(m => `(${getMessageStateKeySeqNoId(states.get(m))})`).join(", ");
    context.debug(`Finished sequencing messages - found ${firstMessagesToProcess.length} first message${firstMessagesToProcess.length !== 1 ? 's' : ''} to process out of ${messages.length} message${messages.length !== 1 ? 's' : ''} - first [${firstMessagesDetails}]`);
  }

  return firstMessagesToProcess;
}
function sequenceMessages ( batch , context ) { const states = batch . states ; const messages = batch . messages ; if ( messages . length < 2 || ! context . streamProcessing . sequencingRequired ) { return messages ; } // First prepare the messages for sequencing by normalizing all of the messages' sequence numbers prepareMessagesForSequencing ( messages , states , context ) ; const firstMessagesToProcess = [ ] ; const sequencingPerKey = context . streamProcessing . sequencingPerKey ; const comparator = sequencingPerKey ? ( m1 , m2 ) => compareSameKeyMessages ( m1 , m2 , states , context ) : ( m1 , m2 ) => compareAnyKeyMessages ( m1 , m2 , states , context ) ; // Group all of the messages by the stringified versions of their keys (to avoid strict equal failures) const messagesByKeyString = sequencingPerKey ? groupBy ( messages , msg => { const msgState = states . get ( msg ) ; const keys = msgState . keys ; return keys && keys . length > 0 ? msgState . key : '?' ; } ) : { '*' : messages } ; // Sort all of the messages that share the same keys by their sequence numbers and then link them to each other via // their prevMessage and nextMessage state properties const keyStrings = Object . getOwnPropertyNames ( messagesByKeyString ) ; for ( let i = 0 ; i < keyStrings . length ; ++ i ) { const keyString = keyStrings [ i ] ; const msgs = messagesByKeyString [ keyString ] ; if ( context . traceEnabled && msgs . length > 1 ) context . trace ( ` ${ keyString } ${ stringify ( msgs . map ( m => states . get ( m ) . seqNo ) ) } ` ) ; msgs . sort ( comparator ) ; if ( context . traceEnabled ) context . trace ( ` ${ keyString } ${ stringify ( msgs . map ( m => states . get ( m ) . seqNo ) ) } ` ) ; let prevMessage = undefined ; let prevMessageState = undefined ; firstMessagesToProcess . push ( msgs [ 0 ] ) ; for ( let m = 0 ; m < msgs . length ; ++ m ) { const nextMessage = msgs [ m ] ; const nextMessageState = states . get ( nextMessage ) ; setPrevMessage ( nextMessageState , prevMessage ) ; if ( prevMessageState ) { setNextMessage ( prevMessageState , nextMessage ) ; } prevMessage = nextMessage ; prevMessageState = nextMessageState ; } if ( prevMessageState ) { setNextMessage ( prevMessageState , undefined ) ; } } if ( context . debugEnabled ) { const firstMessagesDetails = firstMessagesToProcess . map ( m => ` ${ getMessageStateKeySeqNoId ( states . get ( m ) ) } ` ) . join ( ", " ) ; context . debug ( ` ${ firstMessagesToProcess . length } ${ firstMessagesToProcess . length !== 1 ? 's' : '' } ${ messages . length } ${ messages . length !== 1 ? 's' : '' } ${ firstMessagesDetails } ` ) ; } return firstMessagesToProcess ; }
Sequences the given entire batch of messages by linking messages with the same keys to each other via their
nextMessage and prevMessage properties in sequences determined by their sequence numbers.
@param {Batch} batch - the batch containing the messages to be sequenced
@param {StreamConsumerContext} context - the context to use
@returns {Message[]} the first messages to be processed (i.e. all of the messages that do NOT have a defined prevMessage
Sequences the given entire batch of messages by linking messages with the same keys to each other via their nextMessage and prevMessage properties in sequences determined by their sequence numbers .