persisting.js
function toBatchStateItem(batch, context) {
  const states = batch.states;
  // const messages = batch.allMessages();
  const messages = batch.messages;
  const rejectedMessages = batch.rejectedMessages;
  const unusableRecords = batch.unusableRecords;

  // Resolve the messages' states to be saved (if any)
  const messageStates = messages && messages.length > 0 ?
    messages.map(msg => toStorableMessageState(states.get(msg), msg, context)).filter(s => !!s) : [];

  // Resolve the rejected messages' states to be saved (if any)
  const rejectedMessageStates = rejectedMessages && rejectedMessages.length > 0 ?
    rejectedMessages.map(msg => toStorableMessageState(states.get(msg), msg, context)).filter(s => !!s) : [];

  // Resolve the unusable records' states to be saved (if any)
  const unusableRecordStates = unusableRecords && unusableRecords.length > 0 ?
    unusableRecords.map(uRec => toStorableUnusableRecordState(states.get(uRec), uRec, context)).filter(s => !!s) : [];

  const batchState = toStorableBatchState(states.get(batch), batch, context);

  return {
    streamConsumerId: batch.streamConsumerId, // hash key
    shardOrEventID: batch.shardOrEventID, // range key
    messageStates: messageStates,
    rejectedMessageStates: rejectedMessageStates,
    unusableRecordStates: unusableRecordStates,
    batchState: batchState || null
  };
}
function toBatchStateItem(batch, context) {
  const states = batch.states;
  // const messages = batch.allMessages();
  const messages = batch.messages;
  const rejectedMessages = batch.rejectedMessages;
  const unusableRecords = batch.unusableRecords;

  // Resolve the messages' states to be saved (if any)
  const messageStates = messages && messages.length > 0 ?
    messages.map(msg => toStorableMessageState(states.get(msg), msg, context)).filter(s => !!s) : [];

  // Resolve the rejected messages' states to be saved (if any)
  const rejectedMessageStates = rejectedMessages && rejectedMessages.length > 0 ?
    rejectedMessages.map(msg => toStorableMessageState(states.get(msg), msg, context)).filter(s => !!s) : [];

  // Resolve the unusable records' states to be saved (if any)
  const unusableRecordStates = unusableRecords && unusableRecords.length > 0 ?
    unusableRecords.map(uRec => toStorableUnusableRecordState(states.get(uRec), uRec, context)).filter(s => !!s) : [];

  const batchState = toStorableBatchState(states.get(batch), batch, context);

  return {
    streamConsumerId: batch.streamConsumerId, // hash key
    shardOrEventID: batch.shardOrEventID, // range key
    messageStates: messageStates,
    rejectedMessageStates: rejectedMessageStates,
    unusableRecordStates: unusableRecordStates,
    batchState: batchState || null
  };
}
function toBatchStateItem ( batch , context ) { const states = batch . states ; // const messages = batch.allMessages(); const messages = batch . messages ; const rejectedMessages = batch . rejectedMessages ; const unusableRecords = batch . unusableRecords ; // Resolve the messages' states to be saved (if any) const messageStates = messages && messages . length > 0 ? messages . map ( msg => toStorableMessageState ( states . get ( msg ) , msg , context ) ) . filter ( s => ! ! s ) : [ ] ; // Resolve the rejected messages' states to be saved (if any) const rejectedMessageStates = rejectedMessages && rejectedMessages . length > 0 ? rejectedMessages . map ( msg => toStorableMessageState ( states . get ( msg ) , msg , context ) ) . filter ( s => ! ! s ) : [ ] ; // Resolve the unusable records' states to be saved (if any) const unusableRecordStates = unusableRecords && unusableRecords . length > 0 ? unusableRecords . map ( uRec => toStorableUnusableRecordState ( states . get ( uRec ) , uRec , context ) ) . filter ( s => ! ! s ) : [ ] ; const batchState = toStorableBatchState ( states . get ( batch ) , batch , context ) ; return { streamConsumerId : batch . streamConsumerId , // hash key shardOrEventID : batch . shardOrEventID , // range key messageStates : messageStates , rejectedMessageStates : rejectedMessageStates , unusableRecordStates : unusableRecordStates , batchState : batchState || null } ; }
Converts the given batch into a stream consumer batch state item to be subsequently persisted to DynamoDB.
@param {Batch} batch - the batch to be converted into a batch state item
@param {StreamProcessing} context - the context to use
@returns {BatchStateItem} the stream consumer batch state item
Converts the given batch into a stream consumer batch state item to be subsequently persisted to DynamoDB .