Merge branch '4.3'

* 4.3:
  [Messenger] Disable the SchemaAssetsFilter when setup the transport
  [Messenger] Fix missing auto_setup for RedisTransport
  [Contracts] split in one package per sub-contracts

Author: fabpot

Tests/Transport/Doctrine/ConnectionTest.php

@@ -108,7 +108,10 @@ private function getDBALConnectionMock()
         $platform = $this->getMockBuilder(AbstractPlatform::class)
             ->getMock();
         $platform->method('getWriteLockSQL')->willReturn('FOR UPDATE');
+        $configuration = $this->getMockBuilder(\Doctrine\DBAL\Configuration::class)
+            ->getMock();
         $driverConnection->method('getDatabasePlatform')->willReturn($platform);
+        $driverConnection->method('getConfiguration')->willReturn($configuration);
 
         return $driverConnection;
     }

Tests/Transport/RedisExt/ConnectionTest.php

@@ -42,13 +42,13 @@ public function testFromDsn()
     public function testFromDsnWithOptions()
     {
         $this->assertEquals(
-            new Connection(['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1'], [
+            new Connection(['stream' => 'queue', 'group' => 'group1', 'consumer' => 'consumer1', 'auto_setup' => false], [
                 'host' => 'localhost',
                 'port' => 6379,
             ], [
                 'serializer' => 2,
             ]),
-            Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2])
+            Connection::fromDsn('redis://localhost/queue/group1/consumer1', ['serializer' => 2, 'auto_setup' => false])
         );
     }
 
@@ -117,10 +117,6 @@ public function testGetAfterReject()
     {
         $redis = new \Redis();
         $connection = Connection::fromDsn('redis://localhost/messenger-rejectthenget', [], $redis);
-        try {
-            $connection->setup();
-        } catch (TransportException $e) {
-        }
 
         $connection->add('1', []);
         $connection->add('2', []);
@@ -139,10 +135,6 @@ public function testGetNonBlocking()
         $redis = new \Redis();
 
         $connection = Connection::fromDsn('redis://localhost/messenger-getnonblocking', [], $redis);
-        try {
-            $connection->setup();
-        } catch (TransportException $e) {
-        }
 
         $this->assertNull($connection->get()); // no message, should return null immediately
         $connection->add('1', []);

Transport/Doctrine/Connection.php

@@ -197,7 +197,25 @@ public function reject(string $id): bool
 
     public function setup(): void
     {
+        $configuration = $this->driverConnection->getConfiguration();
+        // Since Doctrine 2.9 the getFilterSchemaAssetsExpression is deprecated
+        $hasFilterCallback = method_exists($configuration, 'getSchemaAssetsFilter');
+
+        if ($hasFilterCallback) {
+            $assetFilter = $this->driverConnection->getConfiguration()->getSchemaAssetsFilter();
+            $this->driverConnection->getConfiguration()->setSchemaAssetsFilter(null);
+        } else {
+            $assetFilter = $this->driverConnection->getConfiguration()->getFilterSchemaAssetsExpression();
+            $this->driverConnection->getConfiguration()->setFilterSchemaAssetsExpression(null);
+        }
+
         $this->schemaSynchronizer->updateSchema($this->getSchema(), true);
+
+        if ($hasFilterCallback) {
+            $this->driverConnection->getConfiguration()->setSchemaAssetsFilter($assetFilter);
+        } else {
+            $this->driverConnection->getConfiguration()->setFilterSchemaAssetsExpression($assetFilter);
+        }
     }
 
     public function getMessageCount(): int

Transport/RedisExt/Connection.php

@@ -27,20 +27,29 @@
  */
 class Connection
 {
+    private const DEFAULT_OPTIONS = [
+        'stream' => 'messages',
+        'group' => 'symfony',
+        'consumer' => 'consumer',
+        'auto_setup' => true,
+    ];
+
     private $connection;
     private $stream;
     private $group;
     private $consumer;
+    private $autoSetup;
     private $couldHavePendingMessages = true;
 
     public function __construct(array $configuration, array $connectionCredentials = [], array $redisOptions = [], \Redis $redis = null)
     {
         $this->connection = $redis ?: new \Redis();
         $this->connection->connect($connectionCredentials['host'] ?? '127.0.0.1', $connectionCredentials['port'] ?? 6379);
         $this->connection->setOption(\Redis::OPT_SERIALIZER, $redisOptions['serializer'] ?? \Redis::SERIALIZER_PHP);
-        $this->stream = $configuration['stream'] ?? '' ?: 'messages';
-        $this->group = $configuration['group'] ?? '' ?: 'symfony';
-        $this->consumer = $configuration['consumer'] ?? '' ?: 'consumer';
+        $this->stream = $configuration['stream'] ?? self::DEFAULT_OPTIONS['stream'];
+        $this->group = $configuration['group'] ?? self::DEFAULT_OPTIONS['group'];
+        $this->consumer = $configuration['consumer'] ?? self::DEFAULT_OPTIONS['consumer'];
+        $this->autoSetup = $configuration['auto_setup'] ?? self::DEFAULT_OPTIONS['auto_setup'];
     }
 
     public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $redis = null): self
@@ -51,9 +60,9 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
 
         $pathParts = explode('/', $parsedUrl['path'] ?? '');
 
-        $stream = $pathParts[1] ?? '';
-        $group = $pathParts[2] ?? '';
-        $consumer = $pathParts[3] ?? '';
+        $stream = $pathParts[1] ?? null;
+        $group = $pathParts[2] ?? null;
+        $consumer = $pathParts[3] ?? null;
 
         $connectionCredentials = [
             'host' => $parsedUrl['host'] ?? '127.0.0.1',
@@ -64,11 +73,21 @@ public static function fromDsn(string $dsn, array $redisOptions = [], \Redis $re
             parse_str($parsedUrl['query'], $redisOptions);
         }
 
-        return new self(['stream' => $stream, 'group' => $group, 'consumer' => $consumer], $connectionCredentials, $redisOptions, $redis);
+        $autoSetup = null;
+        if (\array_key_exists('auto_setup', $redisOptions)) {
+            $autoSetup = filter_var($redisOptions['auto_setup'], FILTER_VALIDATE_BOOLEAN);
+            unset($redisOptions['auto_setup']);
+        }
+
+        return new self(['stream' => $stream, 'group' => $group, 'consumer' => $consumer, 'auto_setup' => $autoSetup], $connectionCredentials, $redisOptions, $redis);
     }
 
     public function get(): ?array
     {
+        if ($this->autoSetup) {
+            $this->setup();
+        }
+
         $messageId = '>'; // will receive new messages
 
         if ($this->couldHavePendingMessages) {
@@ -141,6 +160,10 @@ public function reject(string $id): void
 
     public function add(string $body, array $headers): void
     {
+        if ($this->autoSetup) {
+            $this->setup();
+        }
+
         $e = null;
         try {
             $added = $this->connection->xadd($this->stream, '*', ['message' => json_encode(
@@ -161,5 +184,7 @@ public function setup(): void
         } catch (\RedisException $e) {
             throw new TransportException($e->getMessage(), 0, $e);
         }
+
+        $this->autoSetup = false;
     }
 }