WIP

examples/multi/consumer_bar.php

@@ -21,4 +21,4 @@
 /** @var AMQPConsumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->run([$dispatcher, 'dispatch']);

examples/multi/consumer_foo.php

@@ -21,4 +21,4 @@
 /** @var AMQPConsumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->run([$dispatcher, 'dispatch']);

examples/remote/consumer.php

@@ -38,7 +38,7 @@
     ]));
 }
 
-$consumer->run($dispatcher, [
+$consumer->run([$dispatcher, 'dispatch'], [
     AMQPConsumer::OPTION_ATTEMPTS => 10,
     AMQPConsumer::OPTION_INTERVAL => 100000, // 100 ms
 ]);

examples/router/consumer_bar.php

@@ -21,4 +21,4 @@
 /** @var AMQPConsumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->run([$dispatcher, 'dispatch']);

examples/router/consumer_foo.php

@@ -21,4 +21,4 @@
 /** @var AMQPConsumer $consumer */
 $consumer = $transport->consume();
 $consumer->listen('#');
-$consumer->run($dispatcher);
+$consumer->run([$dispatcher, 'dispatch']);

src/Context.php

@@ -8,7 +8,7 @@
 
 final class Context
 {
-    private const OPTION_LOCAL = 'local';
+    public const OPTION_LOCAL = 'local';
 
     /**
      * @param Dispatcher           $dispatcher

src/Dispatcher.php

@@ -33,4 +33,17 @@ public function dispatch(object $message, array $options = []): void
             $this->dispatch($deferred->message, $deferred->options);
         }
     }
+
+    /**
+     * @param object               $message
+     * @param array<string, mixed> $options
+     *
+     * @return void
+     */
+    public function execute(object $message, array $options = []): void
+    {
+        $this->dispatch($message, array_replace($options, [
+            Context::OPTION_LOCAL => true,
+        ]));
+    }
 }

src/Remote/AMQP/AMQPConsumer.php

@@ -4,7 +4,6 @@
 
 namespace Onliner\CommandBus\Remote\AMQP;
 
-use Onliner\CommandBus\Dispatcher;
 use Onliner\CommandBus\Remote\Consumer;
 use Onliner\CommandBus\Remote\Envelope;
 use PhpAmqpLib\Channel\AMQPChannel;
@@ -51,13 +50,14 @@ public function __construct(
     }
 
     /**
-     * @param string $pattern
+     * @param string      $name
+     * @param string|null $pattern
      *
      * @return void
      */
-    public function listen(string $pattern): void
+    public function listen(string $name, string $pattern = null): void
     {
-        $this->consume(new Queue($pattern, $pattern, $this->exchange->flags()));
+        $this->consume(new Queue($name, $pattern, $this->exchange->flags()));
     }
 
     /**
@@ -71,11 +71,11 @@ public function consume(Queue $queue): void
     /**
      * {@inheritDoc}
      */
-    public function run(Dispatcher $dispatcher, array $options = []): void
+    public function run(callable $handler, array $options = []): void
     {
         $this->running = true;
 
-        $channel = $this->channel($dispatcher, $options);
+        $channel = $this->channel($handler, $options);
 
         /** @phpstan-ignore-next-line */
         while ($this->running && $channel->is_consuming()) {
@@ -87,7 +87,7 @@ public function run(Dispatcher $dispatcher, array $options = []): void
                     throw $error;
                 }
 
-                $channel = $this->channel($dispatcher, $options);
+                $channel = $this->channel($handler, $options);
             } catch (Throwable $error) {
                 $this->logger->error((string) $error);
             }
@@ -105,29 +105,28 @@ public function stop(): void
     }
 
     /**
-     * @param Dispatcher           $dispatcher
+     * @param callable             $handler
      * @param array<string, mixed> $options
      *
      * @return AMQPChannel
      * @throws AMQPIOException
      */
-    private function channel(Dispatcher $dispatcher, array $options): AMQPChannel
+    private function channel(callable $handler, array $options): AMQPChannel
     {
         $channel = $this->connect($options);
-        $handler = function (AMQPMessage $message) use ($dispatcher) {
-            try {
-                $this->handle($message, $dispatcher);
-            } catch (Throwable $error) {
-                $this->logger->error((string) $error);
-            } finally {
-                $message->ack();
-            }
-        };
 
         $this->exchange->declare($channel);
 
         foreach ($this->queues as $queue) {
-            $queue->consume($channel, $this->exchange, $handler);
+            $queue->consume($channel, $this->exchange, function (AMQPMessage $message) use ($handler) {
+                try {
+                    $this->handle($message, $handler);
+                } catch (Throwable $error) {
+                    $this->logger->error((string) $error);
+                } finally {
+                    $message->ack();
+                }
+            });
         }
 
         return $channel;
@@ -172,11 +171,11 @@ private function connect(array $options): AMQPChannel
 
     /**
      * @param AMQPMessage $message
-     * @param Dispatcher  $dispatcher
+     * @param callable    $handler
      *
      * @return void
      */
-    private function handle(AMQPMessage $message, Dispatcher $dispatcher): void
+    private function handle(AMQPMessage $message, callable $handler): void
     {
         $headers = $message->get('application_headers');
 
@@ -203,6 +202,6 @@ private function handle(AMQPMessage $message, Dispatcher $dispatcher): void
         /** @var class-string $class */
         $class = $headers[Exchange::HEADER_MESSAGE_TYPE];
 
-        $dispatcher->dispatch(new Envelope($class, $message->getBody(), $headers));
+        $handler(new Envelope($class, $message->getBody(), $headers));
     }
 }

src/Remote/Consumer.php

@@ -4,17 +4,15 @@
 
 namespace Onliner\CommandBus\Remote;
 
-use Onliner\CommandBus\Dispatcher;
-
 interface Consumer
 {
     /**
-     * @param Dispatcher           $dispatcher
+     * @param callable             $handler
      * @param array<string, mixed> $options
      *
      * @return void
      */
-    public function run(Dispatcher $dispatcher, array $options = []): void;
+    public function run(callable $handler, array $options = []): void;
 
     /**
      * @return void

src/Remote/Transport/MemoryTransport.php

@@ -4,7 +4,6 @@
 
 namespace Onliner\CommandBus\Remote\Transport;
 
-use Onliner\CommandBus\Dispatcher;
 use Onliner\CommandBus\Remote\Consumer;
 use Onliner\CommandBus\Remote\Envelope;
 use Onliner\CommandBus\Remote\Transport;
@@ -41,7 +40,7 @@ public function consume(): Consumer
     /**
      * {@inheritDoc}
      */
-    public function run(Dispatcher $dispatcher, array $options = []): void
+    public function run(callable $handler, array $options = []): void
     {
         if ($this->running) {
             return;
@@ -53,8 +52,8 @@ public function run(Dispatcher $dispatcher, array $options = []): void
             foreach ($this->envelopes as $type => $envelopes) {
                 foreach ($envelopes as $i => $envelope) {
                     try {
-                        $dispatcher->dispatch($envelope);
-                    } catch (Throwable $error) {
+                        $handler($envelope);
+                    } catch (Throwable) {
                         unset($this->envelopes[$type][$i]);
                     }
                 }
@@ -87,13 +86,15 @@ public function empty(): bool
     }
 
     /**
-     * @param string $type
+     * @param string|null $type
      *
      * @return array<Envelope>
      */
-    public function receive(string $type): array
+    public function receive(string $type = null): array
     {
-        return $this->envelopes[$type] ?? [];
+        return $type !== null
+            ? $this->envelopes[$type] ?? []
+            : $this->envelopes;
     }
 
     /**