azsqs: fix retrying

vmihailenco · vmihailenco · commit 00596eab0ec3 · 2017-11-17T17:09:53.000+02:00
diff --git a/azsqs/queue.go b/azsqs/queue.go
@@ -356,7 +356,7 @@ func (q *Queue) addBatch(msgs []*msgqueue.Message) error {
 
 		body, err := msg.EncodeArgs()
 		if err != nil {
-			internal.Logf("EncodeArgs failed: %s", err)
+			internal.Logf("azsqs: EncodeArgs failed: %s", err)
 			continue
 		}
 		if body == "" {
@@ -384,20 +384,24 @@ func (q *Queue) addBatch(msgs []*msgqueue.Message) error {
 
 	out, err := q.sqs.SendMessageBatch(in)
 	if err != nil {
-		internal.Logf("sqs.SendMessageBatch failed: %s", err)
+		internal.Logf("azsqs: SendMessageBatch failed: %s", err)
 		return err
 	}
 
 	for _, entry := range out.Failed {
-		internal.Logf(
-			"sqs.SendMessageBatch failed with code=%s message=%q",
-			tos(entry.Code), tos(entry.Message),
-		)
 		if entry.SenderFault != nil && *entry.SenderFault {
-			msg := findMessageById(msgs, tos(entry.Id))
-			if msg != nil {
-				msg.Err = fmt.Errorf("%s: %s", tos(entry.Code), tos(entry.Message))
-			}
+			internal.Logf(
+				"azsqs: SendMessageBatch failed with code=%s message=%q",
+				tos(entry.Code), tos(entry.Message),
+			)
+			continue
+		}
+
+		msg := findMessageById(msgs, tos(entry.Id))
+		if msg != nil {
+			msg.Err = fmt.Errorf("%s: %s", tos(entry.Code), tos(entry.Message))
+		} else {
+			internal.Logf("azsqs: can't find message with id=%s", tos(entry.Id))
 		}
 	}
 
@@ -412,13 +416,13 @@ func (q *Queue) splitAddBatch(msgs []*msgqueue.Message) ([]*msgqueue.Message, []
 	for i, msg := range msgs {
 		msg, err := msgutil.UnwrapMessage(msg)
 		if err != nil {
-			internal.Logf(err.Error())
+			internal.Logf("azsqs: UnwrapMessage failed: %s", err)
 			continue
 		}
 
 		body, err := msg.EncodeArgs()
 		if err != nil {
-			internal.Logf("EncodeArgs failed: %s", err)
+			internal.Logf("azsqs: EncodeArgs failed: %s", err)
 			continue
 		}
 
@@ -465,20 +469,24 @@ func (q *Queue) deleteBatch(msgs []*msgqueue.Message) error {
 	}
 	out, err := q.sqs.DeleteMessageBatch(in)
 	if err != nil {
-		internal.Logf("sqs.DeleteMessageBatch failed: %s", err)
+		internal.Logf("azsqs: DeleteMessageBatch failed: %s", err)
 		return err
 	}
 
 	for _, entry := range out.Failed {
-		internal.Logf(
-			"sqs.DeleteMessageBatch failed with code=%s message=%q",
-			tos(entry.Code), tos(entry.Message),
-		)
 		if entry.SenderFault != nil && *entry.SenderFault {
-			msg := findMessageById(msgs, tos(entry.Id))
-			if msg != nil {
-				msg.Err = fmt.Errorf("%s: %s", tos(entry.Code), tos(entry.Message))
-			}
+			internal.Logf(
+				"azsqs: DeleteMessageBatch failed with code=%s message=%q",
+				tos(entry.Code), tos(entry.Message),
+			)
+			continue
+		}
+
+		msg := findMessageById(msgs, tos(entry.Id))
+		if msg != nil {
+			msg.Err = fmt.Errorf("%s: %s", tos(entry.Code), tos(entry.Message))
+		} else {
+			internal.Logf("azsqs: can't find message with id=%s", tos(entry.Id))
 		}
 	}
 	return nil
diff --git a/ironmq/queue.go b/ironmq/queue.go
@@ -305,7 +305,7 @@ func (q *Queue) deleteBatch(msgs []*msgqueue.Message) error {
 		return q.q.DeleteReservedMessages(mqMsgs)
 	})
 	if err != nil {
-		internal.Logf("mq.DeleteReservedMessages failed: %s", err)
+		internal.Logf("ironmq: DeleteReservedMessages failed: %s", err)
 		return err
 	}
 

Original file line number	Diff line number	Diff line change
`@@ -305,7 +305,7 @@ func (q Queue) deleteBatch(msgs []msgqueue.Message) error {`
`305`	`305`	`return q.q.DeleteReservedMessages(mqMsgs)`
`306`	`306`	`})`
`307`	`307`	`if err != nil {`
`308`		`- internal.Logf("mq.DeleteReservedMessages failed: %s", err)`
	`308`	`+ internal.Logf("ironmq: DeleteReservedMessages failed: %s", err)`
`309`	`309`	`return err`
`310`	`310`	`}`
`311`	`311`