Delete bulk message

Author: palPalani

src/Sqs/Queue.php

@@ -2,6 +2,7 @@
 
 namespace palPalani\SqsQueueReader\Sqs;
 
+use Aws\Exception\AwsException;
 use Illuminate\Queue\Jobs\SqsJob;
 use Illuminate\Queue\SqsQueue;
 use Illuminate\Support\Facades\Config;
@@ -64,26 +65,31 @@ public function pop($queue = null)
     {
         $queue = $this->getQueue($queue);
 
-        $response = $this->sqs->receiveMessage([
-            'QueueUrl' => $queue,
-            'AttributeNames' => ['ApproximateReceiveCount'],
-            'MaxNumberOfMessages' => 5,
-            'MessageAttributeNames' => ['All'],
-        ]);
-
-        if (isset($response['Messages']) && count($response['Messages']) > 0) {
-            Log::debug('Messages==', [$response['Messages']]);
-            $queueId = explode('/', $queue);
-            $queueId = array_pop($queueId);
-
-            $class = (array_key_exists($queueId, $this->container['config']->get('sqs-queue-reader.handlers')))
-                ? $this->container['config']->get('sqs-queue-reader.handlers')[$queueId]
-                : $this->container['config']->get('sqs-queue-reader.default-handler');
-
-            $response = $this->modifyPayload($response['Messages'], $class);
-            Log::debug('New $responseV2==', [$response]);
-
-            return new SqsJob($this->container, $this->sqs, $response, $this->connectionName, $queue);
+        try {
+            $response = $this->sqs->receiveMessage([
+                'QueueUrl' => $queue,
+                'AttributeNames' => ['ApproximateReceiveCount'],
+                'MaxNumberOfMessages' => 5,
+                'MessageAttributeNames' => ['All'],
+            ]);
+
+            if (isset($response['Messages']) && count($response['Messages']) > 0) {
+                Log::debug('Messages==', [$response['Messages']]);
+                $queueId = explode('/', $queue);
+                $queueId = array_pop($queueId);
+
+                $class = (array_key_exists($queueId, $this->container['config']->get('sqs-queue-reader.handlers')))
+                    ? $this->container['config']->get('sqs-queue-reader.handlers')[$queueId]
+                    : $this->container['config']->get('sqs-queue-reader.default-handler');
+
+                $response = $this->modifyPayload($response['Messages'], $class);
+                Log::debug('New $responseV2==', [$response]);
+
+                return new SqsJob($this->container, $this->sqs, $response, $this->connectionName, $queue);
+            }
+        } catch (AwsException $e) {
+            $msg = 'Line: '. $e->getLine() .', '. $e->getFile() . ', '. $e->getMessage();
+            throw new \RuntimeException("Aws SQS error: " . $msg);
         }
     }
 
@@ -111,9 +117,19 @@ private function modifyPayload($payload, $class)
 
         $body = [];
         $attributes = [];
+        $batchIds = [];
+
         foreach ($payload as $item) {
             //Log::debug('Each Messages==', [$item]);
-            $body[] = json_decode($item['Body'], true);
+            //$body[] = json_decode($item['Body'], true);
+            $body[] = [
+                'messages' => json_decode($item['Body'], true),
+                'attributes' => $item['Attributes'],
+                'batchIds' => [
+                    'Id' => $item['MessageId'],
+                    'ReceiptHandle' => $item['ReceiptHandle'],
+                ]
+            ];
             $attributes = $item['Attributes'];
             $messageId = $item['MessageId'];
             $receiptHandle = $item['ReceiptHandle'];

src/SqsQueueReaderServiceProvider.php

@@ -20,7 +20,28 @@ public function boot(): void
             Queue::after(static function (JobProcessed $event) {
                 $data = $event->job->payload();
                 Log::debug('Job payload==', [$data]);
-                $event->job->delete();
+                //$event->job->delete();
+
+                $batchIds = array_column($data, 'batchIds');
+
+                $batchIds = array_chunk($batchIds, 10);
+
+                foreach ($batchIds as $batch) {
+                    //Deletes up to ten messages from the specified queue.
+                    $result = $event->job->deleteMessageBatch([
+                        'Entries' => $batch,
+                        'QueueUrl' => $event->job->getQueue(),
+                    ]);
+
+                    if ($result['Failed']) {
+                        $msg = '';
+                        foreach ($result['Failed'] as $failed) {
+                            $msg .= sprintf("Deleting message failed, code = %s, id = %s, msg = %s, senderfault = %s", $failed['Code'], $failed['Id'], $failed['Message'], $failed['SenderFault']);
+                        }
+                        Log::error('Cannot delete some SQS messages: ', [$msg]);
+                        throw new \RuntimeException("Cannot delete some messages, consult log for more info!");
+                    }
+                }
             });
         }
     }