Added mandatory support; Simplify AMQP transport; Allow queue multipl…

…e bindings
onliner · Aug 13, 2024 · 3ab416a · 3ab416a
1 parent 23e661a
commit 3ab416a
Show file tree

Hide file tree

Showing 23 changed files with 223 additions and 197 deletions.
diff --git a/examples/multi/builder.php b/examples/multi/builder.php
@@ -3,19 +3,15 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
+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';
 
-$transportFoo = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
-    'exchange' => 'foo',
-]);
-$transportBar = AMQPTransport::create('amqp://guest:guest@localhost:5673', [
-    'exchange' => 'bar',
-]);
+$transportFoo = Transport::create('amqp://guest:guest@localhost:5672', 'foo');
+$transportBar = Transport::create('amqp://guest:guest@localhost:5673', 'bar');
 
 $transport = new Transport\MultiTransport($transportFoo);
 $transport->add('Bar\*', $transportBar);

diff --git a/examples/multi/consumer_bar.php b/examples/multi/consumer_bar.php
@@ -3,8 +3,8 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
-use Onliner\CommandBus\Remote\AMQP\AMQPConsumer;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
 
 /** @var Builder $builder */
 $builder = require __DIR__ . '/builder.php';
@@ -14,11 +14,11 @@
 
 $dispatcher = $builder->build();
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5673', [
+$transport = Transport::create('amqp://guest:guest@localhost:5673', [
     'exchange' => 'bar',
 ]);
 
-/** @var AMQPConsumer $consumer */
+/** @var Consumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
 $consumer->run($dispatcher);
diff --git a/examples/multi/consumer_foo.php b/examples/multi/consumer_foo.php
@@ -3,8 +3,8 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
-use Onliner\CommandBus\Remote\AMQP\AMQPConsumer;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
 
 /** @var Builder $builder */
 $builder = require __DIR__ . '/builder.php';
@@ -14,11 +14,11 @@
 
 $dispatcher = $builder->build();
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'foo',
 ]);
 
-/** @var AMQPConsumer $consumer */
+/** @var Consumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
 $consumer->run($dispatcher);
diff --git a/examples/remote/builder.php b/examples/remote/builder.php
@@ -3,13 +3,13 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
+use Onliner\CommandBus\Remote\AMQP\Transport;
 use Onliner\CommandBus\Remote\RemoteExtension;
 
 require __DIR__ . '/../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'foo',
 ]);
 

diff --git a/examples/remote/consumer.php b/examples/remote/consumer.php
@@ -3,8 +3,8 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
-use Onliner\CommandBus\Remote\AMQP\AMQPConsumer;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
 use Onliner\CommandBus\Remote\AMQP\Queue;
 
 /** @var Builder $builder */
@@ -17,11 +17,11 @@
 
 $dispatcher = $builder->build();
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'foo',
 ]);
 
-/** @var AMQPConsumer $consumer */
+/** @var Consumer $consumer */
 $consumer = $transport->consume();
 
 $pattern  = $argv[1] ?? '#';
@@ -39,6 +39,6 @@
 }
 
 $consumer->run($dispatcher, [
-    AMQPConsumer::OPTION_ATTEMPTS => 10,
-    AMQPConsumer::OPTION_INTERVAL => 100000, // 100 ms
+    Consumer::OPTION_ATTEMPTS => 10,
+    Consumer::OPTION_INTERVAL => 100000, // 100 ms
 ]);
diff --git a/examples/router/builder.php b/examples/router/builder.php
@@ -3,14 +3,14 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
+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 = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'foo',
     'routes' => [
         'Bar\*' => 'bar',

diff --git a/examples/router/consumer_bar.php b/examples/router/consumer_bar.php
@@ -3,8 +3,8 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
-use Onliner\CommandBus\Remote\AMQP\AMQPConsumer;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
 
 /** @var Builder $builder */
 $builder = require __DIR__ . '/builder.php';
@@ -14,11 +14,11 @@
 
 $dispatcher = $builder->build();
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'bar',
 ]);
 
-/** @var AMQPConsumer $consumer */
+/** @var Consumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
 $consumer->run($dispatcher);
diff --git a/examples/router/consumer_foo.php b/examples/router/consumer_foo.php
@@ -3,8 +3,8 @@
 declare(strict_types=1);
 
 use Onliner\CommandBus\Builder;
-use Onliner\CommandBus\Remote\AMQP\AMQPTransport;
-use Onliner\CommandBus\Remote\AMQP\AMQPConsumer;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
 
 /** @var Builder $builder */
 $builder = require __DIR__ . '/builder.php';
@@ -14,11 +14,11 @@
 
 $dispatcher = $builder->build();
 
-$transport = AMQPTransport::create('amqp://guest:guest@localhost:5672', [
+$transport = Transport::create('amqp://guest:guest@localhost:5672', [
     'exchange' => 'foo',
 ]);
 
-/** @var AMQPConsumer $consumer */
+/** @var Consumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
 $consumer->run($dispatcher);
diff --git a/src/Context.php b/src/Context.php
@@ -8,8 +8,6 @@
 
 final class Context
 {
-    private const OPTION_LOCAL = 'local';
-
     /**
      * @param array<string, mixed> $options
      */
@@ -27,16 +25,6 @@ public function dispatch(object $message, array $options = []): void
         $this->dispatcher->dispatch($message, $options);
     }
 
-    /**
-     * @param array<string, mixed> $options
-     */
-    public function execute(object $message, array $options = []): void
-    {
-        $this->dispatcher->dispatch($message, array_replace($options, [
-            self::OPTION_LOCAL => true,
-        ]));
-    }
-
     /**
      * @param array<string, mixed> $options
      */
@@ -78,9 +66,4 @@ public function del(string $option): self
 
         return $this;
     }
-
-    public function isLocal(): bool
-    {
-        return $this->has(self::OPTION_LOCAL);
-    }
 }
diff --git a/src/Remote/AMQP/AMQPTransport.php b/src/Remote/AMQP/AMQPTransport.php
diff --git a/src/Remote/AMQP/AMQPConsumer.php → src/Remote/AMQP/Consumer.php b/src/Remote/AMQP/AMQPConsumer.php → src/Remote/AMQP/Consumer.php
@@ -5,18 +5,16 @@
 namespace Onliner\CommandBus\Remote\AMQP;
 
 use Onliner\CommandBus\Dispatcher;
-use Onliner\CommandBus\Remote\Consumer;
-use Onliner\CommandBus\Remote\Envelope;
+use Onliner\CommandBus\Remote\Consumer as ConsumerContract;
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Exception\AMQPConnectionClosedException;
 use PhpAmqpLib\Exception\AMQPIOException;
 use PhpAmqpLib\Message\AMQPMessage;
-use PhpAmqpLib\Wire\AMQPAbstractCollection;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Throwable;
 
-final class AMQPConsumer implements Consumer
+final class Consumer implements ConsumerContract
 {
     public const
         OPTION_ATTEMPTS = 'attempts',
@@ -39,15 +37,18 @@ final class AMQPConsumer implements Consumer
 
     public function __construct(
         private Connector $connector,
-        private Exchange $exchange,
+        private Packager $packager,
         LoggerInterface $logger = null,
     ) {
         $this->logger = $logger ?? new NullLogger();
     }
 
-    public function listen(string $pattern): void
+    /**
+     * @param string|array<string> $bindings
+     */
+    public function listen(string $name, array|string $bindings = [], Flags $flags = null): void
     {
-        $this->consume(new Queue($pattern, $pattern, $this->exchange->flags));
+        $this->consume(new Queue($name, $name, (array) $bindings, $flags ?? Flags::default()));
     }
 
     public function consume(Queue $queue): void
@@ -101,10 +102,8 @@ private function channel(Dispatcher $dispatcher, array $options): AMQPChannel
             }
         };
 
-        $this->exchange->declare($channel);
-
         foreach ($this->queues as $queue) {
-            $queue->consume($channel, $this->exchange, $handler);
+            $queue->consume($channel, $handler);
         }
 
         return $channel;
@@ -146,31 +145,12 @@ private function connect(array $options): AMQPChannel
 
     private function handle(AMQPMessage $message, Dispatcher $dispatcher): void
     {
-        $headers = $message->get('application_headers');
-
-        if (!$headers instanceof AMQPAbstractCollection) {
-            $this->logger->warning('Message headers not found.');
-
-            return;
-        }
-
-        $headers = array_replace($headers->getNativeData(), [
-            Exchange::HEADER_EXCHANGE => $message->getExchange(),
-            Exchange::HEADER_REDELIVERED => $message->isRedelivered(),
-            Exchange::HEADER_ROUTING_KEY => $message->getRoutingKey(),
-            Exchange::HEADER_CONSUMER_TAG => $message->getConsumerTag(),
-            Exchange::HEADER_DELIVERY_TAG => $message->getDeliveryTag(),
-        ]);
-
-        if (!isset($headers[Exchange::HEADER_MESSAGE_TYPE])) {
-            $this->logger->warning(sprintf('Header "%s" not found in message.', Exchange::HEADER_MESSAGE_TYPE));
-
-            return;
+        if ($message->isRedelivered()) {
+            throw new AMQPIOException('Message redelivered');
         }
 
-        /** @var class-string $class */
-        $class = $headers[Exchange::HEADER_MESSAGE_TYPE];
+        $envelope = $this->packager->unpack($message);
 
-        $dispatcher->dispatch(new Envelope($class, $message->getBody(), $headers));
+        $dispatcher->dispatch($envelope);
     }
 }