Method handleMessage

pkg/dequeuer/dequeuer.go:180–218

(message *sqs.Message, messageHandler MessageHandler, done chan struct{})

}

func (d *SQSDequeuer) handleMessage(message *sqs.Message, messageHandler MessageHandler, done chan struct{}) error {
	messageErr := messageHandler.Handle(message) // handle error later

	done <- struct{}{}
	isOnJobComplete := isOnJobCompleteMessage(message)

	if messageErr != nil && d.hasDeadLetterQueue && !isOnJobComplete {
		// expire messages when dead letter queue is configured to facilitate redrive policy.
		// always delete onJobComplete messages regardless of redrive policy because a new one will
		// be added if an onJobComplete message has been consumed prematurely
		_, err := d.aws.SQS().ChangeMessageVisibility(
			&sqs.ChangeMessageVisibilityInput{
				QueueUrl:          &d.config.QueueURL,
				ReceiptHandle:     message.ReceiptHandle,
				VisibilityTimeout: aws.Int64(0),
			},
		)
		if err != nil {
			return errors.Wrap(err, "failed to change sqs message visibility")
		}
		return nil
	}

	_, err := d.aws.SQS().DeleteMessage(
		&sqs.DeleteMessageInput{
			QueueUrl:      &d.config.QueueURL,
			ReceiptHandle: message.ReceiptHandle,
		},
	)
	if err != nil {
		return errors.Wrap(err, "failed to delete sqs message")
	}

	if messageErr != nil {
		return messageErr
	}

	return nil
}

func (d *SQSDequeuer) StartMessageRenewer(receiptHandle string) chan struct{} {
	done := make(chan struct{})