Merge pull request #703 from php-enqueue/client-generic-driver-should-able-send-to-router-processor

[client] sendToProcessor should able to send message to router processor.

Author: makasim

pkg/enqueue/Client/Driver/GenericDriver.php

@@ -71,16 +71,18 @@ public function sendToProcessor(Message $message): DriverSendResult
 
         /** @var InteropQueue $queue */
         $queue = null;
-        if ($topic && $processor = $message->getProperty(Config::PROCESSOR)) {
+        $routerProcessor = $this->config->getRouterProcessor();
+        $processor = $message->getProperty(Config::PROCESSOR);
+        if ($topic && $processor && $processor !== $routerProcessor) {
             $route = $this->routeCollection->topicAndProcessor($topic, $processor);
             if (false == $route) {
                 throw new \LogicException(sprintf('There is no route for topic "%s" and processor "%s"', $topic, $processor));
             }
 
             $message->setProperty(Config::PROCESSOR, $route->getProcessor());
             $queue = $this->createRouteQueue($route);
-        } elseif ($topic && false == $message->getProperty(Config::PROCESSOR)) {
-            $message->setProperty(Config::PROCESSOR, $this->config->getRouterProcessor());
+        } elseif ($topic && (false == $processor || $processor === $routerProcessor)) {
+            $message->setProperty(Config::PROCESSOR, $routerProcessor);
 
             $queue = $this->createQueue($this->config->getRouterQueue());
         } elseif ($command) {

pkg/enqueue/Tests/Client/Driver/GenericDriverTestsTrait.php

@@ -445,6 +445,54 @@ public function testThrowIfCommandSetOnSendToRouter()
         $driver->sendToRouter($message);
     }
 
+    public function testShouldSendMessageToRouterProcessor()
+    {
+        $queue = $this->createQueue('');
+        $transportMessage = $this->createMessage();
+
+        $producer = $this->createProducerMock();
+        $producer
+            ->expects($this->once())
+            ->method('send')
+            ->with($this->identicalTo($queue), $this->identicalTo($transportMessage))
+        ;
+        $context = $this->createContextMock();
+        $context
+            ->expects($this->once())
+            ->method('createQueue')
+            ->with($this->getDefaultQueueTransportName())
+            ->willReturn($queue)
+        ;
+        $context
+            ->expects($this->once())
+            ->method('createProducer')
+            ->willReturn($producer)
+        ;
+        $context
+            ->expects($this->once())
+            ->method('createMessage')
+            ->willReturn($transportMessage)
+        ;
+
+        $config = $this->createDummyConfig();
+
+        $driver = $this->createDriver(
+            $context,
+            $config,
+            new RouteCollection([
+                new Route('topic', Route::TOPIC, 'processor', [
+                    'queue' => 'custom',
+                ]),
+            ])
+        );
+
+        $message = new Message();
+        $message->setProperty(Config::TOPIC, 'topic');
+        $message->setProperty(Config::PROCESSOR, $config->getRouterProcessor());
+
+        $driver->sendToProcessor($message);
+    }
+
     public function testShouldSendTopicMessageToProcessorToDefaultQueue()
     {
         $queue = $this->createQueue('');