Draft: implement compression/decompression using a middleware

symfony · glengemann · Sep 16, 2022 · Sep 26, 2022 · Oct 21, 2022 · Oct 21, 2022
commit ecaf750a5f4c49663eb0b21cca9e803b1ba8a06d
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/Transport/AmqpReceiver.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/AmqpReceiver.php
@@ -64,11 +64,13 @@ 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

diff --git a/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php b/src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
@@ -335,12 +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/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;
+    }
+}