| 706 | |
| 707 | |
| 708 | class _PerMessageDeflateCompressor(object): |
| 709 | def __init__( |
| 710 | self, |
| 711 | persistent: bool, |
| 712 | max_wbits: Optional[int], |
| 713 | compression_options: Optional[Dict[str, Any]] = None, |
| 714 | ) -> None: |
| 715 | if max_wbits is None: |
| 716 | max_wbits = zlib.MAX_WBITS |
| 717 | # There is no symbolic constant for the minimum wbits value. |
| 718 | if not (8 <= max_wbits <= zlib.MAX_WBITS): |
| 719 | raise ValueError( |
| 720 | "Invalid max_wbits value %r; allowed range 8-%d", |
| 721 | max_wbits, |
| 722 | zlib.MAX_WBITS, |
| 723 | ) |
| 724 | self._max_wbits = max_wbits |
| 725 | |
| 726 | if ( |
| 727 | compression_options is None |
| 728 | or "compression_level" not in compression_options |
| 729 | ): |
| 730 | self._compression_level = tornado.web.GZipContentEncoding.GZIP_LEVEL |
| 731 | else: |
| 732 | self._compression_level = compression_options["compression_level"] |
| 733 | |
| 734 | if compression_options is None or "mem_level" not in compression_options: |
| 735 | self._mem_level = 8 |
| 736 | else: |
| 737 | self._mem_level = compression_options["mem_level"] |
| 738 | |
| 739 | if persistent: |
| 740 | self._compressor = self._create_compressor() # type: Optional[_Compressor] |
| 741 | else: |
| 742 | self._compressor = None |
| 743 | |
| 744 | def _create_compressor(self) -> "_Compressor": |
| 745 | return zlib.compressobj( |
| 746 | self._compression_level, zlib.DEFLATED, -self._max_wbits, self._mem_level |
| 747 | ) |
| 748 | |
| 749 | def compress(self, data: bytes) -> bytes: |
| 750 | compressor = self._compressor or self._create_compressor() |
| 751 | data = compressor.compress(data) + compressor.flush(zlib.Z_SYNC_FLUSH) |
| 752 | assert data.endswith(b"\x00\x00\xff\xff") |
| 753 | return data[:-4] |
| 754 | |
| 755 | |
| 756 | class _PerMessageDeflateDecompressor(object): |