symfony · glengemann · Sep 16, 2022 · Sep 26, 2022 · Oct 21, 2022
diff --git a/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php b/src/Symfony/Bundle/FrameworkBundle/DependencyInjection/FrameworkExtension.php
@@ -2016,6 +2016,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
                 ['id' => 'failed_message_processing_middleware'],
             ],
             'after' => [
+                ['id' => 'compress_middleware'],
                 ['id' => 'send_message'],
                 ['id' => 'handle_message'],
             ],

diff --git a/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php b/src/Symfony/Bundle/FrameworkBundle/Resources/config/messenger.php
@@ -25,6 +25,7 @@
 use Symfony\Component\Messenger\EventListener\StopWorkerOnRestartSignalListener;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnSigtermSignalListener;
 use Symfony\Component\Messenger\Middleware\AddBusNameStampMiddleware;
+use Symfony\Component\Messenger\Middleware\CompressMiddleware;
 use Symfony\Component\Messenger\Middleware\DispatchAfterCurrentBusMiddleware;
 use Symfony\Component\Messenger\Middleware\FailedMessageProcessingMiddleware;
 use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;
@@ -110,6 +111,9 @@
                 service('router'),
             ])
 
+        ->set('messenger.middleware.compress_middleware', CompressMiddleware::class)
+            ->abstract()
+
         // Discovery
         ->set('messenger.receiver_locator', ServiceLocator::class)
             ->args([

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/CompressorFactory.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/CompressorFactory.php
@@ -0,0 +1,26 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Amqp\Compressor;
+
+use Symfony\Component\Messenger\Exception\InvalidArgumentException;
+
+class CompressorFactory
+{
+    public static function createCompressor(string $mimeContentEncoding)
+    {
+        return match ($mimeContentEncoding) {
+            Gzip::CONTENT_ENCODING => new Gzip(),
+            Deflate::CONTENT_ENCODING => new Deflate(),
+            default => throw new InvalidArgumentException(sprintf('The MIME content encoding of the message cannot be decompressed "%s".', $mimeContentEncoding)),
+        };
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/CompressorInterface.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/CompressorInterface.php
@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Amqp\Compressor;
+
+interface CompressorInterface
+{
+    public function compress(mixed $data): string;
+
+    public function decompress(mixed $data): mixed;
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/Deflate.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/Deflate.php
@@ -0,0 +1,31 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Amqp\Compressor;
+
+class Deflate implements CompressorInterface
+{
+    public const CONTENT_ENCODING = 'deflate';
+
+    public function compress(mixed $data): string
+    {
+        return gzdeflate($data);
+    }
+
+    public function decompress(mixed $data): mixed
+    {
+        if (\function_exists('gzinflate')) {
+            return @gzinflate($data) ?: $data;
+        }
+
+        return $data;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/Gzip.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Compressor/Gzip.php
@@ -0,0 +1,31 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Amqp\Compressor;
+
+class Gzip implements CompressorInterface
+{
+    public const CONTENT_ENCODING = 'gzip';
+
+    public function compress(mixed $data): string
+    {
+        return gzencode($data);
+    }
+
+    public function decompress(mixed $data): mixed
+    {
+        if (\function_exists('gzdecode')) {
+            return @gzdecode($data) ?: $data;
+        }
+
+        return $data;
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Compressor/DeflateTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Compressor/DeflateTest.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace Compressor;
+
+use Symfony\Component\Messenger\Bridge\Amqp\Compressor\Deflate;
+use PHPUnit\Framework\TestCase;
+
+class DeflateTest extends TestCase
+{
+    public function testDecompress()
+    {
+        $compressor = new Deflate();
+
+        $actual = $compressor->decompress('string no compressed');
+        $this->assertEquals('string no compressed', $actual);
+
+        $actual = $compressor->decompress(gzdeflate('string compressed'));
+        $this->assertEquals('string compressed', $actual);
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Compressor/GzipTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Compressor/GzipTest.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace Compressor;
+
+use Symfony\Component\Messenger\Bridge\Amqp\Compressor\Gzip;
+use PHPUnit\Framework\TestCase;
+
+class GzipTest extends TestCase
+{
+    public function testDecompress()
+    {
+        $compressor = new Gzip();
+
+        $actual = $compressor->decompress('string no compressed');
+        $this->assertEquals('string no compressed', $actual);
+
+        $actual = $compressor->decompress(gzencode('string compressed'));
+        $this->assertEquals('string compressed', $actual);
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/AmqpReceiverTest.php
@@ -46,6 +46,40 @@ public function testItReturnsTheDecodedMessageToTheHandler()
         $this->assertEquals(new DummyMessage('Hi'), $actualEnvelopes[0]->getMessage());
     }
 
+    public function testItReturnsTheGzipDecompressMessageToTheHandler()
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $amqpEnvelope = $this->createAMQPEnvelopeWithContentEncodingGzip();
+        $connection = $this->createMock(Connection::class);
+        $connection->method('getQueueNames')->willReturn(['queueName']);
+        $connection->method('get')->with('queueName')->willReturn($amqpEnvelope);
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $actualEnvelopes = iterator_to_array($receiver->get());
+        $this->assertCount(1, $actualEnvelopes);
+        $this->assertEquals(new DummyMessage('Hi'), $actualEnvelopes[0]->getMessage());
+    }
+
+    public function testItReturnsTheDeflateDecompressMessageToTheHandler()
+    {
+        $serializer = new Serializer(
+            new SerializerComponent\Serializer([new ObjectNormalizer()], ['json' => new JsonEncoder()])
+        );
+
+        $amqpEnvelope = $this->createAMQPEnvelopeWithContentEncodingGzip();
+        $connection = $this->createMock(Connection::class);
+        $connection->method('getQueueNames')->willReturn(['queueName']);
+        $connection->method('get')->with('queueName')->willReturn($amqpEnvelope);
+
+        $receiver = new AmqpReceiver($connection, $serializer);
+        $actualEnvelopes = iterator_to_array($receiver->get());
+        $this->assertCount(1, $actualEnvelopes);
+        $this->assertEquals(new DummyMessage('Hi'), $actualEnvelopes[0]->getMessage());
+    }
+
     public function testItThrowsATransportExceptionIfItCannotAcknowledgeMessage()
     {
         $this->expectException(TransportException::class);
@@ -84,4 +118,28 @@ private function createAMQPEnvelope(): \AMQPEnvelope
 
         return $envelope;
     }
+
+    private function createAMQPEnvelopeWithContentEncodingGzip(): \AMQPEnvelope
+    {
+        $envelope = $this->createMock(\AMQPEnvelope::class);
+        $envelope->method('getBody')->willReturn(gzencode('{"message": "Hi"}'));
+        $envelope->method('getContentEncoding')->willReturn('gzip');
+        $envelope->method('getHeaders')->willReturn([
+            'type' => DummyMessage::class,
+        ]);
+
+        return $envelope;
+    }
+
+    private function createAMQPEnvelopeWithContentEncodingDeflate(): \AMQPEnvelope
+    {
+        $envelope = $this->createMock(\AMQPEnvelope::class);
+        $envelope->method('getBody')->willReturn(gzdeflate('{"message": "Hi"}'));
+        $envelope->method('getContentEncoding')->willReturn('gzip');
+        $envelope->method('getHeaders')->willReturn([
+            'type' => DummyMessage::class,
+        ]);
+
+        return $envelope;
+    }
 }
diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\Messenger\Bridge\Amqp\Transport;
 
+use Symfony\Component\Messenger\Bridge\Amqp\Compressor\CompressorFactory;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\LogicException;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
@@ -63,6 +64,14 @@ private function getEnvelope(string $queueName): iterable
         $body = $amqpEnvelope->getBody();
 
         try {
+            /*
+            $contentEncoding = $amqpEnvelope->getContentEncoding();
+            if ($contentEncoding) {
+                $compressor = CompressorFactory::createCompressor($contentEncoding);
+                $body = $compressor->decompress($body);
+            }
+            */
+
             $envelope = $this->serializer->decode([
                 'body' => false === $body ? '' : $body, // workaround https://github.com/pdezwart/php-amqp/issues/351
                 'headers' => $amqpEnvelope->getHeaders(),

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
@@ -11,6 +11,7 @@
 
 namespace Symfony\Component\Messenger\Bridge\Amqp\Transport;
 
+use Symfony\Component\Messenger\Bridge\Amqp\Compressor\CompressorFactory;
 use Symfony\Component\Messenger\Exception\InvalidArgumentException;
 use Symfony\Component\Messenger\Exception\LogicException;
 
@@ -334,7 +335,12 @@ private function publishOnExchange(\AMQPExchange $exchange, string $body, string
         $attributes['timestamp'] ??= time();
 
         $this->lastActivityTime = time();
-
+        /*
+        if (isset($attributes['content_encoding'])) {
+            $compressor = CompressorFactory::createCompressor($attributes['content_encoding']);
+            $body = $compressor->compress($body);
+        }
+        */
         $exchange->publish(
             $body,
             $routingKey,

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/composer.json b/src/Symfony/Component/Messenger/Bridge/Amqp/composer.json
@@ -24,7 +24,8 @@
         "symfony/event-dispatcher": "^5.4|^6.0",
         "symfony/process": "^5.4|^6.0",
         "symfony/property-access": "^5.4|^6.0",
-        "symfony/serializer": "^5.4|^6.0"
+        "symfony/serializer": "^5.4|^6.0",
+        "ext-zlib": "*"
     },
     "autoload": {
         "psr-4": { "Symfony\\Component\\Messenger\\Bridge\\Amqp\\": "" },

diff --git a/src/Symfony/Component/Messenger/Middleware/CompressMiddleware.php b/src/Symfony/Component/Messenger/Middleware/CompressMiddleware.php
@@ -0,0 +1,61 @@
+<?php
+
+namespace Symfony\Component\Messenger\Middleware;
+
+use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpReceivedStamp;
+use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpStamp;
+use Symfony\Component\Messenger\Envelope;
+
+class CompressMiddleware implements MiddlewareInterface
+{
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        if ($amqpReceivedStamp = $envelope->last(AmqpReceivedStamp::class)) {
+            $contentEncoding ??= $amqpReceivedStamp->getAmqpEnvelope()->getContentEncoding();
+            if (!$contentEncoding) {
+                return $stack->next()->handle($envelope, $stack);
+            }
+
+            $message = $envelope->getMessage();
+            $compress = $this->decompress($contentEncoding, $message);
+            $envelope = new Envelope(
+                $compress,
+                $envelope->all()
+            );
+        } else {
+            $amqpStamp = $envelope->last(AmqpStamp::class);
+            $contentEncoding = $amqpStamp->getAttributes()['content_encoding'] ?? null;
+            if (!$contentEncoding) {
+                return $stack->next()->handle($envelope, $stack);
+            }
+            $message = $envelope->getMessage();
+            // We need a string, but in this point we have an object
+            $compress = $this->compress($contentEncoding, $message);
+            $envelope = new Envelope(
+                $compress,
+                $envelope->all()
+            );
+        }
+
+        return $stack->next()->handle($envelope, $stack);
+    }
+
+    public function compress(string $contentEncoding, mixed $data): string
+    {
+        return match ($contentEncoding) {
+            'gzip' => gzencode($data),
+            'deflate' => gzdeflate($data),
+            default => throw new InvalidArgumentException(sprintf('The MIME content encoding of the message cannot be decompressed "%s".', $contentEncoding)),
+        };
+    }
+
+    public function decompress(string $contentEncoding, mixed $data): mixed
+    {
+        return match ($contentEncoding) {
+            'gzip' => gzdecode($data) ?: $data,
+            'deflate' => gzinflate($data) ?: $data,
+        };
+
+        return $data;
+    }
+}