how to handle early closing a stream on client side? (stream is never cancelled by the server but should support closing from client side)

[breathe]

I want to use httpx's streaming feature to interact with a service that provides a long-lived event stream over http2. Currently I'm only needing to support the http-core sync backend.

The general shape of the code is that the program initiates an httpx stream() and then a handler_method will be triggered for every decoded response from the stream. Within the handler_method, the client may decide to shutdown the stream early -- which leads to a call of the close() method on the httpx.Client object from which the stream() method was called.

There are two issues that I'm having currently having which I'm working around by copy/pasting some code from httpx into my own codebase and making some small edits ...

• problem behavior 1 -- the connection is closed after all the chunks are decoded/the byte decoder loop exits ... I have a little fog on what exactly the issues are here -- I believe this conflicts with the already triggered 'early close()' that the client performed and (I think) this could also result in affecting other http2 streams which happen to share the the same http connection --

  httpx/httpx/_models.py

  Line 889 in 2d37321

  self.close()

• problem behavior 2 -- when the stream is closed from the client side, errors which occur when attempting to recv or sendv the underlying socket will bubble up from the wrong part of the codebase which complicates handling of similar errors should they occur in non-closed code path. See comments in __httpx_iter_raw below for more on this

Here's the code I've copy/pasted and edited within my codebase -- it looks as similar as possible to upstream and my hope is that it provides readable context for understanding my issues ...

    # copy/pasted from here https://github.com/encode/httpx/blob/2d37321842c566b1fd81f9627ab4072064286d16/httpx/_models.py#L808
    # then edited.  The actual_self = self dance is to keep the code below as similar as possible to above
    # The reason this is here is to workaround two behaviors of the upstream implementation:
    # problem behavior 1 -- the connection is closed after all the chunks are decoded which can affect other streams on the connection -- https://github.com/encode/httpx/blob/2d37321842c566b1fd81f9627ab4072064286d16/httpx/_models.py#L889
    # problem behavior 2 -- when the stream is closed from the client side, errors which occur when attempting to recv or sendv the underlying socket will bubble up from the wrong part of the codebase which complictes handling of similar errors should they occur in non-closed code path.  See comments in __httpx_iter_raw for more
    def __httpx_iter_bytes(
            self,
            response,
            chunk_size: typing.Optional[int] = None) -> typing.Iterator[bytes]:
        """
        A byte-iterator over the decoded response content.
        This allows us to handle gzip, deflate, and brotli encoded responses.
        """
        actual_self = self
        self = response
        if hasattr(self, "_content"):
            chunk_size = len(
                self._content) if chunk_size is None else chunk_size
            for i in range(0, len(self._content), max(chunk_size, 1)):
                yield self._content[i:i + chunk_size]
        else:
            decoder = self._get_content_decoder()
            chunker = ByteChunker(chunk_size=chunk_size)
            with request_context(request=self._request):
                for raw_bytes in actual_self.__httpx_iter_raw(self):
                    decoded = decoder.decode(raw_bytes)
                    for chunk in chunker.decode(decoded):
                        yield chunk
                decoded = decoder.flush()
                for chunk in chunker.decode(decoded):
                    yield chunk  # pragma: no cover
                for chunk in chunker.flush():
                    yield chunk

    # copy/pasted from here https://github.com/encode/httpx/blob/2d37321842c566b1fd81f9627ab4072064286d16/httpx/_models.py#L863
    # then edited.  The actual_self = self dance is to keep the code below as similar as possible to above
    def __httpx_iter_raw(
            self,
            response,
            chunk_size: typing.Optional[int] = None) -> typing.Iterator[bytes]:
        """
        A byte-iterator over the raw response content.
        """
        actual_self = self
        self = response
        if self.is_stream_consumed:
            raise StreamConsumed()
        if self.is_closed:
            raise StreamClosed()
        if not isinstance(self.stream, SyncByteStream):
            raise RuntimeError(
                "Attempted to call a sync iterator on an async stream.")

        self.is_stream_consumed = True
        self._num_bytes_downloaded = 0
        chunker = ByteChunker(chunk_size=chunk_size)

        with request_context(request=self._request):
            try:
                for raw_stream_bytes in self.stream:
                    self._num_bytes_downloaded += len(raw_stream_bytes)
                    for chunk in chunker.decode(raw_stream_bytes):
                        yield chunk
            except (ReadError, WriteError) as e:
                # EDIT:
                # add special case error handling for ReadError and WriteError which occur after the application has tried to close the stream.
                # These errors result from the fact that the underlying connection infrastructure (eg os socket) is torn down _before_ the decoder loop is exited
                # ReadError: https://github.com/encode/httpcore/blob/a2f86caa0f230701466e1649c934a87c4dc293f8/httpcore/backends/sync.py#L28 --
                # WriteError: https://github.com/encode/httpcore/blob/7eb20224833b8165d0945d08ac8f714ccb6750b9/httpcore/backends/sync.py#L38 --
                if actual_self._state == "closed":
                    pass
                elif actual_self._state == "closing":
                    pass
                else:
                    raise e

        for chunk in chunker.flush():
            yield chunk
        # EDIT:
        # we don't want to close the connection after decoding a response from the stream ...
        # self.close()

I may not be understanding the library correctly and would definitely appreciate pointers if there's a better way get behavior like the above ... Or if not, it would be great if httpx could be modified to better support this use case!

Many thanks for any help!

[breathe]

I think I figured out how to do what I want -- and this appears to suffice for my current use cases without any of the hax like the above being needed ...

In my case, I could simply adjust the outermost loop consuming and transorming/processing the stream_response.iter_bytes() to simply break out of that loop when an early exit state has been identified. That can be done without anyone calling close() on any of the Client/Response objects. The stream and connection can then go through the normal shutdown logic when the context manager scope has ended and nothing will ever bubble up into my code ...