From 91b975fecbd64f5a4d4070b2c6c37b3800195628 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:03:58 +0800
Subject: [PATCH 01/19] refactor: introduce fluent API for dispatch() helper
 function

This commit refactors the dispatch() helper function to return pending dispatch objects instead of immediately executing the dispatch operation. This enables a more flexible and Laravel-like fluent API for configuring dispatch behavior.

Changes:
- Refactored dispatch() function to return Pending dispatch objects
- Added PendingAsyncQueueDispatch for JobInterface dispatching with chainable methods (onPool, delay, setMaxAttempts)
- Added PendingAmqpProducerMessageDispatch for AMQP message dispatching with chainable methods (onPool, setConfirm, setTimeout)
- Added PendingKafkaProducerMessageDispatch for Kafka message dispatching with chainable methods (onPool, withHeader)
- Removed direct dependencies on DriverFactory, Producer, and ProducerManager from Functions.php
- Improved type safety with enhanced PHPDoc annotations

Benefits:
- Cleaner, more intuitive API: dispatch($job)->onPool('custom')->delay(10)
- Better separation of concerns
- More flexible configuration options
- Follows Laravel's dispatch pattern for familiarity
- Deferred execution via __destruct() allows full configuration before dispatch

Breaking changes: None - the basic dispatch($job) usage remains unchanged, but advanced configurations now use the fluent API instead of positional parameters.
---
 src/helpers/src/Functions.php                 | 30 ++--------
 .../PendingAmqpProducerMessageDispatch.php    | 57 +++++++++++++++++++
 src/helpers/src/PendingAsyncQueueDispatch.php | 56 ++++++++++++++++++
 .../PendingKafkaProducerMessageDispatch.php   | 53 +++++++++++++++++
 4 files changed, 172 insertions(+), 24 deletions(-)
 create mode 100644 src/helpers/src/PendingAmqpProducerMessageDispatch.php
 create mode 100644 src/helpers/src/PendingAsyncQueueDispatch.php
 create mode 100644 src/helpers/src/PendingKafkaProducerMessageDispatch.php

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index 70488500b..786ccc6dd 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -16,10 +16,7 @@
 use Countable;
 use Exception;
 use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
-use FriendsOfHyperf\AsyncTask\TaskInterface as AsyncTaskInterface;
 use Hyperf\Amqp\Message\ProducerMessageInterface;
-use Hyperf\Amqp\Producer;
-use Hyperf\AsyncQueue\Driver\DriverFactory;
 use Hyperf\AsyncQueue\JobInterface;
 use Hyperf\Context\ApplicationContext;
 use Hyperf\Contract\SessionInterface;
@@ -30,7 +27,6 @@
 use Hyperf\HttpMessage\Stream\SwooleStream;
 use Hyperf\HttpServer\Contract\RequestInterface;
 use Hyperf\HttpServer\Contract\ResponseInterface;
-use Hyperf\Kafka\ProducerManager;
 use Hyperf\Logger\LoggerFactory;
 use Hyperf\Stringable\Str;
 use Hyperf\Support\Fluent;
@@ -186,33 +182,19 @@ function di(?string $abstract = null, array $parameters = [])
 }
 
 /**
- * @param AsyncTaskInterface|Closure|JobInterface|ProduceMessage|ProducerMessageInterface|object $job
- * @return bool
+ * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
+ * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
  */
-function dispatch($job, ...$arguments)
+function dispatch($job)
 {
     if ($job instanceof Closure) {
         $job = CallQueuedClosure::create($job);
-        if ($arguments[2] ?? 0) {
-            $job->setMaxAttempts((int) $arguments[2]);
-        }
     }
 
     return match (true) {
-        $job instanceof JobInterface => di(DriverFactory::class)
-            ->get((string) ($arguments[0] ?? (fn () => $this->queue ?? $this->pool ?? 'default')->call($job)))
-            ->push(
-                tap(
-                    $job,
-                    fn ($job) => isset($arguments[2]) && (fn () => $this->maxAttempts = (int) $arguments[2])->call($job)
-                ),
-                (int) ($arguments[1] ?? (fn () => $this->delay ?? 0)->call($job))
-            ),
-        $job instanceof ProducerMessageInterface => di(Producer::class)
-            ->produce($job, ...$arguments),
-        $job instanceof ProduceMessage => di(ProducerManager::class)
-            ->getProducer((string) ($arguments[0] ?? 'default'))
-            ->sendBatch([$job]),
+        $job instanceof ProducerMessageInterface => new PendingAmqpProducerMessageDispatch($job),
+        $job instanceof ProduceMessage => new PendingKafkaProducerMessageDispatch($job),
+        $job instanceof JobInterface => new PendingAsyncQueueDispatch($job),
         default => throw new InvalidArgumentException('Unsupported job type.')
     };
 }
diff --git a/src/helpers/src/PendingAmqpProducerMessageDispatch.php b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
new file mode 100644
index 000000000..4ef272c80
--- /dev/null
+++ b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
@@ -0,0 +1,57 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/components.
+ *
+ * @link     https://github.com/friendsofhyperf/components
+ * @document https://github.com/friendsofhyperf/components/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Helpers;
+
+use Hyperf\Amqp\Message\ProducerMessageInterface;
+use Hyperf\Amqp\Producer;
+use Hyperf\Conditionable\Conditionable;
+use Hyperf\Context\ApplicationContext;
+
+class PendingAmqpProducerMessageDispatch
+{
+    use Conditionable;
+
+    public string $pool = 'default';
+
+    public int $timeout = 5;
+
+    public bool $confirm = false;
+
+    public function __construct(protected ProducerMessageInterface $message)
+    {
+    }
+
+    public function __destruct()
+    {
+        ApplicationContext::getContainer()
+            ->get(Producer::class)
+            ->produce($this->message, $this->confirm, $this->timeout);
+    }
+
+    public function onPool(string $pool): static
+    {
+        $this->pool = $pool;
+        return $this;
+    }
+
+    public function setConfirm(bool $confirm): static
+    {
+        $this->confirm = $confirm;
+        return $this;
+    }
+
+    public function setTimeout(int $timeout): static
+    {
+        $this->timeout = $timeout;
+        return $this;
+    }
+}
diff --git a/src/helpers/src/PendingAsyncQueueDispatch.php b/src/helpers/src/PendingAsyncQueueDispatch.php
new file mode 100644
index 000000000..9da20c5d1
--- /dev/null
+++ b/src/helpers/src/PendingAsyncQueueDispatch.php
@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/components.
+ *
+ * @link     https://github.com/friendsofhyperf/components
+ * @document https://github.com/friendsofhyperf/components/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Helpers;
+
+use Hyperf\AsyncQueue\Driver\DriverFactory;
+use Hyperf\AsyncQueue\JobInterface;
+use Hyperf\Conditionable\Conditionable;
+use Hyperf\Context\ApplicationContext;
+
+class PendingAsyncQueueDispatch
+{
+    use Conditionable;
+
+    public string $pool = 'default';
+
+    public int $delay = 0;
+
+    public function __construct(protected JobInterface $job)
+    {
+    }
+
+    public function __destruct()
+    {
+        ApplicationContext::getContainer()
+            ->get(DriverFactory::class)
+            ->get($this->pool)
+            ->push($this->job, $this->delay);
+    }
+
+    public function setMaxAttempts(int $maxAttempts): static
+    {
+        $this->job->setMaxAttempts($maxAttempts);
+        return $this;
+    }
+
+    public function onPool(string $pool): static
+    {
+        $this->pool = $pool;
+        return $this;
+    }
+
+    public function delay(int $delay): static
+    {
+        $this->delay = $delay;
+        return $this;
+    }
+}
diff --git a/src/helpers/src/PendingKafkaProducerMessageDispatch.php b/src/helpers/src/PendingKafkaProducerMessageDispatch.php
new file mode 100644
index 000000000..16d1d9019
--- /dev/null
+++ b/src/helpers/src/PendingKafkaProducerMessageDispatch.php
@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/components.
+ *
+ * @link     https://github.com/friendsofhyperf/components
+ * @document https://github.com/friendsofhyperf/components/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Helpers;
+
+use Hyperf\Conditionable\Conditionable;
+use Hyperf\Context\ApplicationContext;
+use Hyperf\Kafka\ProducerManager;
+use longlang\phpkafka\Producer\ProduceMessage;
+use longlang\phpkafka\Protocol\RecordBatch\RecordHeader;
+
+/**
+ * @property array $headers
+ */
+class PendingKafkaProducerMessageDispatch
+{
+    use Conditionable;
+
+    public string $pool = 'default';
+
+    public function __construct(protected ProduceMessage $message)
+    {
+    }
+
+    public function __destruct()
+    {
+        ApplicationContext::getContainer()
+            ->get(ProducerManager::class)
+            ->getProducer($this->pool)
+            ->sendBatch([$this->message]);
+    }
+
+    public function onPool(string $pool): static
+    {
+        $this->pool = $pool;
+        return $this;
+    }
+
+    public function withHeader(string $key, string $value): static
+    {
+        $header = (new RecordHeader())->setHeaderKey($key)->setValue($value);
+        (fn () => $this->headers[] = $header)->call($this->message);
+        return $this;
+    }
+}

From 097eed9b30a6eb383dbc9b0f7abb9dc0af41b707 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:06:10 +0800
Subject: [PATCH 02/19] docs: add warning about return value assignment in
 dispatch() function

---
 src/helpers/src/Functions.php | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index 786ccc6dd..877d43a56 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -182,6 +182,7 @@ function di(?string $abstract = null, array $parameters = [])
 }
 
 /**
+ * Do not assign a value to the return value of this function unless you are very clear about the consequences of doing so.
  * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
  * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
  */

From bfd98db6a020fd19f99dfd34a638f7c3d52c92a5 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:11:43 +0800
Subject: [PATCH 03/19] test: add comprehensive tests for fluent dispatch API

Added extensive test coverage for the new fluent dispatch API including:

Test Coverage:
- dispatch() function type detection and routing
  - Closure wrapping in CallQueuedClosure
  - JobInterface handling
  - AMQP ProducerMessage handling
  - Kafka ProduceMessage handling
  - Invalid type rejection

- PendingAsyncQueueDispatch tests
  - onPool() method chaining
  - delay() configuration
  - setMaxAttempts() integration
  - Fluent API chaining
  - Destruct execution verification
  - Conditionable trait (when/unless)

- PendingAmqpProducerMessageDispatch tests
  - onPool() configuration
  - setConfirm() method
  - setTimeout() method
  - Fluent API chaining
  - Destruct execution verification
  - Conditionable trait support

- PendingKafkaProducerMessageDispatch tests
  - onPool() configuration
  - withHeader() method
  - Multiple header support
  - Fluent API chaining
  - Destruct execution verification
  - Conditionable trait support

- Integration tests
  - Backward compatibility with basic dispatch
  - Error handling and exception propagation

All 24 tests passing with proper mocking of:
- DriverFactory and AsyncQueue Driver
- AMQP Producer
- Kafka ProducerManager and Producer
- ApplicationContext container setup
---
 tests/Helpers/DispatchTest.php | 538 +++++++++++++++++++++++++++++++++
 1 file changed, 538 insertions(+)
 create mode 100644 tests/Helpers/DispatchTest.php

diff --git a/tests/Helpers/DispatchTest.php b/tests/Helpers/DispatchTest.php
new file mode 100644
index 000000000..f7af3404c
--- /dev/null
+++ b/tests/Helpers/DispatchTest.php
@@ -0,0 +1,538 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of friendsofhyperf/components.
+ *
+ * @link     https://github.com/friendsofhyperf/components
+ * @document https://github.com/friendsofhyperf/components/blob/main/README.md
+ * @contact  huangdijia@gmail.com
+ */
+
+namespace FriendsOfHyperf\Tests\Helpers;
+
+use Exception;
+use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
+use FriendsOfHyperf\Helpers\PendingAmqpProducerMessageDispatch;
+use FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch;
+use FriendsOfHyperf\Helpers\PendingKafkaProducerMessageDispatch;
+use FriendsOfHyperf\Tests\TestCase;
+use Hyperf\Amqp\Message\ProducerMessage;
+use Hyperf\Amqp\Producer;
+use Hyperf\AsyncQueue\Driver\Driver;
+use Hyperf\AsyncQueue\Driver\DriverFactory;
+use Hyperf\AsyncQueue\JobInterface;
+use Hyperf\Context\ApplicationContext;
+use Hyperf\Kafka\Producer as HyperfKafkaProducer;
+use Hyperf\Kafka\ProducerManager;
+use longlang\phpkafka\Producer\ProduceMessage;
+use longlang\phpkafka\Producer\Producer as KafkaProducer;
+use Mockery as m;
+use Psr\Container\ContainerInterface;
+use ReflectionClass;
+
+use function FriendsOfHyperf\Helpers\dispatch;
+
+/**
+ * @internal
+ */
+#[\PHPUnit\Framework\Attributes\Group('helpers')]
+class DispatchTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        $this->setupDefaultContainerMock();
+    }
+
+    protected function tearDown(): void
+    {
+        parent::tearDown();
+        m::close();
+    }
+
+    protected function setupDefaultContainerMock(): void
+    {
+        $container = m::mock(ContainerInterface::class);
+        $driverFactory = m::mock(DriverFactory::class);
+        $driver = m::mock(Driver::class);
+        $producer = m::mock(Producer::class);
+        $producerManager = m::mock(ProducerManager::class);
+        $kafkaProducer = m::mock(HyperfKafkaProducer::class);
+
+        // Setup for AsyncQueue dispatch
+        $container->shouldReceive('get')
+            ->with(DriverFactory::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($driverFactory);
+
+        $driverFactory->shouldReceive('get')
+            ->withAnyArgs()
+            ->zeroOrMoreTimes()
+            ->andReturn($driver);
+
+        $driver->shouldReceive('push')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        // Setup for AMQP dispatch
+        $container->shouldReceive('get')
+            ->with(Producer::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($producer);
+
+        $producer->shouldReceive('produce')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        // Setup for Kafka dispatch
+        $container->shouldReceive('get')
+            ->with(ProducerManager::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($producerManager);
+
+        $producerManager->shouldReceive('getProducer')
+            ->withAnyArgs()
+            ->zeroOrMoreTimes()
+            ->andReturn($kafkaProducer);
+
+        $kafkaProducer->shouldReceive('sendBatch')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        ApplicationContext::setContainer($container);
+    }
+
+    public function testDispatchWithClosure()
+    {
+        $closure = function () {
+            return 'test';
+        };
+
+        $result = dispatch($closure);
+
+        $this->assertInstanceOf(PendingAsyncQueueDispatch::class, $result);
+
+        // Verify the underlying job is CallQueuedClosure
+        $job = $this->getProperty($result, 'job');
+        $this->assertInstanceOf(CallQueuedClosure::class, $job);
+    }
+
+    public function testDispatchWithJobInterface()
+    {
+        $job = m::mock(JobInterface::class);
+
+        $result = dispatch($job);
+
+        $this->assertInstanceOf(PendingAsyncQueueDispatch::class, $result);
+        $this->assertSame($job, $this->getProperty($result, 'job'));
+    }
+
+    public function testDispatchWithProducerMessage()
+    {
+        $message = m::mock(ProducerMessage::class);
+
+        $result = dispatch($message);
+
+        $this->assertInstanceOf(PendingAmqpProducerMessageDispatch::class, $result);
+        $this->assertSame($message, $this->getProperty($result, 'message'));
+    }
+
+    public function testDispatchWithKafkaProduceMessage()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+
+        $result = dispatch($message);
+
+        $this->assertInstanceOf(PendingKafkaProducerMessageDispatch::class, $result);
+        $this->assertSame($message, $this->getProperty($result, 'message'));
+    }
+
+    public function testDispatchWithInvalidType()
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('Unsupported job type.');
+
+        dispatch(new \stdClass());
+    }
+
+    public function testPendingAsyncQueueDispatchOnPool()
+    {
+        $job = m::mock(JobInterface::class);
+        $pending = dispatch($job);
+
+        $result = $pending->onPool('custom-pool');
+
+        $this->assertSame($pending, $result);
+        $this->assertEquals('custom-pool', $this->getProperty($pending, 'pool'));
+    }
+
+    public function testPendingAsyncQueueDispatchDelay()
+    {
+        $job = m::mock(JobInterface::class);
+        $pending = dispatch($job);
+
+        $result = $pending->delay(60);
+
+        $this->assertSame($pending, $result);
+        $this->assertEquals(60, $this->getProperty($pending, 'delay'));
+    }
+
+    public function testPendingAsyncQueueDispatchSetMaxAttempts()
+    {
+        $job = m::mock(JobInterface::class);
+        $job->shouldReceive('setMaxAttempts')
+            ->with(5)
+            ->once()
+            ->andReturnSelf();
+
+        $pending = dispatch($job);
+
+        $result = $pending->setMaxAttempts(5);
+
+        $this->assertSame($pending, $result);
+    }
+
+    public function testPendingAsyncQueueDispatchFluentChaining()
+    {
+        $job = m::mock(JobInterface::class);
+        $job->shouldReceive('setMaxAttempts')
+            ->with(3)
+            ->once()
+            ->andReturnSelf();
+
+        $pending = dispatch($job)
+            ->onPool('high-priority')
+            ->delay(30)
+            ->setMaxAttempts(3);
+
+        $this->assertEquals('high-priority', $this->getProperty($pending, 'pool'));
+        $this->assertEquals(30, $this->getProperty($pending, 'delay'));
+    }
+
+    public function testPendingAsyncQueueDispatchExecutesOnDestruct()
+    {
+        $job = m::mock(JobInterface::class);
+        $pushed = false;
+
+        $container = m::mock(ContainerInterface::class);
+        $driverFactory = m::mock(DriverFactory::class);
+        $driver = m::mock(Driver::class);
+
+        $container->shouldReceive('get')
+            ->with(DriverFactory::class)
+            ->once()
+            ->andReturn($driverFactory);
+
+        $driverFactory->shouldReceive('get')
+            ->with('test-pool')
+            ->once()
+            ->andReturn($driver);
+
+        $driver->shouldReceive('push')
+            ->with($job, 10)
+            ->once()
+            ->andReturnUsing(function () use (&$pushed) {
+                $pushed = true;
+                return true;
+            });
+
+        ApplicationContext::setContainer($container);
+
+        $pending = dispatch($job)
+            ->onPool('test-pool')
+            ->delay(10);
+
+        // Trigger destruct
+        unset($pending);
+
+        $this->assertTrue($pushed, 'Job should have been pushed to queue');
+    }
+
+    public function testPendingAsyncQueueDispatchWithConditionable()
+    {
+        $job = m::mock(JobInterface::class);
+
+        $pending = dispatch($job)
+            ->when(true, function ($dispatch) {
+                $dispatch->onPool('conditional-pool');
+            })
+            ->unless(false, function ($dispatch) {
+                $dispatch->delay(20);
+            });
+
+        $this->assertEquals('conditional-pool', $this->getProperty($pending, 'pool'));
+        $this->assertEquals(20, $this->getProperty($pending, 'delay'));
+    }
+
+    public function testPendingAmqpProducerMessageDispatchOnPool()
+    {
+        $message = m::mock(ProducerMessage::class);
+        $pending = dispatch($message);
+
+        $result = $pending->onPool('amqp-custom');
+
+        $this->assertSame($pending, $result);
+        $this->assertEquals('amqp-custom', $this->getProperty($pending, 'pool'));
+    }
+
+    public function testPendingAmqpProducerMessageDispatchSetConfirm()
+    {
+        $message = m::mock(ProducerMessage::class);
+        $pending = dispatch($message);
+
+        $result = $pending->setConfirm(true);
+
+        $this->assertSame($pending, $result);
+        $this->assertTrue($this->getProperty($pending, 'confirm'));
+    }
+
+    public function testPendingAmqpProducerMessageDispatchSetTimeout()
+    {
+        $message = m::mock(ProducerMessage::class);
+        $pending = dispatch($message);
+
+        $result = $pending->setTimeout(10);
+
+        $this->assertSame($pending, $result);
+        $this->assertEquals(10, $this->getProperty($pending, 'timeout'));
+    }
+
+    public function testPendingAmqpProducerMessageDispatchFluentChaining()
+    {
+        $message = m::mock(ProducerMessage::class);
+
+        $pending = dispatch($message)
+            ->onPool('amqp-pool')
+            ->setConfirm(true)
+            ->setTimeout(15);
+
+        $this->assertEquals('amqp-pool', $this->getProperty($pending, 'pool'));
+        $this->assertTrue($this->getProperty($pending, 'confirm'));
+        $this->assertEquals(15, $this->getProperty($pending, 'timeout'));
+    }
+
+    public function testPendingAmqpProducerMessageDispatchExecutesOnDestruct()
+    {
+        $message = m::mock(ProducerMessage::class);
+        $produced = false;
+
+        $container = m::mock(ContainerInterface::class);
+        $producer = m::mock(Producer::class);
+
+        $container->shouldReceive('get')
+            ->with(Producer::class)
+            ->once()
+            ->andReturn($producer);
+
+        $producer->shouldReceive('produce')
+            ->with($message, true, 10)
+            ->once()
+            ->andReturnUsing(function () use (&$produced) {
+                $produced = true;
+                return true;
+            });
+
+        ApplicationContext::setContainer($container);
+
+        $pending = dispatch($message)
+            ->setConfirm(true)
+            ->setTimeout(10);
+
+        // Trigger destruct
+        unset($pending);
+
+        $this->assertTrue($produced, 'Message should have been produced');
+    }
+
+    public function testPendingAmqpProducerMessageDispatchWithConditionable()
+    {
+        $message = m::mock(ProducerMessage::class);
+
+        $pending = dispatch($message)
+            ->when(true, function ($dispatch) {
+                $dispatch->setConfirm(true);
+            })
+            ->unless(false, function ($dispatch) {
+                $dispatch->setTimeout(20);
+            });
+
+        $this->assertTrue($this->getProperty($pending, 'confirm'));
+        $this->assertEquals(20, $this->getProperty($pending, 'timeout'));
+    }
+
+    public function testPendingKafkaProducerMessageDispatchOnPool()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+        $pending = dispatch($message);
+
+        $result = $pending->onPool('kafka-custom');
+
+        $this->assertSame($pending, $result);
+        $this->assertEquals('kafka-custom', $this->getProperty($pending, 'pool'));
+    }
+
+    public function testPendingKafkaProducerMessageDispatchWithHeader()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+        $pending = dispatch($message);
+
+        $result = $pending->withHeader('trace-id', '12345');
+
+        $this->assertSame($pending, $result);
+
+        // Verify header was added to the message
+        $headers = $this->getProperty($message, 'headers');
+        $this->assertIsArray($headers);
+        $this->assertNotEmpty($headers);
+    }
+
+    public function testPendingKafkaProducerMessageDispatchFluentChaining()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+
+        $pending = dispatch($message)
+            ->onPool('kafka-pool')
+            ->withHeader('user-id', '123')
+            ->withHeader('request-id', 'abc');
+
+        $this->assertEquals('kafka-pool', $this->getProperty($pending, 'pool'));
+
+        // Verify both headers were added
+        $headers = $this->getProperty($message, 'headers');
+        $this->assertIsArray($headers);
+        $this->assertCount(2, $headers);
+    }
+
+    public function testPendingKafkaProducerMessageDispatchExecutesOnDestruct()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+        $sent = false;
+
+        $container = m::mock(ContainerInterface::class);
+        $producerManager = m::mock(ProducerManager::class);
+        $kafkaProducer = m::mock(HyperfKafkaProducer::class);
+
+        $container->shouldReceive('get')
+            ->with(ProducerManager::class)
+            ->once()
+            ->andReturn($producerManager);
+
+        $producerManager->shouldReceive('getProducer')
+            ->with('kafka-pool')
+            ->once()
+            ->andReturn($kafkaProducer);
+
+        $kafkaProducer->shouldReceive('sendBatch')
+            ->with([$message])
+            ->once()
+            ->andReturnUsing(function () use (&$sent) {
+                $sent = true;
+                return true;
+            });
+
+        ApplicationContext::setContainer($container);
+
+        $pending = dispatch($message)
+            ->onPool('kafka-pool');
+
+        // Trigger destruct
+        unset($pending);
+
+        $this->assertTrue($sent, 'Message should have been sent');
+    }
+
+    public function testPendingKafkaProducerMessageDispatchWithConditionable()
+    {
+        $message = new ProduceMessage('test-topic', 'test-value');
+
+        $pending = dispatch($message)
+            ->when(true, function ($dispatch) {
+                $dispatch->withHeader('conditional', 'true');
+            })
+            ->unless(false, function ($dispatch) {
+                $dispatch->onPool('conditional-pool');
+            });
+
+        $this->assertEquals('conditional-pool', $this->getProperty($pending, 'pool'));
+
+        $headers = $this->getProperty($message, 'headers');
+        $this->assertNotEmpty($headers);
+    }
+
+    public function testBackwardCompatibilityWithBasicDispatch()
+    {
+        $job = m::mock(JobInterface::class);
+
+        $container = m::mock(ContainerInterface::class);
+        $driverFactory = m::mock(DriverFactory::class);
+        $driver = m::mock(Driver::class);
+
+        $container->shouldReceive('get')
+            ->with(DriverFactory::class)
+            ->once()
+            ->andReturn($driverFactory);
+
+        $driverFactory->shouldReceive('get')
+            ->with('default')
+            ->once()
+            ->andReturn($driver);
+
+        $driver->shouldReceive('push')
+            ->with($job, 0)
+            ->once()
+            ->andReturn(true);
+
+        ApplicationContext::setContainer($container);
+
+        // Test basic dispatch without any configuration
+        $pending = dispatch($job);
+
+        // Verify defaults
+        $this->assertEquals('default', $this->getProperty($pending, 'pool'));
+        $this->assertEquals(0, $this->getProperty($pending, 'delay'));
+
+        // Trigger destruct
+        unset($pending);
+    }
+
+    public function testDispatchWithErrorHandling()
+    {
+        $job = m::mock(JobInterface::class);
+
+        $container = m::mock(ContainerInterface::class);
+        $driverFactory = m::mock(DriverFactory::class);
+
+        $container->shouldReceive('get')
+            ->with(DriverFactory::class)
+            ->once()
+            ->andReturn($driverFactory);
+
+        $driverFactory->shouldReceive('get')
+            ->with('default')
+            ->once()
+            ->andThrow(new Exception('Driver not found'));
+
+        ApplicationContext::setContainer($container);
+
+        $this->expectException(Exception::class);
+        $this->expectExceptionMessage('Driver not found');
+
+        $pending = dispatch($job);
+
+        // Trigger destruct which should throw
+        unset($pending);
+    }
+
+    /**
+     * Helper method to get protected/private property value.
+     */
+    protected function getProperty(object $object, string $property): mixed
+    {
+        $reflection = new ReflectionClass($object);
+        $prop = $reflection->getProperty($property);
+        $prop->setAccessible(true);
+
+        return $prop->getValue($object);
+    }
+}

From ab8931a6e1ce3a4b2ab425b2e3492ea99d6c7333 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:12:28 +0800
Subject: [PATCH 04/19] test: refactor DispatchTest to improve container mock
 setup and exception handling

---
 tests/Helpers/DispatchTest.php | 111 +++++++++++++++++----------------
 1 file changed, 56 insertions(+), 55 deletions(-)

diff --git a/tests/Helpers/DispatchTest.php b/tests/Helpers/DispatchTest.php
index f7af3404c..6921e96a1 100644
--- a/tests/Helpers/DispatchTest.php
+++ b/tests/Helpers/DispatchTest.php
@@ -25,11 +25,12 @@
 use Hyperf\Context\ApplicationContext;
 use Hyperf\Kafka\Producer as HyperfKafkaProducer;
 use Hyperf\Kafka\ProducerManager;
+use InvalidArgumentException;
 use longlang\phpkafka\Producer\ProduceMessage;
-use longlang\phpkafka\Producer\Producer as KafkaProducer;
 use Mockery as m;
 use Psr\Container\ContainerInterface;
 use ReflectionClass;
+use stdClass;
 
 use function FriendsOfHyperf\Helpers\dispatch;
 
@@ -51,58 +52,6 @@ protected function tearDown(): void
         m::close();
     }
 
-    protected function setupDefaultContainerMock(): void
-    {
-        $container = m::mock(ContainerInterface::class);
-        $driverFactory = m::mock(DriverFactory::class);
-        $driver = m::mock(Driver::class);
-        $producer = m::mock(Producer::class);
-        $producerManager = m::mock(ProducerManager::class);
-        $kafkaProducer = m::mock(HyperfKafkaProducer::class);
-
-        // Setup for AsyncQueue dispatch
-        $container->shouldReceive('get')
-            ->with(DriverFactory::class)
-            ->zeroOrMoreTimes()
-            ->andReturn($driverFactory);
-
-        $driverFactory->shouldReceive('get')
-            ->withAnyArgs()
-            ->zeroOrMoreTimes()
-            ->andReturn($driver);
-
-        $driver->shouldReceive('push')
-            ->zeroOrMoreTimes()
-            ->andReturnTrue();
-
-        // Setup for AMQP dispatch
-        $container->shouldReceive('get')
-            ->with(Producer::class)
-            ->zeroOrMoreTimes()
-            ->andReturn($producer);
-
-        $producer->shouldReceive('produce')
-            ->zeroOrMoreTimes()
-            ->andReturnTrue();
-
-        // Setup for Kafka dispatch
-        $container->shouldReceive('get')
-            ->with(ProducerManager::class)
-            ->zeroOrMoreTimes()
-            ->andReturn($producerManager);
-
-        $producerManager->shouldReceive('getProducer')
-            ->withAnyArgs()
-            ->zeroOrMoreTimes()
-            ->andReturn($kafkaProducer);
-
-        $kafkaProducer->shouldReceive('sendBatch')
-            ->zeroOrMoreTimes()
-            ->andReturnTrue();
-
-        ApplicationContext::setContainer($container);
-    }
-
     public function testDispatchWithClosure()
     {
         $closure = function () {
@@ -150,10 +99,10 @@ public function testDispatchWithKafkaProduceMessage()
 
     public function testDispatchWithInvalidType()
     {
-        $this->expectException(\InvalidArgumentException::class);
+        $this->expectException(InvalidArgumentException::class);
         $this->expectExceptionMessage('Unsupported job type.');
 
-        dispatch(new \stdClass());
+        dispatch(new stdClass());
     }
 
     public function testPendingAsyncQueueDispatchOnPool()
@@ -524,6 +473,58 @@ public function testDispatchWithErrorHandling()
         unset($pending);
     }
 
+    protected function setupDefaultContainerMock(): void
+    {
+        $container = m::mock(ContainerInterface::class);
+        $driverFactory = m::mock(DriverFactory::class);
+        $driver = m::mock(Driver::class);
+        $producer = m::mock(Producer::class);
+        $producerManager = m::mock(ProducerManager::class);
+        $kafkaProducer = m::mock(HyperfKafkaProducer::class);
+
+        // Setup for AsyncQueue dispatch
+        $container->shouldReceive('get')
+            ->with(DriverFactory::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($driverFactory);
+
+        $driverFactory->shouldReceive('get')
+            ->withAnyArgs()
+            ->zeroOrMoreTimes()
+            ->andReturn($driver);
+
+        $driver->shouldReceive('push')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        // Setup for AMQP dispatch
+        $container->shouldReceive('get')
+            ->with(Producer::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($producer);
+
+        $producer->shouldReceive('produce')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        // Setup for Kafka dispatch
+        $container->shouldReceive('get')
+            ->with(ProducerManager::class)
+            ->zeroOrMoreTimes()
+            ->andReturn($producerManager);
+
+        $producerManager->shouldReceive('getProducer')
+            ->withAnyArgs()
+            ->zeroOrMoreTimes()
+            ->andReturn($kafkaProducer);
+
+        $kafkaProducer->shouldReceive('sendBatch')
+            ->zeroOrMoreTimes()
+            ->andReturnTrue();
+
+        ApplicationContext::setContainer($container);
+    }
+
     /**
      * Helper method to get protected/private property value.
      */

From 1c87ac23ab439652abc94aa081db3ff98d2882d9 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:35:56 +0800
Subject: [PATCH 05/19] Update dispatch() return type in test

Changed the expected return type of dispatch() from bool to PendingAsyncQueueDispatch in the test to reflect updated behavior.
---
 types/Helpers/Functions.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/types/Helpers/Functions.php b/types/Helpers/Functions.php
index db65a9583..d2f517c0b 100644
--- a/types/Helpers/Functions.php
+++ b/types/Helpers/Functions.php
@@ -150,7 +150,7 @@
 
 // dispatch() tests - returns bool
 // Note: dispatch() has complex return types based on job type, testing the common case
-assertType('bool', dispatch(new class implements Hyperf\AsyncQueue\JobInterface {
+assertType('FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch', dispatch(new class implements Hyperf\AsyncQueue\JobInterface {
     public function handle(): void
     {
     }

From 69813b43426a87ea2d38d0813316168e34d6a2bb Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 22:38:04 +0800
Subject: [PATCH 06/19] test: add type assertion for dispatch() with closure

---
 types/Helpers/Functions.php | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/types/Helpers/Functions.php b/types/Helpers/Functions.php
index d2f517c0b..0523660c5 100644
--- a/types/Helpers/Functions.php
+++ b/types/Helpers/Functions.php
@@ -175,6 +175,9 @@ public function getQueueName(): string
     }
 }));
 
+assertType('FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch', dispatch(function () {
+}));
+
 // info() tests
 assertType('mixed', info('message'));
 

From e6891a1bc8f7dd622441cf931dd76e6a039ca23d Mon Sep 17 00:00:00 2001
From: Deeka Wong <huangdijia@gmail.com>
Date: Thu, 13 Nov 2025 22:58:13 +0800
Subject: [PATCH 07/19] Apply suggestion from @huangdijia

---
 src/helpers/src/PendingAmqpProducerMessageDispatch.php | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/helpers/src/PendingAmqpProducerMessageDispatch.php b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
index 4ef272c80..ef8550a77 100644
--- a/src/helpers/src/PendingAmqpProducerMessageDispatch.php
+++ b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
@@ -20,7 +20,7 @@ class PendingAmqpProducerMessageDispatch
 {
     use Conditionable;
 
-    public string $pool = 'default';
+    public ?string $pool = null;
 
     public int $timeout = 5;
 
@@ -32,6 +32,7 @@ public function __construct(protected ProducerMessageInterface $message)
 
     public function __destruct()
     {
+        $this->pool && $this->message->setPoolName($this->pool);
         ApplicationContext::getContainer()
             ->get(Producer::class)
             ->produce($this->message, $this->confirm, $this->timeout);

From e5573f712b7813016f5af77f7832edef9ef90a14 Mon Sep 17 00:00:00 2001
From: Deeka Wong <huangdijia@gmail.com>
Date: Thu, 13 Nov 2025 23:01:24 +0800
Subject: [PATCH 08/19] Apply suggestion from @huangdijia

---
 src/helpers/src/PendingAmqpProducerMessageDispatch.php | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/helpers/src/PendingAmqpProducerMessageDispatch.php b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
index ef8550a77..65303dbc1 100644
--- a/src/helpers/src/PendingAmqpProducerMessageDispatch.php
+++ b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
@@ -44,6 +44,12 @@ public function onPool(string $pool): static
         return $this;
     }
 
+    public function setPayload(mixed $data): static
+    {
+        $this->message->setPayload($data);
+        return $this;
+    }
+
     public function setConfirm(bool $confirm): static
     {
         $this->confirm = $confirm;

From 16dbeb2f1aefa3482f1dfe4bba8a221866dc5ab8 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 23:11:25 +0800
Subject: [PATCH 09/19] feat: add withHeader method to
 PendingAmqpProducerMessageDispatch for setting application headers

---
 .../src/PendingAmqpProducerMessageDispatch.php       | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/src/helpers/src/PendingAmqpProducerMessageDispatch.php b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
index 65303dbc1..91f3992af 100644
--- a/src/helpers/src/PendingAmqpProducerMessageDispatch.php
+++ b/src/helpers/src/PendingAmqpProducerMessageDispatch.php
@@ -16,6 +16,9 @@
 use Hyperf\Conditionable\Conditionable;
 use Hyperf\Context\ApplicationContext;
 
+/**
+ * @property array{application_headers?:AMQPTable} $properties
+ */
 class PendingAmqpProducerMessageDispatch
 {
     use Conditionable;
@@ -50,6 +53,15 @@ public function setPayload(mixed $data): static
         return $this;
     }
 
+    public function withHeader(string $key, mixed $value, ?int $ttl = null): static
+    {
+        (function () use ($key, $value, $ttl) {
+            $this->properties['application_headers'] ??= new \PhpAmqpLib\Wire\AMQPTable(); // @phpstan-ignore-line
+            $this->properties['application_headers']->set($key, $value, $ttl);
+        })->call($this->message);
+        return $this;
+    }
+
     public function setConfirm(bool $confirm): static
     {
         $this->confirm = $confirm;

From 1127b228cf5057c0ed4cc80130ced5d1f9016d79 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 23:14:19 +0800
Subject: [PATCH 10/19] feat: add setKey and setValue methods to
 PendingKafkaProducerMessageDispatch for setting message properties

---
 .../src/PendingKafkaProducerMessageDispatch.php    | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/src/helpers/src/PendingKafkaProducerMessageDispatch.php b/src/helpers/src/PendingKafkaProducerMessageDispatch.php
index 16d1d9019..44a95dc15 100644
--- a/src/helpers/src/PendingKafkaProducerMessageDispatch.php
+++ b/src/helpers/src/PendingKafkaProducerMessageDispatch.php
@@ -19,6 +19,8 @@
 
 /**
  * @property array $headers
+ * @property null|string $key
+ * @property null|string $value
  */
 class PendingKafkaProducerMessageDispatch
 {
@@ -44,6 +46,18 @@ public function onPool(string $pool): static
         return $this;
     }
 
+    public function setKey(string $key): static
+    {
+        (fn () => $this->key = $key)->call($this->message);
+        return $this;
+    }
+
+    public function setValue(string $value): static
+    {
+        (fn () => $this->value = $value)->call($this->message);
+        return $this;
+    }
+
     public function withHeader(string $key, string $value): static
     {
         $header = (new RecordHeader())->setHeaderKey($key)->setValue($value);

From cb2ca093dd79bff633d13084b39676f695d31d75 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Thu, 13 Nov 2025 23:18:05 +0800
Subject: [PATCH 11/19] feat: enhance dispatch function to check class
 existence before type matching for job dispatching

---
 src/helpers/src/Functions.php | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index 877d43a56..44592a4c2 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -185,6 +185,7 @@ function di(?string $abstract = null, array $parameters = [])
  * Do not assign a value to the return value of this function unless you are very clear about the consequences of doing so.
  * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
  * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
+ * @throws InvalidArgumentException
  */
 function dispatch($job)
 {
@@ -193,9 +194,9 @@ function dispatch($job)
     }
 
     return match (true) {
-        $job instanceof ProducerMessageInterface => new PendingAmqpProducerMessageDispatch($job),
-        $job instanceof ProduceMessage => new PendingKafkaProducerMessageDispatch($job),
-        $job instanceof JobInterface => new PendingAsyncQueueDispatch($job),
+        interface_exists(ProducerMessageInterface::class) && $job instanceof ProducerMessageInterface => new PendingAmqpProducerMessageDispatch($job),
+        class_exists(ProduceMessage::class) && $job instanceof ProduceMessage => new PendingKafkaProducerMessageDispatch($job),
+        interface_exists(JobInterface::class) && $job instanceof JobInterface => new PendingAsyncQueueDispatch($job),
         default => throw new InvalidArgumentException('Unsupported job type.')
     };
 }

From dee0cf022392c289a2576d3ebb3e60f62f7ab7f4 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:49:06 +0800
Subject: [PATCH 12/19] Refactor dispatching mechanism and remove deprecated
 classes

- Deleted `PendingAsyncQueueDispatch`, `PendingKafkaProducerMessageDispatch`, and `DispatchTest` classes.
- Introduced new implementations for `PendingAsyncQueueDispatch`, `PendingKafkaProducerMessageDispatch`, and `PendingAmqpProducerMessageDispatch` with enhanced functionality.
- Updated the `dispatch` function to accommodate new classes and ensure compatibility with various job types.
- Added comprehensive tests for the new dispatching classes to ensure expected behavior and error handling.
---
 src/helpers/src/Functions.php                 | 31 ++++++++++++++-----
 src/support/src/Functions.php                 | 24 ++++++++++++++
 .../PendingAmqpProducerMessageDispatch.php    |  2 +-
 .../src/PendingAsyncQueueDispatch.php         |  2 +-
 .../PendingKafkaProducerMessageDispatch.php   |  2 +-
 tests/{Helpers => Support}/DispatchTest.php   | 10 +++---
 6 files changed, 55 insertions(+), 16 deletions(-)
 rename src/{helpers => support}/src/PendingAmqpProducerMessageDispatch.php (98%)
 rename src/{helpers => support}/src/PendingAsyncQueueDispatch.php (97%)
 rename src/{helpers => support}/src/PendingKafkaProducerMessageDispatch.php (97%)
 rename tests/{Helpers => Support}/DispatchTest.php (98%)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index 44592a4c2..a2b320ba9 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -16,7 +16,9 @@
 use Countable;
 use Exception;
 use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
+use Hyperf\Amqp\Annotation\Producer;
 use Hyperf\Amqp\Message\ProducerMessageInterface;
+use Hyperf\AsyncQueue\Driver\DriverFactory;
 use Hyperf\AsyncQueue\JobInterface;
 use Hyperf\Context\ApplicationContext;
 use Hyperf\Contract\SessionInterface;
@@ -27,6 +29,7 @@
 use Hyperf\HttpMessage\Stream\SwooleStream;
 use Hyperf\HttpServer\Contract\RequestInterface;
 use Hyperf\HttpServer\Contract\ResponseInterface;
+use Hyperf\Kafka\ProducerManager;
 use Hyperf\Logger\LoggerFactory;
 use Hyperf\Stringable\Str;
 use Hyperf\Support\Fluent;
@@ -182,21 +185,33 @@ function di(?string $abstract = null, array $parameters = [])
 }
 
 /**
- * Do not assign a value to the return value of this function unless you are very clear about the consequences of doing so.
- * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
- * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
- * @throws InvalidArgumentException
+ * @param AsyncTaskInterface|Closure|JobInterface|ProduceMessage|ProducerMessageInterface|object $job
+ * @return bool
  */
-function dispatch($job)
+function dispatch($job, ...$arguments)
 {
     if ($job instanceof Closure) {
         $job = CallQueuedClosure::create($job);
+        if ($arguments[2] ?? 0) {
+            $job->setMaxAttempts((int) $arguments[2]);
+        }
     }
 
     return match (true) {
-        interface_exists(ProducerMessageInterface::class) && $job instanceof ProducerMessageInterface => new PendingAmqpProducerMessageDispatch($job),
-        class_exists(ProduceMessage::class) && $job instanceof ProduceMessage => new PendingKafkaProducerMessageDispatch($job),
-        interface_exists(JobInterface::class) && $job instanceof JobInterface => new PendingAsyncQueueDispatch($job),
+        $job instanceof JobInterface => di(DriverFactory::class)
+            ->get((string) ($arguments[0] ?? (fn () => $this->queue ?? $this->pool ?? 'default')->call($job)))
+            ->push(
+                tap(
+                    $job,
+                    fn ($job) => isset($arguments[2]) && (fn () => $this->maxAttempts = (int) $arguments[2])->call($job)
+                ),
+                (int) ($arguments[1] ?? (fn () => $this->delay ?? 0)->call($job))
+            ),
+        $job instanceof ProducerMessageInterface => di(Producer::class)
+            ->produce($job, ...$arguments),
+        $job instanceof ProduceMessage => di(ProducerManager::class)
+            ->getProducer((string) ($arguments[0] ?? 'default'))
+            ->sendBatch([$job]),
         default => throw new InvalidArgumentException('Unsupported job type.')
     };
 }
diff --git a/src/support/src/Functions.php b/src/support/src/Functions.php
index cc9f43026..140d19170 100644
--- a/src/support/src/Functions.php
+++ b/src/support/src/Functions.php
@@ -13,9 +13,33 @@
 
 use Closure;
 use Exception;
+use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
+use Hyperf\Amqp\Message\ProducerMessageInterface;
+use Hyperf\AsyncQueue\JobInterface;
+use longlang\phpkafka\Producer\ProduceMessage;
 
 use function Hyperf\Support\value;
 
+/**
+ * Do not assign a value to the return value of this function unless you are very clear about the consequences of doing so.
+ * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
+ * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
+ * @throws InvalidArgumentException
+ */
+function dispatch($job)
+{
+    if ($job instanceof Closure) {
+        $job = CallQueuedClosure::create($job);
+    }
+
+    return match (true) {
+        interface_exists(ProducerMessageInterface::class) && $job instanceof ProducerMessageInterface => new PendingAmqpProducerMessageDispatch($job),
+        class_exists(ProduceMessage::class) && $job instanceof ProduceMessage => new PendingKafkaProducerMessageDispatch($job),
+        interface_exists(JobInterface::class) && $job instanceof JobInterface => new PendingAsyncQueueDispatch($job),
+        default => throw new \InvalidArgumentException('Unsupported job type.')
+    };
+}
+
 /**
  * Retry an operation a given number of times.
  * @template TReturn
diff --git a/src/helpers/src/PendingAmqpProducerMessageDispatch.php b/src/support/src/PendingAmqpProducerMessageDispatch.php
similarity index 98%
rename from src/helpers/src/PendingAmqpProducerMessageDispatch.php
rename to src/support/src/PendingAmqpProducerMessageDispatch.php
index 91f3992af..08690cb79 100644
--- a/src/helpers/src/PendingAmqpProducerMessageDispatch.php
+++ b/src/support/src/PendingAmqpProducerMessageDispatch.php
@@ -9,7 +9,7 @@
  * @contact  huangdijia@gmail.com
  */
 
-namespace FriendsOfHyperf\Helpers;
+namespace FriendsOfHyperf\Support;
 
 use Hyperf\Amqp\Message\ProducerMessageInterface;
 use Hyperf\Amqp\Producer;
diff --git a/src/helpers/src/PendingAsyncQueueDispatch.php b/src/support/src/PendingAsyncQueueDispatch.php
similarity index 97%
rename from src/helpers/src/PendingAsyncQueueDispatch.php
rename to src/support/src/PendingAsyncQueueDispatch.php
index 9da20c5d1..8f66de0ef 100644
--- a/src/helpers/src/PendingAsyncQueueDispatch.php
+++ b/src/support/src/PendingAsyncQueueDispatch.php
@@ -9,7 +9,7 @@
  * @contact  huangdijia@gmail.com
  */
 
-namespace FriendsOfHyperf\Helpers;
+namespace FriendsOfHyperf\Support;
 
 use Hyperf\AsyncQueue\Driver\DriverFactory;
 use Hyperf\AsyncQueue\JobInterface;
diff --git a/src/helpers/src/PendingKafkaProducerMessageDispatch.php b/src/support/src/PendingKafkaProducerMessageDispatch.php
similarity index 97%
rename from src/helpers/src/PendingKafkaProducerMessageDispatch.php
rename to src/support/src/PendingKafkaProducerMessageDispatch.php
index 44a95dc15..38df9c025 100644
--- a/src/helpers/src/PendingKafkaProducerMessageDispatch.php
+++ b/src/support/src/PendingKafkaProducerMessageDispatch.php
@@ -9,7 +9,7 @@
  * @contact  huangdijia@gmail.com
  */
 
-namespace FriendsOfHyperf\Helpers;
+namespace FriendsOfHyperf\Support;
 
 use Hyperf\Conditionable\Conditionable;
 use Hyperf\Context\ApplicationContext;
diff --git a/tests/Helpers/DispatchTest.php b/tests/Support/DispatchTest.php
similarity index 98%
rename from tests/Helpers/DispatchTest.php
rename to tests/Support/DispatchTest.php
index 6921e96a1..25bfde687 100644
--- a/tests/Helpers/DispatchTest.php
+++ b/tests/Support/DispatchTest.php
@@ -9,13 +9,13 @@
  * @contact  huangdijia@gmail.com
  */
 
-namespace FriendsOfHyperf\Tests\Helpers;
+namespace FriendsOfHyperf\Tests\Support;
 
 use Exception;
 use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
-use FriendsOfHyperf\Helpers\PendingAmqpProducerMessageDispatch;
-use FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch;
-use FriendsOfHyperf\Helpers\PendingKafkaProducerMessageDispatch;
+use FriendsOfHyperf\Support\PendingAmqpProducerMessageDispatch;
+use FriendsOfHyperf\Support\PendingAsyncQueueDispatch;
+use FriendsOfHyperf\Support\PendingKafkaProducerMessageDispatch;
 use FriendsOfHyperf\Tests\TestCase;
 use Hyperf\Amqp\Message\ProducerMessage;
 use Hyperf\Amqp\Producer;
@@ -37,7 +37,7 @@
 /**
  * @internal
  */
-#[\PHPUnit\Framework\Attributes\Group('helpers')]
+#[\PHPUnit\Framework\Attributes\Group('support')]
 class DispatchTest extends TestCase
 {
     protected function setUp(): void

From 22dd7e4642646ff0f6a91230495844f49b8d2915 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:49:36 +0800
Subject: [PATCH 13/19] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20AsyncTaskInt?=
 =?UTF-8?q?erface=20=E7=9A=84=E4=BD=BF=E7=94=A8=E5=A3=B0=E6=98=8E?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/helpers/src/Functions.php | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index a2b320ba9..f44cab92b 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -16,6 +16,7 @@
 use Countable;
 use Exception;
 use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
+use FriendsOfHyperf\AsyncTask\TaskInterface as AsyncTaskInterface;
 use Hyperf\Amqp\Annotation\Producer;
 use Hyperf\Amqp\Message\ProducerMessageInterface;
 use Hyperf\AsyncQueue\Driver\DriverFactory;

From 04beb0df9a00167bf9c49d61dc9d8c19842d4e9d Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:50:03 +0800
Subject: [PATCH 14/19] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=AF=BC?=
 =?UTF-8?q?=E5=85=A5=E8=AF=AD=E5=8F=A5=EF=BC=8C=E7=A1=AE=E4=BF=9D=E6=AD=A3?=
 =?UTF-8?q?=E7=A1=AE=E4=BD=BF=E7=94=A8=20Hyperf\Amqp\Producer?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/helpers/src/Functions.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index f44cab92b..70488500b 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -17,8 +17,8 @@
 use Exception;
 use FriendsOfHyperf\AsyncQueueClosureJob\CallQueuedClosure;
 use FriendsOfHyperf\AsyncTask\TaskInterface as AsyncTaskInterface;
-use Hyperf\Amqp\Annotation\Producer;
 use Hyperf\Amqp\Message\ProducerMessageInterface;
+use Hyperf\Amqp\Producer;
 use Hyperf\AsyncQueue\Driver\DriverFactory;
 use Hyperf\AsyncQueue\JobInterface;
 use Hyperf\Context\ApplicationContext;

From 7ef7ce2e81845870aeac20475952f7f702e9ad0f Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:50:40 +0800
Subject: [PATCH 15/19] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20dispatch()=20?=
 =?UTF-8?q?=E6=B5=8B=E8=AF=95=E7=9A=84=E8=BF=94=E5=9B=9E=E7=B1=BB=E5=9E=8B?=
 =?UTF-8?q?=EF=BC=8C=E7=A1=AE=E4=BF=9D=E8=BF=94=E5=9B=9E=E5=B8=83=E5=B0=94?=
 =?UTF-8?q?=E5=80=BC?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 types/Helpers/Functions.php | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/types/Helpers/Functions.php b/types/Helpers/Functions.php
index 0523660c5..db65a9583 100644
--- a/types/Helpers/Functions.php
+++ b/types/Helpers/Functions.php
@@ -150,7 +150,7 @@
 
 // dispatch() tests - returns bool
 // Note: dispatch() has complex return types based on job type, testing the common case
-assertType('FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch', dispatch(new class implements Hyperf\AsyncQueue\JobInterface {
+assertType('bool', dispatch(new class implements Hyperf\AsyncQueue\JobInterface {
     public function handle(): void
     {
     }
@@ -175,9 +175,6 @@ public function getQueueName(): string
     }
 }));
 
-assertType('FriendsOfHyperf\Helpers\PendingAsyncQueueDispatch', dispatch(function () {
-}));
-
 // info() tests
 assertType('mixed', info('message'));
 

From dc71e9c2b6ebb02436c6ad0c337ae54f6491e9f2 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:52:24 +0800
Subject: [PATCH 16/19] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=AF=BC?=
 =?UTF-8?q?=E5=85=A5=E8=AF=AD=E5=8F=A5=EF=BC=8C=E7=A1=AE=E4=BF=9D=E6=AD=A3?=
 =?UTF-8?q?=E7=A1=AE=E4=BD=BF=E7=94=A8=20FriendsOfHyperf\Support\dispatch?=
 =?UTF-8?q?=20=E5=87=BD=E6=95=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/Support/DispatchTest.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/Support/DispatchTest.php b/tests/Support/DispatchTest.php
index 25bfde687..4aa01b152 100644
--- a/tests/Support/DispatchTest.php
+++ b/tests/Support/DispatchTest.php
@@ -32,7 +32,7 @@
 use ReflectionClass;
 use stdClass;
 
-use function FriendsOfHyperf\Helpers\dispatch;
+use function FriendsOfHyperf\Support\dispatch;
 
 /**
  * @internal

From 809559558afd5ea71456431002e2e38f3838bcdf Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:54:35 +0800
Subject: [PATCH 17/19] =?UTF-8?q?fix:=20=E6=B7=BB=E5=8A=A0=E5=BC=83?=
 =?UTF-8?q?=E7=94=A8=E6=B3=A8=E9=87=8A=EF=BC=8C=E5=BB=BA=E8=AE=AE=E4=BD=BF?=
 =?UTF-8?q?=E7=94=A8=20FriendsOfHyperf\Support\dispatch()=20=E6=9B=BF?=
 =?UTF-8?q?=E4=BB=A3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/helpers/src/Functions.php | 1 +
 1 file changed, 1 insertion(+)

diff --git a/src/helpers/src/Functions.php b/src/helpers/src/Functions.php
index 70488500b..61175ae06 100644
--- a/src/helpers/src/Functions.php
+++ b/src/helpers/src/Functions.php
@@ -186,6 +186,7 @@ function di(?string $abstract = null, array $parameters = [])
 }
 
 /**
+ * @deprecated use FriendsOfHyperf\Support\dispatch() instead
  * @param AsyncTaskInterface|Closure|JobInterface|ProduceMessage|ProducerMessageInterface|object $job
  * @return bool
  */

From ae5fd97d3f29c3687c9000aa64fc747ca8861481 Mon Sep 17 00:00:00 2001
From: Deeka Wong <8337659+huangdijia@users.noreply.github.com>
Date: Fri, 14 Nov 2025 06:54:42 +0800
Subject: [PATCH 18/19] =?UTF-8?q?fix:=20=E6=B7=BB=E5=8A=A0=E5=AF=B9=20Prod?=
 =?UTF-8?q?ucerMessage=20=E7=9A=84=20setPoolName=20=E6=96=B9=E6=B3=95?=
 =?UTF-8?q?=E7=9A=84=E6=A8=A1=E6=8B=9F=EF=BC=8C=E7=A1=AE=E4=BF=9D=E6=AD=A3?=
 =?UTF-8?q?=E7=A1=AE=E5=A4=84=E7=90=86=E8=87=AA=E5=AE=9A=E4=B9=89=E6=B1=A0?=
 =?UTF-8?q?=E5=90=8D=E7=A7=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/Support/DispatchTest.php | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/tests/Support/DispatchTest.php b/tests/Support/DispatchTest.php
index 4aa01b152..00fddee9b 100644
--- a/tests/Support/DispatchTest.php
+++ b/tests/Support/DispatchTest.php
@@ -217,6 +217,11 @@ public function testPendingAsyncQueueDispatchWithConditionable()
     public function testPendingAmqpProducerMessageDispatchOnPool()
     {
         $message = m::mock(ProducerMessage::class);
+        $message->shouldReceive('setPoolName')
+            ->with('amqp-custom')
+            ->once()
+            ->andReturnSelf();
+
         $pending = dispatch($message);
 
         $result = $pending->onPool('amqp-custom');
@@ -250,6 +255,10 @@ public function testPendingAmqpProducerMessageDispatchSetTimeout()
     public function testPendingAmqpProducerMessageDispatchFluentChaining()
     {
         $message = m::mock(ProducerMessage::class);
+        $message->shouldReceive('setPoolName')
+            ->with('amqp-pool')
+            ->once()
+            ->andReturnSelf();
 
         $pending = dispatch($message)
             ->onPool('amqp-pool')

From af2c45a2b04cbe52089b048e08812f76fc77af90 Mon Sep 17 00:00:00 2001
From: Deeka Wong <huangdijia@gmail.com>
Date: Fri, 14 Nov 2025 07:03:42 +0800
Subject: [PATCH 19/19] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20Functions.php?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/support/src/Functions.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/support/src/Functions.php b/src/support/src/Functions.php
index 140d19170..d910ae46a 100644
--- a/src/support/src/Functions.php
+++ b/src/support/src/Functions.php
@@ -24,7 +24,7 @@
  * Do not assign a value to the return value of this function unless you are very clear about the consequences of doing so.
  * @param Closure|JobInterface|ProduceMessage|ProducerMessageInterface|mixed $job
  * @return ($job is Closure ? PendingAsyncQueueDispatch : ($job is JobInterface ? PendingAsyncQueueDispatch : ($job is ProducerMessageInterface ? PendingAmqpProducerMessageDispatch : PendingKafkaProducerMessageDispatch)))
- * @throws InvalidArgumentException
+ * @throws \InvalidArgumentException
  */
 function dispatch($job)
 {