Fixes to 2 issues

[gavinklfong]

2 issues:

1. Duplicated message sent to SQS as buffer is not cleaned up
2. Message is not sent to SQS when buffer size equals to batch size

Please find amended version of _write() and _final() below

async _write(obj, enc, cb) {
this.emit('msgReceived', obj);
try {
if (!this.queueUrl) {
await this.getQueueUrl();
}

  // construct message
  const sqsMsg = _.clone(obj);
  if (obj.Id && obj.MessageBody) {
    this.buffer.push(sqsMsg);
  } else {
    const msg = { MessageBody: JSON.stringify(obj), Id: uuidv4() };
    if (this.options.MessageGroupId) {
      msg.MessageGroupId = this.options.MessageGroupId;
    }
    this.buffer.push(msg);
  }

  // send message by batch
  if (this.buffer.length >= this.options.batchSize) {
    console.log("_write() - sending sqs message: " + JSON.stringify(this.buffer));
    await this.sqs.sendMessageBatch({ Entries: this.buffer, QueueUrl: this.queueUrl }).promise();
    this.buffer = [];
  } 

  this.emit('msgProcessed', obj);
  return cb();
} catch (err) {
  this.emit('msgProcessingErr', obj, err);
  return cb(err);
}

}

async _final(cb) {
try {
if (this.buffer.length > 0) {
console.log("_final() - sending sqs message: " + JSON.stringify(this.buffer));
await this.sqs.sendMessageBatch({ Entries: this.buffer, QueueUrl: this.queueUrl }).promise();
this.buffer = [];
}

  return cb();
} catch (err) {
  this.emit('streamFinishingErr', err);
  return cb(err);
}

}