Method onJobComplete

pkg/dequeuer/batch_handler.go:175–226

(message *sqs.Message)

}

func (h *BatchMessageHandler) onJobComplete(message *sqs.Message) error {
	shouldRunOnJobComplete := false
	h.log.Info("received job_complete message")
	for {
		queueAttributes, err := GetQueueAttributes(h.aws, h.config.QueueURL)
		if err != nil {
			return err
		}

		totalMessages := queueAttributes.TotalMessages()

		if totalMessages > 1 {
			time.Sleep(h.jobCompleteMessageDelay)
			h.log.Infow("found other messages in queue, requeuing job_complete message", "id", *message.MessageId)
			newMessageID := uuid.NewRandom().String()
			if _, err = h.aws.SQS().SendMessage(
				&sqs.SendMessageInput{
					QueueUrl:    &h.config.QueueURL,
					MessageBody: aws.String("job_complete"),
					MessageAttributes: map[string]*sqs.MessageAttributeValue{
						"job_complete": {
							DataType:    aws.String("String"),
							StringValue: aws.String("true"),
						},
						"api_name": {
							DataType:    aws.String("String"),
							StringValue: aws.String(h.config.APIName),
						},
						"job_id": {
							DataType:    aws.String("String"),
							StringValue: aws.String(h.config.JobID),
						},
					},
					MessageDeduplicationId: aws.String(newMessageID),
					MessageGroupId:         aws.String(newMessageID),
				},
			); err != nil {
				return errors.WithStack(err)
			}

			return nil
		}

		if shouldRunOnJobComplete {
			h.log.Infow("processing job_complete message", "id", *message.MessageId)
			return h.submitRequest(*message.Body, true)
		}
		shouldRunOnJobComplete = true

		time.Sleep(h.jobCompleteMessageDelay)
	}
}

func isOnJobCompleteMessage(message *sqs.Message) bool {
	_, found := message.MessageAttributes["job_complete"]