highland_aws_sqs_batch.js

const BATCH_SIZE = 10
const buildParamforSqsBatch = (products, queueUrl) => {
  const entries = products.map(product => toOutputMessage(product))
  return {
    Entries: entries,
    QueueUrl: queueUrl
  }
}

const emitUpdateEvent = (params) => {
  return SQS().sendMessageBatch(params).promise()
}

exports.emitUpdateEventBatch = async (variantEvents) => {
  let errors = []
  const res = await highland(variantEvents)
    .batch(BATCH_SIZE)
    .map(batch => buildParamforSqsBatch(batch, queueUrl)) // [{Entries, QueueUrl}]
    .ratelimit(2000, 1000) // 2000 batched records per 1000ms
    .flatMap(events => highland(emitUpdateEvent(events))) // highland([ Promise<>, ])
    .errors(error => errors.push(error))
    .collect()
    .toPromise(Promise)

  logger.info({ errors }, `Error on emitting records variant parent keycodes`)
  return res
}