Function cancel_on_disconnect

tests/middleware/test_base.py:512–521

(
                *,
                task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
            )

        async with anyio.create_task_group() as task_group:

            async def cancel_on_disconnect(
                *,
                task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
            ) -> None:
                task_status.started()
                while True:
                    message = await receive()
                    if message["type"] == "http.disconnect":  # pragma: no branch
                        task_group.cancel_scope.cancel()
                        break

            # Using start instead of start_soon to ensure that
            # cancel_on_disconnect is scheduled by the event loop