Added result handling for async queue jobs with Hyperf\AsyncQueue\Result enum.

Author: huangdijia

src/Driver/Driver.php

@@ -18,6 +18,7 @@
 use Hyperf\AsyncQueue\Event\QueueLength;
 use Hyperf\AsyncQueue\Event\RetryHandle;
 use Hyperf\AsyncQueue\MessageInterface;
+use Hyperf\AsyncQueue\Result;
 use Hyperf\Codec\Packer\PhpSerializerPacker;
 use Hyperf\Collection\Arr;
 use Hyperf\Contract\PackerInterface;
@@ -112,11 +113,22 @@ protected function getCallback($data, $message): callable
             try {
                 if ($message instanceof MessageInterface) {
                     $this->event?->dispatch(new BeforeHandle($message));
-                    $message->job()->handle();
+
+                    $result = $message->job()->handle();
+
+                    match ($result) {
+                        Result::REQUEUE => $this->remove($data) && $this->retry($data),
+                        Result::RETRY => $this->remove($data) && $message->attempts() && $this->retry($message),
+                        Result::DROP => $this->remove($data),
+                        Result::ACK => $this->ack($data),
+                        default => $this->ack($data),
+                    };
+
                     $this->event?->dispatch(new AfterHandle($message));
+                } else {
+                    // If the message is invalid, just ack it.
+                    $this->ack($data);
                 }
-
-                $this->ack($data);
             } catch (Throwable $ex) {
                 if (isset($message, $data)) {
                     if ($message->attempts() && $this->remove($data)) {

src/JobInterface.php

@@ -20,6 +20,7 @@ public function fail(Throwable $e): void;
 
     /**
      * Handle the job.
+     * @return mixed|never|Result
      */
     public function handle();
 

src/Result.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\AsyncQueue;
+
+enum Result: string
+{
+    /**
+     * Acknowledge the message.
+     */
+    case ACK = 'ack';
+
+    /**
+     * Reject the message and requeue it.
+     */
+    case REQUEUE = 'requeue';
+
+    /**
+     * Retry the message.
+     */
+    case RETRY = 'retry';
+
+    /**
+     * Reject the message and drop it.
+     */
+    case DROP = 'drop';
+}