PR fix

examples/dispatch_deferred_after_handle.php → examples/deferred.php

@@ -9,22 +9,16 @@
 
 class Task
 {
-    public $task;
-
-    public function __construct(string $task)
-    {
-        $this->task = $task;
-    }
+    public function __construct(
+        public string $task,
+    ) {}
 }
 
 class ResultMessage
 {
-    public $message;
-
-    public function __construct(string $message)
-    {
-        $this->message = $message;
-    }
+    public function __construct(
+        public string $message,
+    ) {}
 }
 
 class TaskHandler
@@ -61,4 +55,3 @@ public function __invoke(ResultMessage $command, Context $context)
  * Task: build report processing end
  * Result: success
  */
-

examples/hello.php

@@ -8,12 +8,9 @@
 
 class Hello
 {
-    public $message;
-
-    public function __construct(string $message)
-    {
-        $this->message = $message;
-    }
+    public function __construct(
+        public string $message,
+    ) {}
 }
 
 $dispatcher = (new Builder())

examples/invoke.php

@@ -8,12 +8,9 @@
 
 class Hello
 {
-    public $message;
-
-    public function __construct(string $message)
-    {
-        $this->message = $message;
-    }
+    public function __construct(
+        public string $message,
+    ) {}
 }
 
 class HelloHandler

examples/remote/builder.php → examples/remote/bindings/builder.php

@@ -6,7 +6,7 @@
 use Onliner\CommandBus\Remote\AMQP\Transport;
 use Onliner\CommandBus\Remote\RemoteExtension;
 
-require __DIR__ . '/../../vendor/autoload.php';
+require __DIR__ . '/../../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
 $transport = Transport::create('amqp://guest:guest@localhost:5672', 'foo');

examples/remote/bindings/consumer.php

@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+use Onliner\CommandBus\Builder;
+use Onliner\CommandBus\Context;
+use Onliner\CommandBus\Remote\AMQP\Exchange;
+use Onliner\CommandBus\Remote\AMQP\Flags;
+use Onliner\CommandBus\Remote\AMQP\Packager;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\AMQP\Consumer;
+use Onliner\CommandBus\Remote\AMQP\Queue;
+
+/** @var Builder $builder */
+$builder = require __DIR__ . '/builder.php';
+$builder->handle(SendEmail::class, function (SendEmail $command, Context $context) {
+    $exchange = $context->get(Packager::OPTION_EXCHANGE);
+    $routingKey = $context->get(Packager::OPTION_ROUTING_KEY);
+
+    echo sprintf('Received message from %s with routing key %s', $exchange, $routingKey), PHP_EOL;
+
+    // Throw exception to trigger DLE
+    if ($exchange === 'foo') {
+        throw new Exception("Something went wrong...");
+    }
+});
+
+$transport = Transport::create('amqp://guest:guest@localhost:5672');
+$transport->declare(Exchange::create(['name' => 'dle']));
+$transport->declare(Exchange::create(['name' => 'foo']));
+
+$consumer = $transport->consume();
+$consumer->consume(new Queue('my-queue', [
+    'foo' => '#',
+    'dle' => 'sendemail',
+], Flags::default(), args: [
+    Queue::DEAD_LETTER => 'dle',
+]));
+
+$consumer->run($builder->build(), [
+    Consumer::OPTION_ATTEMPTS => 10,
+    Consumer::OPTION_INTERVAL => 100000, // 100 ms
+]);

examples/multi/builder.php → examples/remote/multi/builder.php

@@ -7,7 +7,7 @@
 use Onliner\CommandBus\Remote\RemoteExtension;
 use Onliner\CommandBus\Remote\Transport\MultiTransport;
 
-require __DIR__ . '/../../vendor/autoload.php';
+require __DIR__ . '/../../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
 $transportFoo = Transport::create('amqp://guest:guest@localhost:5672', 'foo');

examples/router/builder.php → examples/remote/router/builder.php

@@ -6,7 +6,7 @@
 use Onliner\CommandBus\Remote\AMQP\Transport;
 use Onliner\CommandBus\Remote\RemoteExtension;
 
-require __DIR__ . '/../../vendor/autoload.php';
+require __DIR__ . '/../../../vendor/autoload.php';
 require __DIR__ . '/messages.php';
 
 $transport = Transport::create('amqp://guest:guest@localhost:5672', 'foo', [

examples/remote/simple/builder.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+use Onliner\CommandBus\Builder;
+use Onliner\CommandBus\Remote\AMQP\Transport;
+use Onliner\CommandBus\Remote\RemoteExtension;
+
+require __DIR__ . '/../../../vendor/autoload.php';
+require __DIR__ . '/messages.php';
+
+$transport = Transport::create('amqp://guest:guest@localhost:5672', 'foo');
+
+return (new Builder())
+    ->use(new RemoteExtension($transport));

examples/remote/simple/messages.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+class SendEmail
+{
+    public function __construct(
+        public string $to,
+        public string $subject,
+        public string $content,
+    ) {}
+}

examples/remote/simple/sender.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+use Onliner\CommandBus\Builder;
+
+/** @var Builder $builder */
+$builder = require __DIR__ . '/builder.php';
+
+$dispatcher = $builder->build();
+$dispatcher->dispatch(new SendEmail('[email protected]', 'Onliner', 'Hello world!'));

src/Middleware/LoggerMiddleware.php

@@ -22,7 +22,10 @@ public function call(object $message, Context $context, callable $next): void
         try {
             $next($message, $context);
         } catch (Throwable $error) {
-            $this->logger->log($this->level, $error->getMessage());
+            $this->logger->log($this->level, $error->getMessage(), [
+                'file' => $error->getFile(),
+                'line' => $error->getLine(),
+            ]);
 
             throw $error;
         }

src/Remote/AMQP/Connector.php

@@ -9,13 +9,13 @@
 use PhpAmqpLib\Connection\AMQPStreamConnection;
 use PhpAmqpLib\Connection\Heartbeat\PCNTLHeartbeatSender;
 
-class Connector
+final class Connector
 {
     private ?AMQPChannel $channel = null;
     private ?PCNTLHeartbeatSender $heartbeats = null;
 
     /**
-     * @param array<array<mixed>>      $hosts
+     * @param array<array<mixed>> $hosts
      * @param array<string|int, mixed> $options
      */
     public function __construct(

src/Remote/AMQP/Consumer.php

@@ -38,17 +38,17 @@ final class Consumer implements ConsumerContract
     public function __construct(
         private Connector $connector,
         private Packager $packager,
-        LoggerInterface $logger = null,
+        ?LoggerInterface $logger = null,
     ) {
         $this->logger = $logger ?? new NullLogger();
     }
 
     /**
      * @param string|array<string> $bindings
      */
-    public function listen(string $name, array|string $bindings = [], Flags $flags = null): void
+    public function listen(string $name, array|string $bindings = [], ?Flags $flags = null): void
     {
-        $this->consume(new Queue($name, $name, (array) $bindings, $flags ?? Flags::default()));
+        $this->consume(new Queue($name, (array) $bindings, $flags ?? Flags::default()));
     }
 
     public function consume(Queue $queue): void
@@ -95,10 +95,12 @@ private function channel(Dispatcher $dispatcher, array $options): AMQPChannel
         $handler = function (AMQPMessage $message) use ($channel, $dispatcher) {
             try {
                 $this->handle($message, $channel, $dispatcher);
+
+                $message->ack();
             } catch (Throwable $error) {
                 $this->logger->error((string) $error);
-            } finally {
-                $message->ack();
+
+                $message->nack();
             }
         };
 
@@ -155,8 +157,6 @@ private function handle(AMQPMessage $message, AMQPChannel $channel, Dispatcher $
             return;
         }
 
-        $envelope = $this->packager->unpack($message);
-
-        $dispatcher->dispatch($envelope);
+        $dispatcher->dispatch($this->packager->unpack($message));
     }
 }

src/Remote/AMQP/Flags.php

@@ -4,7 +4,7 @@
 
 namespace Onliner\CommandBus\Remote\AMQP;
 
-class Flags
+final class Flags
 {
     public const
         PASSIVE = 1,

src/Remote/AMQP/Headers.php

@@ -4,7 +4,7 @@
 
 namespace Onliner\CommandBus\Remote\AMQP;
 
-class Headers
+final class Headers
 {
     public const
         DELAY = 'x-delay',

src/Remote/AMQP/Packager.php

@@ -10,7 +10,7 @@
 use PhpAmqpLib\Wire\AMQPAbstractCollection;
 use PhpAmqpLib\Wire\AMQPTable;
 
-class Packager
+final class Packager
 {
     /**
      * @deprecated

src/Remote/AMQP/Publish.php

@@ -8,15 +8,15 @@ final class Publish
 {
     public function __construct(
         public string $exchange,
-        public string $queue,
+        public string $route,
         public string $payload,
     ) {}
 
     /**
      * @param array<string, mixed> $payload
      */
-    public static function create(string $exchange, string $queue, array $payload): self
+    public static function create(string $exchange, string $route, array $payload): self
     {
-        return new self($exchange, $queue, json_encode($payload, JSON_THROW_ON_ERROR));
+        return new self($exchange, $route, json_encode($payload, JSON_THROW_ON_ERROR));
     }
 }

src/Remote/AMQP/Queue.php

@@ -23,7 +23,6 @@ final class Queue
      */
     public function __construct(
         public string $name,
-        public string $pattern,
         private array $bindings,
         public Flags $flags,
         public array $args = [],
@@ -35,7 +34,7 @@ public function __construct(
     public static function create(array $options): self
     {
         $pattern = $options['pattern'] ?? '#';
-        $name = $options['queue'] ?? $pattern;
+        $name = $options['name'] ?? $pattern;
         $bindings = $options['bindings'] ?? [];
         $args = $options['args'] ?? [];
 
@@ -59,7 +58,11 @@ public static function create(array $options): self
             throw new InvalidArgumentException('Queue arguments must be an array');
         }
 
-        return new self($name, $pattern, $bindings, Flags::compute($options), $args);
+        if (array_is_list($bindings)) {
+            $bindings = array_fill_keys($bindings, $pattern);
+        }
+
+        return new self($name, $bindings, Flags::compute($options), $args);
     }
 
     public function is(int $flag): bool
@@ -76,11 +79,11 @@ public function consume(AMQPChannel $channel, callable $handler): void
             $this->is(Flags::EXCLUSIVE),
             $this->is(Flags::DELETE),
             $this->is(Flags::NO_WAIT),
-            new AMQPTable($this->args)
+            new AMQPTable($this->args),
         );
 
-        foreach ($this->bindings as $binding) {
-            $channel->queue_bind($this->name, $binding, $this->pattern);
+        foreach ($this->bindings as $exchange => $pattern) {
+            $channel->queue_bind($this->name, $exchange, $pattern);
         }
 
         $channel->basic_consume(

src/Remote/AMQP/Transport.php

@@ -53,15 +53,8 @@ public function consume(): Consumer
         return new Consumer($this->connector, $this->packager, $this->logger);
     }
 
-    /**
-     * @param Exchange|array<string, mixed> $exchange
-     */
-    public function declare(Exchange|array $exchange): void
+    public function declare(Exchange $exchange): void
     {
-        if (is_array($exchange)) {
-            $exchange = Exchange::create($exchange);
-        }
-
         $exchange->declare($this->connector->connect());
     }
 
@@ -70,7 +63,7 @@ public function setup(Builder $builder): void
         $builder->handle(Publish::class, function (Publish $message, Context $context) {
             $this->publish(
                 new Envelope(Publish::class, $message->payload, $context->all()),
-                new Route($message->exchange, $message->queue),
+                new Route($message->exchange, $message->route),
             );
         });
     }

src/Remote/RemoteExtension.php

@@ -19,7 +19,7 @@ final class RemoteExtension implements Extension
         Envelope::class,
     ];
 
-    public function __construct(Transport $transport = null, Serializer $serializer = null)
+    public function __construct(?Transport $transport = null, ?Serializer $serializer = null)
     {
         $this->transport = $transport ?? new Transport\MemoryTransport();
         $this->serializer = $serializer ?? new Serializer\NativeSerializer();

src/Remote/Transport/MultiTransport.php

@@ -8,7 +8,7 @@
 use Onliner\CommandBus\Remote\Envelope;
 use Onliner\CommandBus\Remote\Transport;
 
-class MultiTransport implements Transport
+final class MultiTransport implements Transport
 {
     /**
      * @var array<string, Transport>

src/Retry/RetryExtension.php

@@ -16,7 +16,7 @@ final class RetryExtension implements Extension
      */
     private array $policies = [];
 
-    public function __construct(Policy $default = null)
+    public function __construct(?Policy $default = null)
     {
         $this->default = $default ?? new Policy\ThrowPolicy();
     }