[PR #9780/eac8fb84 backport][3.11] Replace get_event_loop with get_running_loop in the compressor (#9789)

Co-authored-by: J. Nick Koston <[email protected]>

Author: patchback[bot]

aiohttp/compression_utils.py

@@ -70,6 +70,14 @@ def compress_sync(self, data: bytes) -> bytes:
         return self._compressor.compress(data)
 
     async def compress(self, data: bytes) -> bytes:
+        """Compress the data and returned the compressed bytes.
+
+        Note that flush() must be called after the last call to compress()
+
+        If the data size is large than the max_sync_chunk_size, the compression
+        will be done in the executor. Otherwise, the compression will be done
+        in the event loop.
+        """
         async with self._compress_lock:
             # To ensure the stream is consistent in the event
             # there are multiple writers, we need to lock
@@ -79,8 +87,8 @@ async def compress(self, data: bytes) -> bytes:
                 self._max_sync_chunk_size is not None
                 and len(data) > self._max_sync_chunk_size
             ):
-                return await asyncio.get_event_loop().run_in_executor(
-                    self._executor, self.compress_sync, data
+                return await asyncio.get_running_loop().run_in_executor(
+                    self._executor, self._compressor.compress, data
                 )
             return self.compress_sync(data)
 
@@ -107,12 +115,18 @@ def decompress_sync(self, data: bytes, max_length: int = 0) -> bytes:
         return self._decompressor.decompress(data, max_length)
 
     async def decompress(self, data: bytes, max_length: int = 0) -> bytes:
+        """Decompress the data and return the decompressed bytes.
+
+        If the data size is large than the max_sync_chunk_size, the decompression
+        will be done in the executor. Otherwise, the decompression will be done
+        in the event loop.
+        """
         if (
             self._max_sync_chunk_size is not None
             and len(data) > self._max_sync_chunk_size
         ):
-            return await asyncio.get_event_loop().run_in_executor(
-                self._executor, self.decompress_sync, data, max_length
+            return await asyncio.get_running_loop().run_in_executor(
+                self._executor, self._decompressor.decompress, data, max_length
             )
         return self.decompress_sync(data, max_length)
 

tests/test_compression_utils.py

@@ -0,0 +1,23 @@
+"""Tests for compression utils."""
+
+from aiohttp.compression_utils import ZLibCompressor, ZLibDecompressor
+
+
+async def test_compression_round_trip_in_executor() -> None:
+    """Ensure that compression and decompression work correctly in the executor."""
+    compressor = ZLibCompressor(max_sync_chunk_size=1)
+    decompressor = ZLibDecompressor(max_sync_chunk_size=1)
+    data = b"Hi" * 100
+    compressed_data = await compressor.compress(data) + compressor.flush()
+    decompressed_data = await decompressor.decompress(compressed_data)
+    assert data == decompressed_data
+
+
+async def test_compression_round_trip_in_event_loop() -> None:
+    """Ensure that compression and decompression work correctly in the event loop."""
+    compressor = ZLibCompressor(max_sync_chunk_size=10000)
+    decompressor = ZLibDecompressor(max_sync_chunk_size=10000)
+    data = b"Hi" * 100
+    compressed_data = await compressor.compress(data) + compressor.flush()
+    decompressed_data = await decompressor.decompress(compressed_data)
+    assert data == decompressed_data