Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``.
| 706 | |
| 707 | |
| 708 | class _GzipMessageDelegate(httputil.HTTPMessageDelegate): |
| 709 | """Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``.""" |
| 710 | |
| 711 | def __init__(self, delegate: httputil.HTTPMessageDelegate, chunk_size: int) -> None: |
| 712 | self._delegate = delegate |
| 713 | self._chunk_size = chunk_size |
| 714 | self._decompressor = None # type: Optional[GzipDecompressor] |
| 715 | |
| 716 | def headers_received( |
| 717 | self, |
| 718 | start_line: Union[httputil.RequestStartLine, httputil.ResponseStartLine], |
| 719 | headers: httputil.HTTPHeaders, |
| 720 | ) -> Optional[Awaitable[None]]: |
| 721 | if headers.get("Content-Encoding", "").lower() == "gzip": |
| 722 | self._decompressor = GzipDecompressor() |
| 723 | # Downstream delegates will only see uncompressed data, |
| 724 | # so rename the content-encoding header. |
| 725 | # (but note that curl_httpclient doesn't do this). |
| 726 | headers.add("X-Consumed-Content-Encoding", headers["Content-Encoding"]) |
| 727 | del headers["Content-Encoding"] |
| 728 | return self._delegate.headers_received(start_line, headers) |
| 729 | |
| 730 | async def data_received(self, chunk: bytes) -> None: |
| 731 | if self._decompressor: |
| 732 | compressed_data = chunk |
| 733 | while compressed_data: |
| 734 | decompressed = self._decompressor.decompress( |
| 735 | compressed_data, self._chunk_size |
| 736 | ) |
| 737 | if decompressed: |
| 738 | ret = self._delegate.data_received(decompressed) |
| 739 | if ret is not None: |
| 740 | await ret |
| 741 | compressed_data = self._decompressor.unconsumed_tail |
| 742 | if compressed_data and not decompressed: |
| 743 | raise httputil.HTTPInputError( |
| 744 | "encountered unconsumed gzip data without making progress" |
| 745 | ) |
| 746 | else: |
| 747 | ret = self._delegate.data_received(chunk) |
| 748 | if ret is not None: |
| 749 | await ret |
| 750 | |
| 751 | def finish(self) -> None: |
| 752 | if self._decompressor is not None: |
| 753 | tail = self._decompressor.flush() |
| 754 | if tail: |
| 755 | # The tail should always be empty: decompress returned |
| 756 | # all that it can in data_received and the only |
| 757 | # purpose of the flush call is to detect errors such |
| 758 | # as truncated input. If we did legitimately get a new |
| 759 | # chunk at this point we'd need to change the |
| 760 | # interface to make finish() a coroutine. |
| 761 | raise ValueError( |
| 762 | "decompressor.flush returned data; possible truncated input" |
| 763 | ) |
| 764 | return self._delegate.finish() |
| 765 |