Push redelivered messages to queue end; Examples fixes

Makefile

@@ -1,5 +1,11 @@
-phpstan:
+#!make
+
+help: ## Show this help
+	@printf "\033[33m%s:\033[0m\n" 'Run: make <target> where <target> is one of the following'
+	@awk 'BEGIN {FS = ":.*?## "} /^[a-zA-Z_-]+:.*?## / {printf "  \033[32m%-18s\033[0m %s\n", $$1, $$2}' $(MAKEFILE_LIST)
+
+phpstan: ## Run PHPStan
 	vendor/bin/phpstan analyse --level max src tests
 
-test: phpstan
+test: phpstan ## Run PHPUnit
 	vendor/bin/phpunit --verbose

examples/multi/builder.php

@@ -5,17 +5,16 @@
 use Onliner\CommandBus\Builder;
 use Onliner\CommandBus\Remote\AMQP\Transport;
 use Onliner\CommandBus\Remote\RemoteExtension;
-use Onliner\CommandBus\Remote\Transport;
+use Onliner\CommandBus\Remote\Transport\MultiTransport;
 
 require __DIR__ . '/../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
 $transportFoo = Transport::create('amqp://guest:guest@localhost:5672', 'foo');
 $transportBar = Transport::create('amqp://guest:guest@localhost:5673', 'bar');
 
-$transport = new Transport\MultiTransport($transportFoo);
+$transport = new MultiTransport($transportFoo);
 $transport->add('Bar\*', $transportBar);
 
 return (new Builder())
-    ->use(new RemoteExtension($transport))
-;
+    ->use(new RemoteExtension($transport));

examples/multi/consumer_bar.php

@@ -12,13 +12,7 @@
     echo sprintf('Hello %s from bar!', $command->name), PHP_EOL;
 });
 
-$dispatcher = $builder->build();
-
-$transport = Transport::create('amqp://guest:guest@localhost:5673', [
-    'exchange' => 'bar',
-]);
-
-/** @var Consumer $consumer */
+$transport = Transport::create('amqp://guest:guest@localhost:5673');
 $consumer = $transport->consume();
-$consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->listen('#', 'bar');
+$consumer->run($builder->build());

examples/multi/consumer_foo.php

@@ -12,13 +12,7 @@
     echo sprintf('Hello %s from foo!', $command->name), PHP_EOL;
 });
 
-$dispatcher = $builder->build();
-
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-]);
-
-/** @var Consumer $consumer */
+$transport = Transport::create('amqp://guest:guest@localhost:5672');
 $consumer = $transport->consume();
-$consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->listen('#', 'foo');
+$consumer->run($builder->build());

examples/multi/messages.php

@@ -5,35 +5,17 @@
 namespace Foo {
     class Hello
     {
-        /**
-         * @var string
-         */
-        public $name;
-
-        /**
-         * @param string $name
-         */
-        public function __construct(string $name)
-        {
-            $this->name = $name;
-        }
+        public function __construct(
+            public string $name,
+        ) {}
     }
 }
 
 namespace Bar {
     class Hello
     {
-        /**
-         * @var string
-         */
-        public $name;
-
-        /**
-         * @param string $name
-         */
-        public function __construct(string $name)
-        {
-            $this->name = $name;
-        }
+        public function __construct(
+            public string $name,
+        ) {}
     }
 }

examples/remote/builder.php

@@ -9,10 +9,7 @@
 require __DIR__ . '/../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-]);
+$transport = Transport::create('amqp://guest:guest@localhost:5672', 'foo');
 
 return (new Builder())
-    ->use(new RemoteExtension($transport))
-;
+    ->use(new RemoteExtension($transport));

examples/remote/consumer.php

@@ -15,30 +15,25 @@
     echo 'CONTENT: ', $command->content, \PHP_EOL;
 });
 
-$dispatcher = $builder->build();
-
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-]);
-
-/** @var Consumer $consumer */
+$transport = Transport::create('amqp://guest:guest@localhost:5672');
 $consumer = $transport->consume();
 
 $pattern  = $argv[1] ?? '#';
 $priority = isset($argv[2]) ? (int) $argv[2] : 0;
 
 if ($priority === 0) {
-    $consumer->listen($pattern);
+    $consumer->listen($pattern, 'foo');
 } else {
     $consumer->consume(Queue::create([
         'pattern' => $pattern,
+        'bindings' => ['foo'],
         'args' => [
             Queue::MAX_PRIORITY => $priority,
         ],
     ]));
 }
 
-$consumer->run($dispatcher, [
+$consumer->run($builder->build(), [
     Consumer::OPTION_ATTEMPTS => 10,
     Consumer::OPTION_INTERVAL => 100000, // 100 ms
 ]);

examples/remote/messages.php

@@ -4,30 +4,9 @@
 
 class SendEmail
 {
-    /**
-     * @var string
-     */
-    public $to;
-
-    /**
-     * @var string
-     */
-    public $subject;
-
-    /**
-     * @var string
-     */
-    public $content;
-
-    /**
-     * @param string $to
-     * @param string $subject
-     * @param string $content
-     */
-    public function __construct(string $to, string $subject, string $content)
-    {
-        $this->to      = $to;
-        $this->subject = $subject;
-        $this->content = $content;
-    }
+    public function __construct(
+        public string $to,
+        public string $subject,
+        public string $content,
+    ) {}
 }

examples/router/builder.php

@@ -5,18 +5,13 @@
 use Onliner\CommandBus\Builder;
 use Onliner\CommandBus\Remote\AMQP\Transport;
 use Onliner\CommandBus\Remote\RemoteExtension;
-use Onliner\CommandBus\Remote\Transport;
 
 require __DIR__ . '/../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-    'routes' => [
-        'Bar\*' => 'bar',
-    ],
+$transport = Transport::create('amqp://guest:guest@localhost:5672', 'foo', [
+    'Bar\*' => 'bar',
 ]);
 
 return (new Builder())
-    ->use(new RemoteExtension($transport))
-;
+    ->use(new RemoteExtension($transport));

examples/router/consumer_bar.php

@@ -13,12 +13,9 @@
 });
 
 $dispatcher = $builder->build();
-
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'bar',
-]);
+$transport = Transport::create('amqp://guest:guest@localhost:5672');
 
 /** @var Consumer $consumer */
 $consumer = $transport->consume();
-$consumer->listen('#');
+$consumer->listen('#', 'bar');
 $consumer->run($dispatcher);

examples/router/consumer_foo.php

@@ -14,11 +14,9 @@
 
 $dispatcher = $builder->build();
 
-$transport = Transport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-]);
+$transport = Transport::create('amqp://guest:guest@localhost:5672');
 
 /** @var Consumer $consumer */
 $consumer = $transport->consume();
-$consumer->listen('#');
+$consumer->listen('#', 'foo');
 $consumer->run($dispatcher);

examples/router/messages.php

@@ -5,35 +5,17 @@
 namespace Foo {
     class Hello
     {
-        /**
-         * @var string
-         */
-        public $name;
-
-        /**
-         * @param string $name
-         */
-        public function __construct(string $name)
-        {
-            $this->name = $name;
-        }
+        public function __construct(
+            public string $name,
+        ) {}
     }
 }
 
 namespace Bar {
     class Hello
     {
-        /**
-         * @var string
-         */
-        public $name;
-
-        /**
-         * @param string $name
-         */
-        public function __construct(string $name)
-        {
-            $this->name = $name;
-        }
+        public function __construct(
+            public string $name,
+        ) {}
     }
 }

src/Remote/AMQP/Consumer.php

@@ -92,9 +92,9 @@ public function stop(): void
     private function channel(Dispatcher $dispatcher, array $options): AMQPChannel
     {
         $channel = $this->connect($options);
-        $handler = function (AMQPMessage $message) use ($dispatcher) {
+        $handler = function (AMQPMessage $message) use ($channel, $dispatcher) {
             try {
-                $this->handle($message, $dispatcher);
+                $this->handle($message, $channel, $dispatcher);
             } catch (Throwable $error) {
                 $this->logger->error((string) $error);
             } finally {
@@ -143,10 +143,16 @@ private function connect(array $options): AMQPChannel
         throw new AMQPIOException();
     }
 
-    private function handle(AMQPMessage $message, Dispatcher $dispatcher): void
+    private function handle(AMQPMessage $message, AMQPChannel $channel, Dispatcher $dispatcher): void
     {
         if ($message->isRedelivered()) {
-            throw new AMQPIOException('Message redelivered');
+            $channel->basic_publish(
+                new AMQPMessage($message->body, $message->get_properties()),
+                (string) $message->getExchange(),
+                (string) $message->getRoutingKey(),
+            );
+
+            return;
         }
 
         $envelope = $this->packager->unpack($message);

src/Remote/Gateway.php

@@ -5,6 +5,7 @@
 namespace Onliner\CommandBus\Remote;
 
 use Onliner\CommandBus\Context;
+use Onliner\CommandBus\Remote\AMQP\Packager;
 
 final class Gateway
 {
@@ -24,6 +25,8 @@ public function receive(Envelope $envelope, Context $context): void
     {
         $message = $this->serializer->unserialize($envelope);
 
-        $context->execute($message, $envelope->headers);
+        $context->dispatch($message, array_replace($envelope->headers, [
+            Packager::OPTION_LOCAL => true,
+        ]));
     }
 }

tests/Remote/AMQP/TransportTest.php

@@ -57,7 +57,7 @@ public function testSend(): void
         $channel
             ->expects(self::exactly(2))
             ->method('basic_publish')
-            ->with($message, 'amqp.topic', strtolower(str_replace('\\', '.', $envelope->class)), false, false)
+            ->with($message, 'foo', strtolower(str_replace('\\', '.', $envelope->class)), false, false)
         ;
 
         $connector = self::createMock(Connector::class);
@@ -67,7 +67,7 @@ public function testSend(): void
             ->willReturn($channel)
         ;
 
-        $transport = new Transport($connector, new Packager(), new SimpleRouter());
+        $transport = new Transport($connector, new Packager(), new SimpleRouter('foo'));
         $transport->send($envelope);
         $transport->send($envelope);
     }