[Messenger] Add TransportInterface as first class citizen sender+receiver

Author: nicolas-grekas

Transport/AmqpExt/AmqpSender.php

@@ -33,7 +33,7 @@ public function __construct(EncoderInterface $messageEncoder, Connection $connec
     /**
      * {@inheritdoc}
      */
-    public function send($message)
+    public function send($message): void
     {
         $encodedMessage = $this->messageEncoder->encode($message);
 

Transport/AmqpExt/AmqpTransport.php

@@ -0,0 +1,79 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Transport\AmqpExt;
+
+use Symfony\Component\Messenger\Transport\Serialization\DecoderInterface;
+use Symfony\Component\Messenger\Transport\Serialization\EncoderInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+/**
+ * @author Nicolas Grekas <[email protected]>
+ */
+class AmqpTransport implements TransportInterface
+{
+    private $encoder;
+    private $decoder;
+    private $dsn;
+    private $options;
+    private $debug;
+    private $connection;
+    private $receiver;
+    private $sender;
+
+    public function __construct(EncoderInterface $encoder, DecoderInterface $decoder, string $dsn, array $options, bool $debug)
+    {
+        $this->encoder = $encoder;
+        $this->decoder = $decoder;
+        $this->dsn = $dsn;
+        $this->options = $options;
+        $this->debug = $debug;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function receive(callable $handler): void
+    {
+        ($this->receiver ?? $this->getReceiver())->receive($hander);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function stop(): void
+    {
+        ($this->receiver ?? $this->getReceiver())->stop();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function send($message): void
+    {
+        ($this->sender ?? $this->getSender())->send($message);
+    }
+
+    private function getReceiver()
+    {
+        return $this->receiver = new AmqpReceiver($this->decoder, $this->connection ?? $this->getConnection());
+    }
+
+    private function getSender()
+    {
+        return $this->sender = new AmqpSender($this->encoder, $this->connection ?? $this->getConnection());
+    }
+
+    private function getConnection()
+    {
+        return $this->connection = new Connection($this->dsn, $this->options, $this->debug);
+    }
+}

Transport/AmqpExt/AmqpTransportFactory.php

@@ -11,11 +11,10 @@
 
 namespace Symfony\Component\Messenger\Transport\AmqpExt;
 
-use Symfony\Component\Messenger\Transport\Factory\TransportFactoryInterface;
-use Symfony\Component\Messenger\Transport\ReceiverInterface;
-use Symfony\Component\Messenger\Transport\SenderInterface;
 use Symfony\Component\Messenger\Transport\Serialization\DecoderInterface;
 use Symfony\Component\Messenger\Transport\Serialization\EncoderInterface;
+use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
 
 /**
  * @author Samuel Roze <[email protected]>
@@ -33,14 +32,9 @@ public function __construct(EncoderInterface $encoder, DecoderInterface $decoder
         $this->debug = $debug;
     }
 
-    public function createReceiver(string $dsn, array $options): ReceiverInterface
+    public function createTransport(string $dsn, array $options): TransportInterface
     {
-        return new AmqpReceiver($this->decoder, Connection::fromDsn($dsn, $options, $this->debug));
-    }
-
-    public function createSender(string $dsn, array $options): SenderInterface
-    {
-        return new AmqpSender($this->encoder, Connection::fromDsn($dsn, $options, $this->debug));
+        return new AmqpTransport($this->encoder, $this->decoder, $dsn, $options, $thid->debug);
     }
 
     public function supports(string $dsn, array $options): bool

Transport/SenderInterface.php

@@ -23,5 +23,5 @@ interface SenderInterface
      *
      * @param object $message
      */
-    public function send($message);
+    public function send($message): void;
 }

Transport/Factory/ChainTransportFactory.php renamed to Transport/TransportFactory.php

@@ -9,15 +9,12 @@
  * file that was distributed with this source code.
  */
 
-namespace Symfony\Component\Messenger\Transport\Factory;
-
-use Symfony\Component\Messenger\Transport\ReceiverInterface;
-use Symfony\Component\Messenger\Transport\SenderInterface;
+namespace Symfony\Component\Messenger\Transport;
 
 /**
  * @author Samuel Roze <[email protected]>
  */
-class ChainTransportFactory implements TransportFactoryInterface
+class TransportFactory implements TransportFactoryInterface
 {
     private $factories;
 
@@ -29,22 +26,11 @@ public function __construct(iterable $factories)
         $this->factories = $factories;
     }
 
-    public function createReceiver(string $dsn, array $options): ReceiverInterface
-    {
-        foreach ($this->factories as $factory) {
-            if ($factory->supports($dsn, $options)) {
-                return $factory->createReceiver($dsn, $options);
-            }
-        }
-
-        throw new \InvalidArgumentException(sprintf('No transport supports the given DSN "%s".', $dsn));
-    }
-
-    public function createSender(string $dsn, array $options): SenderInterface
+    public function createTransport(string $dsn, array $options): TransportInterface
     {
         foreach ($this->factories as $factory) {
             if ($factory->supports($dsn, $options)) {
-                return $factory->createSender($dsn, $options);
+                return $factory->createTransport($dsn, $options);
             }
         }
 

Transport/Factory/TransportFactoryInterface.php renamed to Transport/TransportFactoryInterface.php

@@ -9,10 +9,7 @@
  * file that was distributed with this source code.
  */
 
-namespace Symfony\Component\Messenger\Transport\Factory;
-
-use Symfony\Component\Messenger\Transport\ReceiverInterface;
-use Symfony\Component\Messenger\Transport\SenderInterface;
+namespace Symfony\Component\Messenger\Transport;
 
 /**
  * Creates a Messenger transport.

Transport/TransportInterface.php

@@ -0,0 +1,21 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Transport;
+
+/**
+ * @author Nicolas Grekas <[email protected]>
+ *
+ * @experimental in 4.1
+ */
+interface TransportInterface extends ReceiverInterface, SenderInterface
+{
+}