Merge pull request #101 from php-enqueue/client-router-impro

[client] Fixes edge cases in client's routing logic.

Author: makasim

pkg/enqueue/Client/ConsumptionExtension/SetRouterPropertiesExtension.php

@@ -35,8 +35,13 @@ public function onPreReceived(Context $context)
             return;
         }
 
+        $config = $this->driver->getConfig();
+        if ($context->getPsrQueue()->getQueueName() != $config->createTransportQueueName($config->getRouterQueueName())) {
+            return;
+        }
+
         // RouterProcessor is our default message processor when that header is not set
-        $message->setProperty(Config::PARAMETER_PROCESSOR_NAME, $this->driver->getConfig()->getRouterProcessorName());
-        $message->setProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME, $this->driver->getConfig()->getRouterQueueName());
+        $message->setProperty(Config::PARAMETER_PROCESSOR_NAME, $config->getRouterProcessorName());
+        $message->setProperty(Config::PARAMETER_PROCESSOR_QUEUE_NAME, $config->getRouterQueueName());
     }
 }

pkg/enqueue/Client/RouterProcessor.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Client;
 
+use Enqueue\Consumption\Result;
 use Enqueue\Psr\PsrContext;
 use Enqueue\Psr\PsrMessage;
 use Enqueue\Psr\PsrProcessor;
@@ -45,7 +46,7 @@ public function process(PsrMessage $message, PsrContext $context)
     {
         $topicName = $message->getProperty(Config::PARAMETER_TOPIC_NAME);
         if (false == $topicName) {
-            throw new \LogicException(sprintf(
+            return Result::reject(sprintf(
                 'Got message without required parameter: "%s"',
                 Config::PARAMETER_TOPIC_NAME
             ));

pkg/enqueue/Tests/Client/ConsumptionExtension/SetRouterPropertiesExtensionTest.php

@@ -8,6 +8,7 @@
 use Enqueue\Consumption\Context;
 use Enqueue\Consumption\ExtensionInterface;
 use Enqueue\Null\NullMessage;
+use Enqueue\Null\NullQueue;
 use Enqueue\Psr\PsrContext;
 use Enqueue\Test\ClassExtensionTrait;
 use PHPUnit\Framework\TestCase;
@@ -26,13 +27,13 @@ public function testCouldBeConstructedWithRequiredArguments()
         new SetRouterPropertiesExtension($this->createDriverMock());
     }
 
-    public function testShouldSetRouterProcessorPropertyIfNotSet()
+    public function testShouldSetRouterProcessorPropertyIfNotSetAndOnRouterQueue()
     {
-        $config = new Config('', '', '', 'router-queue', '', 'router-processor-name');
+        $config = Config::create('test', '', '', 'router-queue', '', 'router-processor-name');
 
         $driver = $this->createDriverMock();
         $driver
-            ->expects(self::exactly(2))
+            ->expects($this->once())
             ->method('getConfig')
             ->willReturn($config)
         ;
@@ -41,6 +42,7 @@ public function testShouldSetRouterProcessorPropertyIfNotSet()
 
         $context = new Context($this->createPsrContextMock());
         $context->setPsrMessage($message);
+        $context->setPsrQueue(new NullQueue('test.router-queue'));
 
         $extension = new SetRouterPropertiesExtension($driver);
         $extension->onPreReceived($context);
@@ -51,11 +53,34 @@ public function testShouldSetRouterProcessorPropertyIfNotSet()
         ], $message->getProperties());
     }
 
+    public function testShouldNotSetRouterProcessorPropertyIfNotSetAndNotOnRouterQueue()
+    {
+        $config = Config::create('test', '', '', 'router-queue', '', 'router-processor-name');
+
+        $driver = $this->createDriverMock();
+        $driver
+            ->expects($this->once())
+            ->method('getConfig')
+            ->willReturn($config)
+        ;
+
+        $message = new NullMessage();
+
+        $context = new Context($this->createPsrContextMock());
+        $context->setPsrMessage($message);
+        $context->setPsrQueue(new NullQueue('test.another-queue'));
+
+        $extension = new SetRouterPropertiesExtension($driver);
+        $extension->onPreReceived($context);
+
+        $this->assertEquals([], $message->getProperties());
+    }
+
     public function testShouldNotSetAnyPropertyIfProcessorNamePropertyAlreadySet()
     {
         $driver = $this->createDriverMock();
         $driver
-            ->expects(self::never())
+            ->expects($this->never())
             ->method('getConfig')
         ;
 

pkg/enqueue/Tests/Client/RouterProcessorTest.php

@@ -30,14 +30,15 @@ public function testCouldBeConstructedWithSessionAndRoutes()
         $this->assertAttributeEquals($routes, 'routes', $router);
     }
 
-    public function testShouldThrowExceptionIfTopicNameParameterIsNotSet()
+    public function testShouldRejectIfTopicNameParameterIsNotSet()
     {
         $router = new RouterProcessor($this->createDriverMock());
 
-        $this->expectException(\LogicException::class);
-        $this->expectExceptionMessage('Got message without required parameter: "enqueue.topic_name"');
+        $result = $router->process(new NullMessage(), new NullContext());
 
-        $router->process(new NullMessage(), new NullContext());
+        $this->assertInstanceOf(Result::class, $result);
+        $this->assertEquals(Result::REJECT, $result->getStatus());
+        $this->assertEquals('Got message without required parameter: "enqueue.topic_name"', $result->getReason());
     }
 
     public function testShouldRouteOriginalMessageToRecipient()