Added publish handler for raw messages

Author: zloyuser

src/Remote/AMQP/Publish.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Onliner\CommandBus\Remote\AMQP;
+
+final class Publish
+{
+    public function __construct(
+        public string $exchange,
+        public string $queue,
+        public string $payload,
+    ) {}
+
+    /**
+     * @param array<string, mixed> $payload
+     */
+    public static function create(string $exchange, string $queue, array $payload): self
+    {
+        return new self($exchange, $queue, json_encode($payload, JSON_THROW_ON_ERROR));
+    }
+}

src/Remote/AMQP/Transport.php

@@ -4,11 +4,14 @@
 
 namespace Onliner\CommandBus\Remote\AMQP;
 
+use Onliner\CommandBus\Builder;
+use Onliner\CommandBus\Context;
+use Onliner\CommandBus\Extension;
 use Onliner\CommandBus\Remote\Envelope;
 use Onliner\CommandBus\Remote\Transport as TransportContract;
 use Psr\Log\LoggerInterface;
 
-final class Transport implements TransportContract
+final class Transport implements TransportContract, Extension
 {
     public function __construct(
         private Connector $connector,
@@ -28,9 +31,13 @@ public static function create(string $dsn, string $exchange = '', array $routes
     }
 
     public function send(Envelope $envelope): void
+    {
+        $this->publish($envelope, $this->router->match($envelope));
+    }
+
+    public function publish(Envelope $envelope, Route $route): void
     {
         $message = $this->packager->pack($envelope);
-        $route = $this->router->match($envelope);
 
         $channel = $this->connector->connect();
         $channel->basic_publish(
@@ -57,4 +64,14 @@ public function declare(Exchange|array $exchange): void
 
         $exchange->declare($this->connector->connect());
     }
+
+    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),
+            );
+        });
+    }
 }

src/Remote/RemoteExtension.php

@@ -32,6 +32,10 @@ public function local(string ...$local): void
 
     public function setup(Builder $builder): void
     {
+        if ($this->transport instanceof Extension) {
+            $this->transport->setup($builder);
+        }
+
         $gateway = new Gateway($this->transport, $this->serializer);
 
         $builder->middleware(new RemoteMiddleware($gateway, $this->local));