Class BodyReceiver

gunicorn/asgi/protocol.py:162–332

Body receiver for callback-based parsers. Body chunks are fed directly via the feed() method from parser callbacks. Uses Future-based waiting for efficient async receive().

class BodyReceiver:
    """Body receiver for callback-based parsers.

    Body chunks are fed directly via the feed() method from parser callbacks.
    Uses Future-based waiting for efficient async receive().
    """

    __slots__ = ('_chunks', '_complete', '_body_finished', '_closed',
                 '_body_wait_expired', '_waiter', 'request', 'protocol')

    def __init__(self, request, protocol):
        self.request = request
        self.protocol = protocol
        self._chunks = []
        self._complete = False
        self._body_finished = False  # True after returning more_body=False
        # _closed means the client transport has gone away (signal_disconnect
        # was called or the protocol detected a disconnect).  _body_wait_expired
        # means the body did not finish framing within the configured timeout
        # but the transport itself may still be open.  Both surface as
        # http.disconnect to the app, but they are distinct conditions.
        self._closed = False
        self._body_wait_expired = False
        self._waiter = None

    def feed(self, chunk):
        """Feed a body chunk directly (called by parser callback)."""
        if chunk:
            self._chunks.append(chunk)
            self._wake_waiter()

    def set_complete(self):
        """Mark body as complete (called when message ends)."""
        self._complete = True
        self._wake_waiter()

    def signal_disconnect(self):
        """Signal that the client transport has gone away."""
        self._closed = True
        self._wake_waiter()

    @property
    def _disconnected(self):
        """True when the receiver should yield http.disconnect to the app."""
        return self._closed or self._body_wait_expired

    def _wake_waiter(self):
        """Wake up any pending receive() call."""
        if self._waiter is not None and not self._waiter.done():
            self._waiter.set_result(None)

    async def receive(self):  # pylint: disable=too-many-return-statements
        """ASGI receive callable - returns body chunks or disconnect."""
        # Already disconnected (transport closed or body wait timed out)
        if self._disconnected:
            return {"type": "http.disconnect"}

        # Body finished but not disconnected - wait for actual disconnect