Adds Hyperf\AsyncQueue\Job::fail() method (#6124)

Author: huangdijia

src/Driver/Driver.php

@@ -57,6 +57,7 @@ public function consume(): void
 
         while (ProcessManager::isRunning()) {
             try {
+                /** @var MessageInterface $message */
                 [$data, $message] = $this->pop();
 
                 if ($data === false) {
@@ -95,6 +96,10 @@ protected function checkQueueLength(): void
         }
     }
 
+    /**
+     * @param mixed $data
+     * @param MessageInterface $message
+     */
     protected function getCallback($data, $message): callable
     {
         return function () use ($data, $message) {
@@ -114,6 +119,7 @@ protected function getCallback($data, $message): callable
                     } else {
                         $this->event?->dispatch(new FailedHandle($message, $ex));
                         $this->fail($data);
+                        $message->job()->fail($ex);
                     }
                 }
             }

src/Job.php

@@ -13,11 +13,16 @@
 
 use Hyperf\Contract\CompressInterface;
 use Hyperf\Contract\UnCompressInterface;
+use Throwable;
 
 abstract class Job implements JobInterface, CompressInterface, UnCompressInterface
 {
     protected int $maxAttempts = 0;
 
+    public function fail(Throwable $e): void
+    {
+    }
+
     public function setMaxAttempts(int $maxAttempts): static
     {
         $this->maxAttempts = $maxAttempts;

src/JobInterface.php

@@ -11,8 +11,12 @@
  */
 namespace Hyperf\AsyncQueue;
 
+use Throwable;
+
 interface JobInterface
 {
+    public function fail(Throwable $e): void;
+
     /**
      * Handle the job.
      */