Add Request is_disconnected (#456)

* Fix #452
* Make the ASGI receive accessible to Python code

CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.0.4] - 2023-12-31 :fireworks:
+
+- Adds a `is_disconnected()` method to the `Request` class, similar to the one
+  available in `Starlette`, which answers if the ASGI server published an
+  `http.disconnected` message for a request.
+  Feature requested by @netanel-haber in [#452](https://github.com/Neoteroi/BlackSheep/issues/452).
+- Makes the `receive` callable of the `ASGI` request accessible to Python code,
+  through the existing `ASGIContent` class. The `receive` property was already
+  included in `contents.pyi` file and it was wrong to keep `receive` private
+  for Cython code.
+- Removes `consts.pxi` because it used a deprecated Cython feature.
+- Upgrades the versions of Hypercorn and uvicorn for integration tests.
+
 ## [2.0.3] - 2023-12-18 :gift:
 
 - Fixes #450, about missing `Access-Control-Allow-Credentials` response header

blacksheep/__init__.py

@@ -3,7 +3,7 @@
 used types to reduce the verbosity of the imports statements.
 """
 __author__ = "Roberto Prevato <[email protected]>"
-__version__ = "2.0.3"
+__version__ = "2.0.4"
 
 from .contents import Content as Content
 from .contents import FormContent as FormContent

blacksheep/contents.pxd

@@ -16,7 +16,7 @@ cdef class StreamedContent(Content):
 
 
 cdef class ASGIContent(Content):
-    cdef object receive
+    cdef readonly object receive
     cpdef void dispose(self)
 
 

blacksheep/contents.pyi

@@ -1,5 +1,15 @@
 import uuid
-from typing import Any, AsyncIterable, Callable, Dict, List, Optional, Tuple, Union
+from typing import (
+    Any,
+    AsyncIterable,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
 
 class Content:
     def __init__(self, content_type: bytes, data: bytes):
@@ -24,7 +34,7 @@ class StreamedContent(Content):
     async def get_parts(self) -> AsyncIterable[bytes]: ...
 
 class ASGIContent(Content):
-    def __init__(self, receive: Callable[[], bytes]):
+    def __init__(self, receive: Callable[[], Awaitable[dict]]):
         self.type = None
         self.body = None
         self.length = -1

blacksheep/messages.pxd

@@ -34,6 +34,7 @@ cdef class Message:
 
     cdef void remove_headers(self, list headers)
     cdef list get_headers_tuples(self, bytes key)
+    cdef void init_prop(self, str name, object value)
 
     cpdef Message with_content(self, Content content)
     cpdef bint has_body(self)

blacksheep/messages.pyi

@@ -119,6 +119,20 @@ class Request(Message):
     def original_client_ip(self, value: str) -> None: ...
     @property
     def path(self) -> str: ...
+    async def is_disconnected(self) -> bool:
+        """
+        Returns a value indicating whether the web request is still bound to an active
+        connection. In case of long-polling, this method returns True if the client
+        closed the original connection. For requests originated from a web browser, this
+        method returns True also if the user refreshed a page that originated a web
+        request, or the connection got lost and a page initiated a new request.
+
+        Because this method relies on reading incoming ASGI messages, it can only be
+        used for incoming web requests handled through an ASGI server, and it must not
+        be used when reading the request stream, as it cannot be read more than once.
+        When reading the request stream, catch instead MessageAborted exceptions to
+        detect if the client closed the original connection.
+        """
 
 class Response(Message):
     def __init__(

blacksheep/messages.pyx

@@ -1,3 +1,4 @@
+import asyncio
 import http
 import re
 from datetime import datetime, timedelta
@@ -11,9 +12,14 @@ from blacksheep.sessions import Session
 from blacksheep.settings.json import json_settings
 from blacksheep.utils.time import utcnow
 
-from .contents cimport Content, multiparts_to_dictionary, parse_www_form_urlencoded
+from .contents cimport (
+    ASGIContent,
+    Content,
+    multiparts_to_dictionary,
+    parse_www_form_urlencoded,
+)
 from .cookies cimport Cookie, parse_cookie, split_value, write_cookie_for_response
-from .exceptions cimport BadRequest, BadRequestFormat
+from .exceptions cimport BadRequest, BadRequestFormat, MessageAborted
 from .headers cimport Headers
 from .url cimport URL, build_absolute_url
 
@@ -27,6 +33,24 @@ cpdef str parse_charset(bytes value):
     return None
 
 
+async def _read_stream(request):
+    async for _ in request.content.stream():  # type: ignore
+        pass
+
+
+async def _call_soon(coro):
+    """
+    Returns the output of a coroutine if its result is immediately available,
+    otherwise None.
+    """
+    task = asyncio.create_task(coro)
+    asyncio.get_event_loop().call_soon(task.cancel)
+    try:
+        return await task
+    except asyncio.CancelledError:
+        return None
+
+
 cdef class Message:
 
     def __init__(self, list headers):
@@ -60,6 +84,20 @@ cdef class Message:
                 results.append(header[1])
         return results
 
+    cdef void init_prop(self, str name, object value):
+        """
+        This method is for internal use and only accessible in Cython.
+        It initializes a new property on the request object, for rare scenarios
+        where an additional property can be useful. It would also be possible
+        to use a weakref.WeakKeyDictionary to store additional information
+        about request objects when useful, but for simplicity this method uses
+        the object __dict__.
+        """
+        try:
+            getattr(self, name)
+        except AttributeError:
+            setattr(self, name, value)
+
     cdef list get_headers_tuples(self, bytes key):
         cdef list results = []
         cdef tuple header
@@ -491,6 +529,25 @@ cdef class Request(Message):
             return True
         return False
 
+    async def is_disconnected(self):
+        if not isinstance(self.content, ASGIContent):
+            raise TypeError(
+                "This method is only supported when a request is bound to "
+                "an instance of ASGIContent and to an ASGI "
+                "request/response cycle."
+            )
+
+        self.init_prop("_is_disconnected", False)
+        if self._is_disconnected is True:
+            return True
+
+        try:
+            await _call_soon(_read_stream(self))
+        except MessageAborted:
+            self._is_disconnected = True
+
+        return self._is_disconnected
+
 
 cdef class Response(Message):
 

blacksheep/scribe.pxd

@@ -9,6 +9,8 @@ from .cookies cimport Cookie
 from .messages cimport Message, Request, Response
 
 
+cdef int MAX_RESPONSE_CHUNK_SIZE
+
 cpdef bytes get_status_line(int status)
 
 cpdef bint is_small_request(Request request)
@@ -26,4 +28,3 @@ cdef bytes write_small_response(Response response)
 cdef void set_headers_for_content(Message message)
 
 cdef void set_headers_for_response_content(Response message)
-

blacksheep/scribe.pyx

@@ -5,7 +5,8 @@ from .cookies cimport Cookie, write_cookie_for_response
 from .messages cimport Request, Response
 from .url cimport URL
 
-include "includes/consts.pxi"
+
+cdef int MAX_RESPONSE_CHUNK_SIZE = 61440  # 64kb
 
 
 cdef bytes write_header(tuple header):

itests/app_1.py

@@ -17,6 +17,7 @@
     json,
     text,
 )
+from blacksheep.contents import ASGIContent
 from blacksheep.server.compression import use_gzip_compression
 from itests.utils import CrashTest, ensure_folder
 
@@ -242,5 +243,37 @@ async def send_file_with_bytes_io():
     )
 
 
+@app.router.get("/check-disconnected")
+async def check_disconnected(request: Request, expect_disconnected: bool):
+    check_file = pathlib.Path(".is-disconnected.txt")
+    assert await request.is_disconnected() is False
+    # Simulate a delay
+    await asyncio.sleep(0.3)
+
+    if expect_disconnected:
+        # Testing the scenario when the client disconnected
+        assert (
+            await request.is_disconnected()
+        ), "The client disconnected and this should be visible"
+        check_file.write_text("The connection was disconnected")
+    else:
+        assert (
+            await request.is_disconnected() is False
+        ), "The client did not disconnect and this should be visible"
+
+    return "OK"
+
+
+@app.router.get("/read-asgi-receive")
+def check_asgi_receive_readable(request: Request):
+    content = request.content
+    assert isinstance(content, ASGIContent)
+
+    receive = content.receive
+    assert callable(receive)
+
+    return "OK"
+
+
 if __name__ == "__main__":
     uvicorn.run(app, host="127.0.0.1", port=44567, log_level="debug")

itests/requirements.txt

@@ -1,5 +1,5 @@
 Flask
 requests
-uvicorn==0.13.4
-Hypercorn==0.11.2
-websockets==10.1
+uvicorn==0.25.0
+Hypercorn==0.15.0
+websockets==10.1

itests/test_server.py

@@ -1,18 +1,20 @@
 import json
 import os
 import shutil
+import time
 from base64 import urlsafe_b64encode
 from urllib.parse import unquote
 from uuid import uuid4
 
 import pytest
 import websockets
 import yaml
+from requests.exceptions import ReadTimeout
 from websockets.exceptions import ConnectionClosedError, InvalidStatusCode
 
 from .client_fixtures import get_static_path
 from .server_fixtures import *  # NoQA
-from .utils import assert_files_equals, ensure_success, get_test_files_url
+from .utils import assert_files_equals, ensure_success, get_test_files_url, temp_file
 
 
 def test_hello_world(session_1):
@@ -300,7 +302,45 @@ def test_get_file_with_bytesio(session_1):
     ensure_success(response)
 
     text = response.text
-    assert text == """some initial binary data: """
+    assert text == "some initial binary data: "
+
+
+def test_get_is_disconnected_waiting(session_1):
+    response = session_1.get(
+        "/check-disconnected", params={"expect_disconnected": "false"}
+    )
+    ensure_success(response)
+
+    text = response.text
+    assert text == "OK"
+
+
+def test_get_is_disconnected_cancelling(session_1):
+    with temp_file(".is-disconnected.txt") as check_file:
+        try:
+            session_1.get(
+                "/check-disconnected",
+                params={"expect_disconnected": "true"},
+                timeout=0.005,
+            )
+        except ReadTimeout:
+            # wait 310 ms and check a file written by the server after asserting the request
+            # was disconnected
+            time.sleep(0.31)
+
+        try:
+            text = check_file.read_text()
+            assert text == "The connection was disconnected"
+        except FileNotFoundError:
+            pytest.fail("The server did not write the file .is-disconnected.txt")
+
+
+def test_receive_is_accessible_from_python(session_1):
+    response = session_1.get("/read-asgi-receive")
+    ensure_success(response)
+
+    text = response.text
+    assert text == "OK"
 
 
 def test_xml_files_are_not_served(session_1):

itests/utils.py

@@ -1,6 +1,8 @@
 import errno
 import os
 import socket
+from contextlib import contextmanager
+from pathlib import Path
 from urllib.parse import urljoin
 
 import requests
@@ -75,3 +77,15 @@ def get_sleep_time():
 
 def get_test_files_url(url: str):
     return f"my-cdn-foo:{url}"
+
+
+@contextmanager
+def temp_file(name: str):
+    file = Path(name)
+    if file.exists():
+        file.unlink()
+
+    yield file
+
+    if file.exists():
+        file.unlink()