StompJob

src/Queue/Jobs/StompJob.php

@@ -10,6 +10,7 @@
 use Illuminate\Queue\Jobs\JobName;
 use Illuminate\Support\Arr;
 use Illuminate\Support\Facades\Event;
+use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Str;
 use Psr\Log\LoggerInterface;
 use Stomp\Transport\Frame;
@@ -105,8 +106,8 @@ public function getJobId()
     public function fire()
     {
         $this->log->info("$this->session [STOMP] Executing event...");
-
         $this->isNativeLaravelJob() ? $this->fireLaravelJob() : $this->fireExternalJob();
+        $this->ackIfNecessary();
     }
 
     protected function isNativeLaravelJob(): bool
@@ -220,15 +221,28 @@ protected function getBackoff(int $attempts): int
      */
     protected function failed($e)
     {
+        $this->ackIfNecessary();
+
         // External events don't have failed method to call.
         if (!$this->payload || !$this->isNativeLaravelJob()) {
             return;
         }
 
         [$class, $method] = JobName::parse($this->payload['job']);
 
-        if (method_exists($this->instance = $this->resolve($class), 'failed')) {
-            $this->instance->failed($this->payload['data'], $e, $this->payload['uuid']);
+        try {
+            if (method_exists($this->instance = $this->resolve($class), 'failed')) {
+                $this->instance->failed($this->payload['data'], $e, $this->payload['uuid']);
+            }
+        } catch (\Exception $e) {
+            Log::error('Exception in job failing: ' . $e->getMessage());
+        }
+    }
+
+    protected function ackIfNecessary()
+    {
+        if (Config::get('consumer_ack_mode') == StompQueue::ACK_MODE_CLIENT && $this->frame) {
+            $this->stompQueue->client->ack($this->frame);
         }
     }
 }