diff --git a/composer.json b/composer.json
index 38288a6..214a6a4 100644
--- a/composer.json
+++ b/composer.json
@@ -22,9 +22,12 @@
     },
     "require": {
         "php": ">=8.1",
-        "cakephp/cakephp": "dev-5.next as 5.1.0",
-        "enqueue/simple-client": "^0.10",
-        "psr/log": "^3.0"
+        "cakephp/cakephp": "^5.1.0",
+        "psr/log": "^3.0",
+        "enqueue/enqueue": "^0.10",
+        "queue-interop/amqp-interop": "^0.8.2",
+        "queue-interop/queue-interop": "^0.8",
+        "symfony/config": "^5.4|^6.0|^7.0"
     },
     "require-dev": {
         "cakephp/bake": "dev-3.next",
diff --git a/src/Command/RequeueCommand.php b/src/Command/RequeueCommand.php
index de6cb88..b9efd02 100644
--- a/src/Command/RequeueCommand.php
+++ b/src/Command/RequeueCommand.php
@@ -138,7 +138,7 @@ public function execute(Arguments $args, ConsoleIo $io): void
                         'config' => $failedJob->config,
                         'priority' => $failedJob->priority,
                         'queue' => $failedJob->queue,
-                    ]
+                    ],
                 );
 
                 $failedJobsTable->deleteOrFail($failedJob);
diff --git a/src/Command/WorkerCommand.php b/src/Command/WorkerCommand.php
index f97148b..6541f19 100644
--- a/src/Command/WorkerCommand.php
+++ b/src/Command/WorkerCommand.php
@@ -110,7 +110,7 @@ public function getOptionParser(): ConsoleOptionParser
             'short' => 'a',
         ]);
         $parser->setDescription(
-            'Runs a queue worker that consumes from the named queue.'
+            'Runs a queue worker that consumes from the named queue.',
         );
 
         return $parser;
diff --git a/src/Consumption/LimitAttemptsExtension.php b/src/Consumption/LimitAttemptsExtension.php
index 3d2da34..c3a0af6 100644
--- a/src/Consumption/LimitAttemptsExtension.php
+++ b/src/Consumption/LimitAttemptsExtension.php
@@ -65,7 +65,7 @@ public function onResult(MessageResult $context): void
 
         if ($attemptNumber >= $maxAttempts) {
             $context->changeResult(
-                Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts))
+                Result::reject(sprintf('The maximum number of %d allowed attempts was reached.', $maxAttempts)),
             );
 
             $exception = (string)$message->getProperty('jobException');
@@ -73,7 +73,7 @@ public function onResult(MessageResult $context): void
             $this->dispatchEvent(
                 'Consumption.LimitAttemptsExtension.failed',
                 ['exception' => $exception, 'logger' => $context->getLogger()],
-                $jobMessage
+                $jobMessage,
             );
 
             return;
@@ -88,7 +88,7 @@ public function onResult(MessageResult $context): void
         $producer->send($consumer->getQueue(), $newMessage);
 
         $context->changeResult(
-            Result::reject('A copy of the message was sent with an incremented attempt count.')
+            Result::reject('A copy of the message was sent with an incremented attempt count.'),
         );
     }
 }
diff --git a/src/Consumption/LimitConsumedMessagesExtension.php b/src/Consumption/LimitConsumedMessagesExtension.php
index 680f8b4..e202fbe 100644
--- a/src/Consumption/LimitConsumedMessagesExtension.php
+++ b/src/Consumption/LimitConsumedMessagesExtension.php
@@ -79,7 +79,7 @@ protected function shouldBeStopped(LoggerInterface $logger): bool
             $logger->debug(sprintf(
                 '[LimitConsumedMessagesExtension] Message consumption is interrupted since the message limit ' .
                 'reached. limit: "%s"',
-                $this->messageLimit
+                $this->messageLimit,
             ));
 
             return true;
diff --git a/src/Enqueue/SimpleClient.php b/src/Enqueue/SimpleClient.php
new file mode 100644
index 0000000..6577e75
--- /dev/null
+++ b/src/Enqueue/SimpleClient.php
@@ -0,0 +1,403 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * CakePHP(tm) : Rapid Development Framework (https://cakephp.org)
+ * Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ *
+ * Licensed under The MIT License
+ * For full copyright and license information, please see the LICENSE.txt
+ * Redistributions of files must retain the above copyright notice.
+ *
+ * @copyright     Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ * @link          https://cakephp.org CakePHP(tm) Project
+ * @since         3.0.0
+ * @license       https://opensource.org/licenses/MIT MIT License
+ */
+namespace Cake\Queue\Enqueue;
+
+use Enqueue\ArrayProcessorRegistry;
+use Enqueue\Client\ChainExtension as ClientChainExtensions;
+use Enqueue\Client\Config;
+use Enqueue\Client\ConsumptionExtension\DelayRedeliveredMessageExtension;
+use Enqueue\Client\ConsumptionExtension\LogExtension;
+use Enqueue\Client\ConsumptionExtension\SetRouterPropertiesExtension;
+use Enqueue\Client\DelegateProcessor;
+use Enqueue\Client\DriverFactory;
+use Enqueue\Client\DriverInterface;
+use Enqueue\Client\Message;
+use Enqueue\Client\Producer;
+use Enqueue\Client\ProducerInterface;
+use Enqueue\Client\Route;
+use Enqueue\Client\RouteCollection;
+use Enqueue\Client\RouterProcessor;
+use Enqueue\ConnectionFactoryFactory;
+use Enqueue\Consumption\CallbackProcessor;
+use Enqueue\Consumption\ChainExtension as ConsumptionChainExtension;
+use Enqueue\Consumption\Extension\ReplyExtension;
+use Enqueue\Consumption\Extension\SignalExtension;
+use Enqueue\Consumption\ExtensionInterface;
+use Enqueue\Consumption\QueueConsumer;
+use Enqueue\Consumption\QueueConsumerInterface;
+use Enqueue\Rpc\Promise;
+use Enqueue\Rpc\RpcFactory;
+use Enqueue\Symfony\Client\DependencyInjection\ClientFactory;
+use Enqueue\Symfony\DependencyInjection\TransportFactory;
+use Interop\Queue\Processor;
+use JsonSerializable;
+use LogicException;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Symfony\Component\Config\Definition\Builder\TreeBuilder;
+use Symfony\Component\Config\Definition\NodeInterface;
+use Symfony\Component\Config\Definition\Processor as ConfigProcessor;
+
+class SimpleClient
+{
+    /**
+     * @var \Enqueue\Client\DriverInterface
+     */
+    private DriverInterface $driver;
+
+    /**
+     * @var \Enqueue\Client\Producer
+     */
+    private Producer $producer;
+
+    /**
+     * @var \Enqueue\Consumption\QueueConsumer
+     */
+    private QueueConsumer $queueConsumer;
+
+    /**
+     * @var \Enqueue\ArrayProcessorRegistry
+     */
+    private ArrayProcessorRegistry $processorRegistry;
+
+    /**
+     * @var \Enqueue\Client\DelegateProcessor
+     */
+    private DelegateProcessor $delegateProcessor;
+
+    /**
+     * @var \Psr\Log\LoggerInterface
+     */
+    private LoggerInterface $logger;
+
+    /**
+     * The config could be a transport DSN (string) or an array, here's an example of a few DSNs:.
+     *
+     * $config = amqp:
+     * $config = amqp://guest:guest@localhost:5672/%2f?lazy=1&persisted=1
+     * $config = file://foo/bar/
+     * $config = null:
+     *
+     * or an array:
+     *
+     * $config = [
+     *   'transport' => [
+     *      'dsn' => 'amqps://guest:guest@localhost:5672/%2f',
+     *      'ssl_cacert' => '/a/dir/cacert.pem',
+     *      'ssl_cert' => '/a/dir/cert.pem',
+     *      'ssl_key' => '/a/dir/key.pem',
+     * ]
+     *
+     * with custom connection factory class
+     *
+     * $config = [
+     *   'transport' => [
+     *      'dsn' => 'amqps://guest:guest@localhost:5672/%2f',
+     *      'connection_factory_class' => 'aCustomConnectionFactory',
+     *      // other options available options are factory_class and factory_service
+     * ]
+     *
+     * The client config
+     *
+     * $config = [
+     *   'transport' => 'null:',
+     *   'client' => [
+     *     'prefix'                   => 'enqueue',
+     *     'separator'                => '.',
+     *     'app_name'                 => 'app',
+     *     'router_topic'             => 'router',
+     *     'router_queue'             => 'default',
+     *     'default_queue'            => 'default',
+     *     'redelivered_delay_time'   => 0
+     *   ],
+     *   'extensions' => [
+     *     'signal_extension' => true,
+     *     'reply_extension' => true,
+     *   ]
+     * ]
+     *
+     * @param array|string $config
+     * @param \Psr\Log\LoggerInterface|null $logger
+     */
+    public function __construct(string|array $config, ?LoggerInterface $logger = null)
+    {
+        if (is_string($config)) {
+            $config = [
+                'transport' => $config,
+                'client' => true,
+            ];
+        }
+
+        $this->logger = $logger ?: new NullLogger();
+
+        $this->build(['enqueue' => $config]);
+    }
+
+    /**
+     * @param string $topic
+     * @param \Interop\Queue\Processor|callable $processor
+     * @param string|null $processorName
+     */
+    public function bindTopic(string $topic, callable|Processor $processor, ?string $processorName = null): void
+    {
+        if (is_callable($processor)) {
+            $processor = new CallbackProcessor($processor);
+        }
+
+        if (!$processor instanceof Processor) {
+            throw new LogicException('The processor must be either callable or instance of Processor');
+        }
+
+        $processorName = $processorName ?: uniqid($processor::class);
+
+        $this->driver->getRouteCollection()->add(new Route($topic, Route::TOPIC, $processorName));
+        $this->processorRegistry->add($processorName, $processor);
+    }
+
+    /**
+     * @param string $command
+     * @param \Interop\Queue\Processor|callable $processor
+     * @param string|null $processorName
+     */
+    public function bindCommand(string $command, callable|Processor $processor, ?string $processorName = null): void
+    {
+        if (is_callable($processor)) {
+            $processor = new CallbackProcessor($processor);
+        }
+
+        if (!$processor instanceof Processor) {
+            throw new LogicException('The processor must be either callable or instance of Processor');
+        }
+
+        $processorName = $processorName ?: uniqid($processor::class);
+
+        $this->driver->getRouteCollection()->add(new Route($command, Route::COMMAND, $processorName));
+        $this->processorRegistry->add($processorName, $processor);
+    }
+
+    /**
+     * @param string $command
+     * @param \JsonSerializable|\Enqueue\Client\Message|array|string $message
+     * @param bool $needReply
+     * @return \Enqueue\Rpc\Promise|null
+     * @throws \Exception
+     */
+    public function sendCommand(
+        string $command,
+        string|array|JsonSerializable|Message $message,
+        bool $needReply = false,
+    ): ?Promise {
+        return $this->producer->sendCommand($command, $message, $needReply);
+    }
+
+    /**
+     * @param string $topic
+     * @param \Enqueue\Client\Message|array|string $message
+     * @throws \Exception
+     */
+    public function sendEvent(string $topic, string|array|Message $message): void
+    {
+        $this->producer->sendEvent($topic, $message);
+    }
+
+    /**
+     * @param \Enqueue\Consumption\ExtensionInterface|null $runtimeExtension
+     * @return void
+     * @throws \Exception
+     */
+    public function consume(?ExtensionInterface $runtimeExtension = null): void
+    {
+        $this->setupBroker();
+
+        $boundQueues = [];
+
+        $routerQueue = $this->getDriver()->createQueue($this->getDriver()->getConfig()->getRouterQueue());
+        $this->queueConsumer->bind($routerQueue, $this->delegateProcessor);
+        $boundQueues[$routerQueue->getQueueName()] = true;
+
+        foreach ($this->driver->getRouteCollection()->all() as $route) {
+            $queue = $this->getDriver()->createRouteQueue($route);
+            if (array_key_exists($queue->getQueueName(), $boundQueues)) {
+                continue;
+            }
+
+            $this->queueConsumer->bind($queue, $this->delegateProcessor);
+
+            $boundQueues[$queue->getQueueName()] = true;
+        }
+
+        $this->queueConsumer->consume($runtimeExtension);
+    }
+
+    /**
+     * @return \Enqueue\Consumption\QueueConsumerInterface
+     */
+    public function getQueueConsumer(): QueueConsumerInterface
+    {
+        return $this->queueConsumer;
+    }
+
+    /**
+     * @return \Enqueue\Client\DriverInterface
+     */
+    public function getDriver(): DriverInterface
+    {
+        return $this->driver;
+    }
+
+    /**
+     * @param bool $setupBroker
+     * @return \Enqueue\Client\ProducerInterface
+     */
+    public function getProducer(bool $setupBroker = false): ProducerInterface
+    {
+        $setupBroker && $this->setupBroker();
+
+        return $this->producer;
+    }
+
+    /**
+     * @return \Enqueue\Client\DelegateProcessor
+     */
+    public function getDelegateProcessor(): DelegateProcessor
+    {
+        return $this->delegateProcessor;
+    }
+
+    /**
+     * @return void
+     */
+    public function setupBroker(): void
+    {
+        $this->getDriver()->setupBroker();
+    }
+
+    /**
+     * @param array $configs
+     * @return void
+     */
+    public function build(array $configs): void
+    {
+        $configProcessor = new ConfigProcessor();
+        $simpleClientConfig = $configProcessor->process($this->createConfiguration(), $configs);
+
+        if (isset($simpleClientConfig['transport']['factory_service'])) {
+            throw new LogicException('transport.factory_service option is not supported by simple client');
+        }
+        if (isset($simpleClientConfig['transport']['factory_class'])) {
+            throw new LogicException('transport.factory_class option is not supported by simple client');
+        }
+        if (isset($simpleClientConfig['transport']['connection_factory_class'])) {
+            throw new LogicException('transport.connection_factory_class option is not supported by simple client');
+        }
+
+        $connectionFactoryFactory = new ConnectionFactoryFactory();
+        $connection = $connectionFactoryFactory->create($simpleClientConfig['transport']);
+
+        $clientExtensions = new ClientChainExtensions([]);
+
+        $config = new Config(
+            $simpleClientConfig['client']['prefix'],
+            $simpleClientConfig['client']['separator'],
+            $simpleClientConfig['client']['app_name'],
+            $simpleClientConfig['client']['router_topic'],
+            $simpleClientConfig['client']['router_queue'],
+            $simpleClientConfig['client']['default_queue'],
+            'enqueue.client.router_processor',
+            $simpleClientConfig['transport'],
+            [],
+        );
+
+        $routeCollection = new RouteCollection([]);
+        $driverFactory = new DriverFactory();
+
+        $driver = $driverFactory->create(
+            $connection,
+            $config,
+            $routeCollection,
+        );
+
+        $rpcFactory = new RpcFactory($driver->getContext());
+
+        $producer = new Producer($driver, $rpcFactory, $clientExtensions);
+
+        $processorRegistry = new ArrayProcessorRegistry([]);
+
+        $delegateProcessor = new DelegateProcessor($processorRegistry);
+
+        // consumption extensions
+        $consumptionExtensions = [];
+        if ($simpleClientConfig['client']['redelivered_delay_time']) {
+            $consumptionExtensions[] = new DelayRedeliveredMessageExtension(
+                $driver,
+                $simpleClientConfig['client']['redelivered_delay_time'],
+            );
+        }
+
+        if ($simpleClientConfig['extensions']['signal_extension']) {
+            $consumptionExtensions[] = new SignalExtension();
+        }
+
+        if ($simpleClientConfig['extensions']['reply_extension']) {
+            $consumptionExtensions[] = new ReplyExtension();
+        }
+
+        $consumptionExtensions[] = new SetRouterPropertiesExtension($driver);
+        $consumptionExtensions[] = new LogExtension();
+
+        $consumptionChainExtension = new ConsumptionChainExtension($consumptionExtensions);
+        $queueConsumer = new QueueConsumer($driver->getContext(), $consumptionChainExtension, [], $this->logger);
+
+        $routerProcessor = new RouterProcessor($driver);
+
+        $processorRegistry->add($config->getRouterProcessor(), $routerProcessor);
+
+        $this->driver = $driver;
+        $this->producer = $producer;
+        $this->queueConsumer = $queueConsumer;
+        $this->delegateProcessor = $delegateProcessor;
+        $this->processorRegistry = $processorRegistry;
+    }
+
+    /**
+     * @return \Symfony\Component\Config\Definition\NodeInterface
+     */
+    private function createConfiguration(): NodeInterface
+    {
+        $tb = new TreeBuilder('enqueue');
+        $rootNode = $tb->getRootNode();
+
+        $rootNode
+            ->beforeNormalization()
+            ->ifEmpty()->then(function () {
+                return ['transport' => ['dsn' => 'null:']];
+            });
+
+        $rootNode
+            ->append(TransportFactory::getConfiguration())
+            ->append(TransportFactory::getQueueConsumerConfiguration())
+            ->append(ClientFactory::getConfiguration(false));
+
+        $rootNode->children()
+            ->arrayNode('extensions')->addDefaultsIfNotSet()->children()
+            ->booleanNode('signal_extension')->defaultValue(function_exists('pcntl_signal_dispatch'))->end()
+            ->booleanNode('reply_extension')->defaultTrue()->end()
+            ->end();
+
+        return $tb->buildTree();
+    }
+}
diff --git a/src/Job/Message.php b/src/Job/Message.php
index 5627b74..a4eb250 100644
--- a/src/Job/Message.php
+++ b/src/Job/Message.php
@@ -114,7 +114,7 @@ public function getTarget(): array
         if (!is_array($target) || count($target) !== 2) {
             throw new RuntimeException(sprintf(
                 'Message class should be in the form `[class, method]` got `%s`',
-                json_encode($target)
+                json_encode($target),
             ));
         }
 
diff --git a/src/Listener/FailedJobsListener.php b/src/Listener/FailedJobsListener.php
index 1064f37..f60d15e 100644
--- a/src/Listener/FailedJobsListener.php
+++ b/src/Listener/FailedJobsListener.php
@@ -83,7 +83,7 @@ public function storeFailedJob(object $event): void
                 throw new RuntimeException(
                     sprintf('`logger` was not defined on %s event.', $event->getName()),
                     0,
-                    $e
+                    $e,
                 );
             }
 
@@ -91,7 +91,7 @@ public function storeFailedJob(object $event): void
                 throw new RuntimeException(
                     sprintf('`logger` is not an instance of `LoggerInterface` on %s event.', $event->getName()),
                     0,
-                    $e
+                    $e,
                 );
             }
 
diff --git a/src/Mailer/Transport/QueueTransport.php b/src/Mailer/Transport/QueueTransport.php
index 8c7688b..c798d53 100644
--- a/src/Mailer/Transport/QueueTransport.php
+++ b/src/Mailer/Transport/QueueTransport.php
@@ -54,7 +54,7 @@ public function send(Message $message): array
                 'returnPath',
                 'cc',
                 'bcc',
-            ]
+            ],
         );
 
         return ['headers' => $headers, 'message' => 'Message has been enqueued'];
@@ -72,7 +72,7 @@ protected function enqueueJob(array $data, array $options): void
         QueueManager::push(
             [SendMailJob::class, 'execute'],
             $data,
-            $options
+            $options,
         );
     }
 
diff --git a/src/Plugin.php b/src/Plugin.php
index 413fb5c..d607ef0 100644
--- a/src/Plugin.php
+++ b/src/Plugin.php
@@ -53,7 +53,7 @@ public function bootstrap(PluginApplicationInterface $app): void
         if (!Configure::read('Queue')) {
             throw new InvalidArgumentException(
                 'Missing `Queue` configuration key, please check the CakePHP Queue documentation' .
-                ' to complete the plugin setup.'
+                ' to complete the plugin setup.',
             );
         }
 
diff --git a/src/QueueManager.php b/src/QueueManager.php
index 664ec31..5ae3334 100644
--- a/src/QueueManager.php
+++ b/src/QueueManager.php
@@ -20,8 +20,8 @@
 use Cake\Cache\Cache;
 use Cake\Core\App;
 use Cake\Log\Log;
+use Cake\Queue\Enqueue\SimpleClient;
 use Enqueue\Client\Message as ClientMessage;
-use Enqueue\SimpleClient\SimpleClient;
 use InvalidArgumentException;
 use LogicException;
 
@@ -162,7 +162,7 @@ public static function drop(string $key): void
      * Get a queueing engine
      *
      * @param string $name Key name of a configured adapter to get.
-     * @return \Enqueue\SimpleClient\SimpleClient
+     * @return \Cake\Queue\Enqueue\SimpleClient
      */
     public static function engine(string $name): SimpleClient
     {
@@ -232,7 +232,7 @@ public static function push(string|array $className, array $data = [], array $op
         if (!empty($class::$shouldBeUnique)) {
             if (empty($config['uniqueCache'])) {
                 throw new InvalidArgumentException(
-                    "$class::\$shouldBeUnique is set to `true` but `uniqueCache` configuration is missing."
+                    "$class::\$shouldBeUnique is set to `true` but `uniqueCache` configuration is missing.",
                 );
             }
 
@@ -241,7 +241,7 @@ public static function push(string|array $className, array $data = [], array $op
             if (Cache::read($uniqueId, $config['uniqueCacheKey'])) {
                 if ($logger) {
                     $logger->debug(
-                        "An identical instance of $class already exists on the queue. This push will be ignored."
+                        "An identical instance of $class already exists on the queue. This push will be ignored.",
                     );
                 }
 
diff --git a/tests/TestCase/Enqueue/SimpleClientTest.php b/tests/TestCase/Enqueue/SimpleClientTest.php
new file mode 100644
index 0000000..329efe0
--- /dev/null
+++ b/tests/TestCase/Enqueue/SimpleClientTest.php
@@ -0,0 +1,219 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * CakePHP(tm) : Rapid Development Framework (https://cakephp.org)
+ * Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ *
+ * Licensed under The MIT License
+ * For full copyright and license information, please see the LICENSE.txt
+ * Redistributions of files must retain the above copyright notice.
+ *
+ * @copyright     Copyright (c) Cake Software Foundation, Inc. (https://cakefoundation.org/)
+ * @link          https://cakephp.org CakePHP(tm) Project
+ * @since         3.0.0
+ * @license       https://opensource.org/licenses/MIT MIT License
+ */
+namespace Cake\Queue\Test\TestCase\Enqueue;
+
+use Cake\Queue\Enqueue\SimpleClient;
+use DateTime;
+use Enqueue\Consumption\ChainExtension;
+use Enqueue\Consumption\Extension\LimitConsumedMessagesExtension;
+use Enqueue\Consumption\Extension\LimitConsumptionTimeExtension;
+use Enqueue\Consumption\Result;
+use Interop\Queue\Exception\PurgeQueueNotSupportedException;
+use Interop\Queue\Message;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+
+class SimpleClientTest extends TestCase
+{
+    public static function transportConfigDataProvider()
+    {
+        yield 'amqp_dsn' => [[
+            'transport' => getenv('AMQP_DSN'),
+        ], '+1sec'];
+
+        yield 'dbal_dsn' => [[
+            'transport' => getenv('DOCTRINE_DSN'),
+        ], '+1sec'];
+
+        yield 'rabbitmq_stomp' => [[
+            'transport' => [
+                'dsn' => getenv('RABITMQ_STOMP_DSN'),
+                'lazy' => false,
+                'management_plugin_installed' => true,
+            ],
+        ], '+1sec'];
+
+        yield 'predis_dsn' => [[
+            'transport' => [
+                'dsn' => getenv('PREDIS_DSN'),
+                'lazy' => false,
+            ],
+        ], '+1sec'];
+
+        yield 'fs_dsn' => [[
+            'transport' => 'file://' . sys_get_temp_dir(),
+        ], '+1sec'];
+
+        yield 'sqs' => [[
+            'transport' => [
+                'dsn' => getenv('SQS_DSN'),
+            ],
+        ], '+1sec'];
+
+        yield 'mongodb_dsn' => [[
+            'transport' => getenv('MONGO_DSN'),
+        ], '+1sec'];
+    }
+
+    public function testShouldWorkWithStringDsnConstructorArgument()
+    {
+        $actualMessage = null;
+
+        $client = new SimpleClient(getenv('AMQP_DSN'));
+
+        $client->bindTopic('foo_topic', function (Message $message) use (&$actualMessage) {
+            $actualMessage = $message;
+
+            return Result::ACK;
+        });
+
+        $client->setupBroker();
+        $this->purgeQueue($client);
+
+        $client->sendEvent('foo_topic', 'Hello there!');
+
+        $client->getQueueConsumer()->setReceiveTimeout(200);
+        $client->consume(new ChainExtension([
+            new LimitConsumptionTimeExtension(new DateTime('+1sec')),
+            new LimitConsumedMessagesExtension(2),
+        ]));
+
+        $this->assertInstanceOf(Message::class, $actualMessage);
+        $this->assertSame('Hello there!', $actualMessage->getBody());
+    }
+
+    #[DataProvider('transportConfigDataProvider')]
+    public function testSendEventWithOneSubscriber($config, string $timeLimit)
+    {
+        $actualMessage = null;
+
+        $config['client'] = [
+            'prefix' => str_replace('.', '', uniqid('enqueue', true)),
+            'app_name' => 'simple_client',
+            'router_topic' => 'test',
+            'router_queue' => 'test',
+            'default_queue' => 'test',
+        ];
+
+        $client = new SimpleClient($config);
+
+        $client->bindTopic('foo_topic', function (Message $message) use (&$actualMessage) {
+            $actualMessage = $message;
+
+            return Result::ACK;
+        });
+
+        $client->setupBroker();
+        $this->purgeQueue($client);
+
+        $client->sendEvent('foo_topic', 'Hello there!');
+
+        $client->getQueueConsumer()->setReceiveTimeout(200);
+        $client->consume(new ChainExtension([
+            new LimitConsumptionTimeExtension(new DateTime($timeLimit)),
+            new LimitConsumedMessagesExtension(2),
+        ]));
+
+        $this->assertInstanceOf(Message::class, $actualMessage);
+        $this->assertSame('Hello there!', $actualMessage->getBody());
+    }
+
+    #[DataProvider('transportConfigDataProvider')]
+    public function testSendEventWithTwoSubscriber($config, string $timeLimit)
+    {
+        $received = 0;
+
+        $config['client'] = [
+            'prefix' => str_replace('.', '', uniqid('enqueue', true)),
+            'app_name' => 'simple_client',
+            'router_topic' => 'test',
+            'router_queue' => 'test',
+            'default_queue' => 'test',
+        ];
+
+        $client = new SimpleClient($config);
+
+        $client->bindTopic('foo_topic', function () use (&$received) {
+            ++$received;
+
+            return Result::ACK;
+        });
+        $client->bindTopic('foo_topic', function () use (&$received) {
+            ++$received;
+
+            return Result::ACK;
+        });
+
+        $client->setupBroker();
+        $this->purgeQueue($client);
+
+        $client->sendEvent('foo_topic', 'Hello there!');
+        $client->getQueueConsumer()->setReceiveTimeout(200);
+        $client->consume(new ChainExtension([
+            new LimitConsumptionTimeExtension(new DateTime($timeLimit)),
+            new LimitConsumedMessagesExtension(3),
+        ]));
+
+        $this->assertSame(2, $received);
+    }
+
+    #[DataProvider('transportConfigDataProvider')]
+    public function testSendCommand($config, string $timeLimit)
+    {
+        $received = 0;
+
+        $config['client'] = [
+            'prefix' => str_replace('.', '', uniqid('enqueue', true)),
+            'app_name' => 'simple_client',
+            'router_topic' => 'test',
+            'router_queue' => 'test',
+            'default_queue' => 'test',
+        ];
+
+        $client = new SimpleClient($config);
+
+        $client->bindCommand('foo_command', function () use (&$received) {
+            ++$received;
+
+            return Result::ACK;
+        });
+
+        $client->setupBroker();
+        $this->purgeQueue($client);
+
+        $client->sendCommand('foo_command', 'Hello there!');
+        $client->getQueueConsumer()->setReceiveTimeout(200);
+        $client->consume(new ChainExtension([
+            new LimitConsumptionTimeExtension(new DateTime($timeLimit)),
+            new LimitConsumedMessagesExtension(1),
+        ]));
+
+        $this->assertSame(1, $received);
+    }
+
+    protected function purgeQueue(SimpleClient $client): void
+    {
+        $driver = $client->getDriver();
+
+        $queue = $driver->createQueue($driver->getConfig()->getDefaultQueue());
+
+        try {
+            $client->getDriver()->getContext()->purgeQueue($queue);
+        } catch (PurgeQueueNotSupportedException $e) {
+        }
+    }
+}
diff --git a/tests/TestCase/Job/MailerJobTest.php b/tests/TestCase/Job/MailerJobTest.php
index 112a69d..9a617b9 100644
--- a/tests/TestCase/Job/MailerJobTest.php
+++ b/tests/TestCase/Job/MailerJobTest.php
@@ -71,7 +71,7 @@ public function testExecute()
             ->with(
                 $this->equalTo('welcome'),
                 $this->equalTo($this->args),
-                $this->equalTo($this->headers)
+                $this->equalTo($this->headers),
             )
             ->willReturn(['Message sent']);
 
@@ -79,7 +79,7 @@ public function testExecute()
             ->method('getMailer')
             ->with(
                 $this->equalTo('SampleTest'),
-                $this->equalTo($this->mailerConfig)
+                $this->equalTo($this->mailerConfig),
             )->willReturn($this->mailer);
 
         $message = $this->createMessage();
@@ -101,7 +101,7 @@ public function testExecuteMissingMailerException()
             ->method('getMailer')
             ->with(
                 $this->equalTo('SampleTest'),
-                $this->equalTo($this->mailerConfig)
+                $this->equalTo($this->mailerConfig),
             )->willThrowException(new MissingMailerException('Missing mailer for testExecuteMissingMailerException'));
 
         $message = $this->createMessage();
@@ -121,7 +121,7 @@ public function testExecuteBadMethodCallException()
             ->with(
                 $this->equalTo('welcome'),
                 $this->equalTo($this->args),
-                $this->equalTo($this->headers)
+                $this->equalTo($this->headers),
             )
             ->willThrowException(new BadMethodCallException('Welcome is not a valid method'));
 
@@ -129,7 +129,7 @@ public function testExecuteBadMethodCallException()
             ->method('getMailer')
             ->with(
                 $this->equalTo('SampleTest'),
-                $this->equalTo($this->mailerConfig)
+                $this->equalTo($this->mailerConfig),
             )->willReturn($this->mailer);
 
         $message = $this->createMessage();
diff --git a/tests/TestCase/Listener/FailedJobsListenerTest.php b/tests/TestCase/Listener/FailedJobsListenerTest.php
index d08f0cf..3a5227e 100644
--- a/tests/TestCase/Listener/FailedJobsListenerTest.php
+++ b/tests/TestCase/Listener/FailedJobsListenerTest.php
@@ -77,7 +77,7 @@ public function testFailedJobIsAddedWhenEventIsFired()
         $event = new Event(
             'Consumption.LimitAttemptsExtension.failed',
             $message,
-            ['exception' => 'some message']
+            ['exception' => 'some message'],
         );
 
         /** @var \Cake\Queue\Model\Table\FailedJobsTable $failedJobsTable */
@@ -161,7 +161,7 @@ public function testStoreFailedJobException($eventData, $exceptionMessage)
         $event = new Event(
             'Consumption.LimitAttemptsExtension.failed',
             $message,
-            $eventData
+            $eventData,
         );
 
         $this->expectException(RuntimeException::class);
diff --git a/tests/TestCase/Mailer/Transport/QueueTransportTest.php b/tests/TestCase/Mailer/Transport/QueueTransportTest.php
index 6539e0b..61ff9a5 100644
--- a/tests/TestCase/Mailer/Transport/QueueTransportTest.php
+++ b/tests/TestCase/Mailer/Transport/QueueTransportTest.php
@@ -67,7 +67,7 @@ public function testSend()
                 'returnPath',
                 'cc',
                 'bcc',
-            ]
+            ],
         );
 
         $expected = ['headers' => $headers, 'message' => 'Message has been enqueued'];
@@ -93,7 +93,7 @@ public function testSendCustomTransport()
             'queue' => 'default',
             'url' => $this->getFsQueueUrl(),
         ]);
-        $message = (new Message());
+        $message = new Message();
 
         $transport = new QueueTransport([
             'transport' => SmtpTransport::class,
@@ -124,7 +124,7 @@ public function testSendBcTransport()
             'queue' => 'default',
             'url' => $this->getFsQueueUrl(),
         ]);
-        $message = (new Message());
+        $message = new Message();
 
         $transport = new QueueTransport([
             'transport' => SmtpTransport::class,
diff --git a/tests/TestCase/Task/JobTaskTest.php b/tests/TestCase/Task/JobTaskTest.php
index 8236c86..a49324f 100644
--- a/tests/TestCase/Task/JobTaskTest.php
+++ b/tests/TestCase/Task/JobTaskTest.php
@@ -73,7 +73,7 @@ public function testMain()
         $this->assertOutputContains('Creating file ' . $this->generatedFile);
         $this->assertSameAsFile(
             $this->comparisonDir . 'JobTask.php',
-            file_get_contents($this->generatedFile)
+            file_get_contents($this->generatedFile),
         );
     }
 
@@ -87,7 +87,7 @@ public function testMainWithUnique()
         $this->assertOutputContains('Creating file ' . $this->generatedFile);
         $this->assertSameAsFile(
             $this->comparisonDir . 'JobTaskWithUnique.php',
-            file_get_contents($this->generatedFile)
+            file_get_contents($this->generatedFile),
         );
     }
 
@@ -101,7 +101,7 @@ public function testMainWithMaxAttempts()
         $this->assertOutputContains('Creating file ' . $this->generatedFile);
         $this->assertSameAsFile(
             $this->comparisonDir . 'JobTaskWithMaxAttempts.php',
-            file_get_contents($this->generatedFile)
+            file_get_contents($this->generatedFile),
         );
     }
 }