Fix various bugs in the SQS. Implement SQS tests for the

jobs_ok.php/jobs_err.php workers.

Signed-off-by: Valery Piashchynski <[email protected]>

3 files changed, 35 insertions, 9 deletions

diff --git a/plugins/jobs/drivers/sqs/consumer.go b/plugins/jobs/drivers/sqs/consumer.go
index 5d741358..9ce37543 100644
--- a/plugins/jobs/drivers/sqs/consumer.go
+++ b/plugins/jobs/drivers/sqs/consumer.go
@@ -262,9 +262,11 @@ func (j *JobConsumer) Push(ctx context.Context, jb *job.Job) error {
 }
 
 func (j *JobConsumer) handleItem(ctx context.Context, msg *Item) error {
-	// The new value for the message's visibility timeout (in seconds). Values range: 0
-	// to 43200. Maximum: 12 hours.
-	_, err := j.client.SendMessage(ctx, msg.pack(j.queueURL))
+	d, err := msg.pack(j.queueURL)
+	if err != nil {
+		return err
+	}
+	_, err = j.client.SendMessage(ctx, d)
 	if err != nil {
 		return err
 	}
diff --git a/plugins/jobs/drivers/sqs/item.go b/plugins/jobs/drivers/sqs/item.go
index f5fac0b3..eac06731 100644
--- a/plugins/jobs/drivers/sqs/item.go
+++ b/plugins/jobs/drivers/sqs/item.go
@@ -17,6 +17,7 @@ import (
 const (
 	StringType              string = "String"
 	NumberType              string = "Number"
+	BinaryType              string = "Binary"
 	ApproximateReceiveCount string = "ApproximateReceiveCount"
 )
 
@@ -25,6 +26,7 @@ var itemAttributes = []string{
 	job.RRDelay,
 	job.RRTimeout,
 	job.RRPriority,
+	job.RRHeaders,
 }
 
 type Item struct {
@@ -128,7 +130,13 @@ func (i *Item) Ack() error {
 }
 
 func (i *Item) Nack() error {
-	_, err := i.Options.client.DeleteMessage(context.Background(), &sqs.DeleteMessageInput{
+	// requeue message
+	err := i.Options.requeueFn(context.Background(), i)
+	if err != nil {
+		return err
+	}
+
+	_, err = i.Options.client.DeleteMessage(context.Background(), &sqs.DeleteMessageInput{
 		QueueUrl:      i.Options.queue,
 		ReceiptHandle: i.Options.receiptHandler,
 	})
@@ -183,7 +191,13 @@ func fromJob(job *job.Job) *Item {
 	}
 }
 
-func (i *Item) pack(queue *string) *sqs.SendMessageInput {
+func (i *Item) pack(queue *string) (*sqs.SendMessageInput, error) {
+	// pack headers map
+	data, err := json.Marshal(i.Headers)
+	if err != nil {
+		return nil, err
+	}
+
 	return &sqs.SendMessageInput{
 		MessageBody:  aws.String(i.Payload),
 		QueueUrl:     queue,
@@ -192,9 +206,10 @@ func (i *Item) pack(queue *string) *sqs.SendMessageInput {
 			job.RRJob:      {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(i.Job)},
 			job.RRDelay:    {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Delay)))},
 			job.RRTimeout:  {DataType: aws.String(StringType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Timeout)))},
+			job.RRHeaders:  {DataType: aws.String(BinaryType), BinaryValue: data, BinaryListValues: nil, StringListValues: nil, StringValue: nil},
 			job.RRPriority: {DataType: aws.String(NumberType), BinaryValue: nil, BinaryListValues: nil, StringListValues: nil, StringValue: aws.String(strconv.Itoa(int(i.Options.Priority)))},
 		},
-	}
+	}, nil
 }
 
 func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
@@ -210,6 +225,12 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
 		}
 	}
 
+	var h map[string][]string
+	err := json.Unmarshal(msg.MessageAttributes[job.RRHeaders].BinaryValue, &h)
+	if err != nil {
+		return nil, err
+	}
+
 	delay, err := strconv.Atoi(*msg.MessageAttributes[job.RRDelay].StringValue)
 	if err != nil {
 		return nil, errors.E(op, err)
@@ -233,6 +254,7 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
 	item := &Item{
 		Job:     *msg.MessageAttributes[job.RRJob].StringValue,
 		Payload: *msg.Body,
+		Headers: h,
 		Options: &Options{
 			Delay:    int64(delay),
 			Timeout:  int64(to),
@@ -241,7 +263,7 @@ func (j *JobConsumer) unpack(msg *types.Message) (*Item, error) {
 			// private
 			approxReceiveCount: int64(recCount),
 			client:             j.client,
-			queue:              j.queue,
+			queue:              j.queueURL,
 			receiptHandler:     msg.ReceiptHandle,
 			requeueFn:          j.handleItem,
 		},
diff --git a/plugins/jobs/drivers/sqs/listener.go b/plugins/jobs/drivers/sqs/listener.go
index a283b4a3..9efef90d 100644
--- a/plugins/jobs/drivers/sqs/listener.go
+++ b/plugins/jobs/drivers/sqs/listener.go
@@ -30,8 +30,10 @@ func (j *JobConsumer) listen(ctx context.Context) { //nolint:gocognit
 				MaxNumberOfMessages:   j.prefetch,
 				AttributeNames:        []types.QueueAttributeName{types.QueueAttributeName(ApproximateReceiveCount)},
 				MessageAttributeNames: []string{All},
-				VisibilityTimeout:     j.visibilityTimeout,
-				WaitTimeSeconds:       j.waitTime,
+				// The new value for the message's visibility timeout (in seconds). Values range: 0
+				// to 43200. Maximum: 12 hours.
+				VisibilityTimeout: j.visibilityTimeout,
+				WaitTimeSeconds:   j.waitTime,
 			})
 
 			if err != nil {