diff --git a/docs/topics/download-handlers.rst b/docs/topics/download-handlers.rst index e0501c169..94e75ab6f 100644 --- a/docs/topics/download-handlers.rst +++ b/docs/topics/download-handlers.rst @@ -203,9 +203,6 @@ Other limitations: - IPv6 support requires setting :setting:`TWISTED_DNS_RESOLVER` to ``scrapy.resolver.CachingHostnameResolver``. -- No support for the :signal:`bytes_received` and :signal:`headers_received` - signals. - Known limitations of the HTTP/2 support: - No support for HTTP/2 Cleartext (h2c), since no major browser supports diff --git a/scrapy/core/downloader/handlers/http2.py b/scrapy/core/downloader/handlers/http2.py index f60c58d1b..9b3d4fbd4 100644 --- a/scrapy/core/downloader/handlers/http2.py +++ b/scrapy/core/downloader/handlers/http2.py @@ -40,7 +40,7 @@ class H2DownloadHandler(BaseDownloadHandler): from twisted.internet import reactor - self._pool = H2ConnectionPool(reactor, crawler.settings) + self._pool = H2ConnectionPool(reactor, crawler) self._context_factory = _load_context_factory_from_settings(crawler) self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS") diff --git a/scrapy/core/http2/agent.py b/scrapy/core/http2/agent.py index aa55e29a0..042557208 100644 --- a/scrapy/core/http2/agent.py +++ b/scrapy/core/http2/agent.py @@ -21,8 +21,8 @@ if TYPE_CHECKING: from twisted.internet.base import ReactorBase from twisted.internet.endpoints import HostnameEndpoint + from scrapy.crawler import Crawler from scrapy.http import Request, Response - from scrapy.settings import Settings from scrapy.spiders import Spider @@ -30,9 +30,9 @@ ConnectionKeyT = tuple[bytes, bytes, int] class H2ConnectionPool: - def __init__(self, reactor: ReactorBase, settings: Settings) -> None: + def __init__(self, reactor: ReactorBase, crawler: Crawler) -> None: self._reactor = reactor - self.settings = settings + self._crawler = crawler # Store a dictionary which is used to get the respective # H2ClientProtocolInstance using the key as Tuple(scheme, hostname, port) @@ -43,7 +43,7 @@ class H2ConnectionPool: ConnectionKeyT, deque[Deferred[H2ClientProtocol]] ] = {} - self._tls_verbose_logging: bool = settings.getbool( + self._tls_verbose_logging: bool = crawler.settings.getbool( "DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING" ) @@ -77,7 +77,7 @@ class H2ConnectionPool: factory = H2ClientFactory( uri, - self.settings, + self._crawler, conn_lost_deferred, tls_verbose_logging=self._tls_verbose_logging, ) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 7136e829e..2d59aba31 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -44,7 +44,7 @@ if TYPE_CHECKING: from twisted.python.failure import Failure from twisted.web.client import URI - from scrapy.settings import Settings + from scrapy.crawler import Crawler from scrapy.spiders import Spider @@ -90,7 +90,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin): def __init__( self, uri: URI, - settings: Settings, + crawler: Crawler, conn_lost_deferred: Deferred[list[BaseException]], *, tls_verbose_logging: bool = False, @@ -100,11 +100,12 @@ class H2ClientProtocol(Protocol, TimeoutMixin): uri -- URI of the base url to which HTTP/2 Connection will be made. uri is used to verify that incoming client requests have correct base URL. - settings -- Scrapy project settings + crawler -- The crawler the requests belong to conn_lost_deferred -- Deferred that fires with the list of underlying exceptions to notify that connection was lost tls_verbose_logging -- Whether to log TLS details """ + self._crawler: Crawler = crawler self._conn_lost_deferred: Deferred[list[BaseException]] = conn_lost_deferred self._tls_verbose_logging: bool = tls_verbose_logging @@ -140,8 +141,8 @@ class H2ClientProtocol(Protocol, TimeoutMixin): # Both ip_address and uri are used by the Stream before # initiating the request to verify that the base address # Variables taken from Project Settings - "default_download_maxsize": settings.getint("DOWNLOAD_MAXSIZE"), - "default_download_warnsize": settings.getint("DOWNLOAD_WARNSIZE"), + "default_download_maxsize": crawler.settings.getint("DOWNLOAD_MAXSIZE"), + "default_download_warnsize": crawler.settings.getint("DOWNLOAD_WARNSIZE"), # Counter to keep track of opened streams. This counter # is used to make sure that not more than MAX_CONCURRENT_STREAMS # streams are opened which leads to ProtocolError @@ -208,6 +209,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin): stream_id=next(self._stream_id_generator), request=request, protocol=self, + crawler=self._crawler, download_maxsize=getattr( spider, "download_maxsize", self.metadata["default_download_maxsize"] ), @@ -461,20 +463,20 @@ class H2ClientFactory(Factory): def __init__( self, uri: URI, - settings: Settings, + crawler: Crawler, conn_lost_deferred: Deferred[list[BaseException]], *, tls_verbose_logging: bool = False, ) -> None: self.uri = uri - self.settings = settings + self.crawler = crawler self.conn_lost_deferred = conn_lost_deferred self.tls_verbose_logging = tls_verbose_logging def buildProtocol(self, addr: IAddress) -> H2ClientProtocol: return H2ClientProtocol( self.uri, - self.settings, + self.crawler, self.conn_lost_deferred, tls_verbose_logging=self.tls_verbose_logging, ) diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index c6226bbca..4fc300d90 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +from contextlib import suppress from enum import Enum from io import BytesIO from typing import TYPE_CHECKING, Any @@ -12,9 +13,11 @@ from twisted.internet.error import ConnectionClosed from twisted.python.failure import Failure from twisted.web.client import ResponseFailed -from scrapy.exceptions import DownloadCancelledError +from scrapy import signals +from scrapy.exceptions import DownloadCancelledError, StopDownload from scrapy.http.headers import Headers from scrapy.utils._download_handlers import ( + check_stop_download, get_maxsize_msg, get_warnsize_msg, make_response, @@ -25,6 +28,7 @@ if TYPE_CHECKING: from collections.abc import Sequence from scrapy.core.http2.protocol import H2ClientProtocol + from scrapy.crawler import Crawler from scrapy.http import Request, Response @@ -82,6 +86,9 @@ class StreamCloseReason(Enum): # Actual response body size is more than allowed limit MAXSIZE_EXCEEDED_ACTUAL = 8 + # A signal handler raised StopDownload + STOP_DOWNLOAD = 9 + class Stream: """Represents a single HTTP/2 Stream. @@ -99,6 +106,7 @@ class Stream: stream_id: int, request: Request, protocol: H2ClientProtocol, + crawler: Crawler, download_maxsize: int = 0, download_warnsize: int = 0, ) -> None: @@ -107,10 +115,13 @@ class Stream: stream_id -- Unique identifier for the stream within a single HTTP/2 connection request -- The HTTP request associated to the stream protocol -- Parent H2ClientProtocol instance + crawler -- The crawler the request belongs to """ self.stream_id: int = stream_id self._request: Request = request self._protocol: H2ClientProtocol = protocol + self._crawler: Crawler = crawler + self._stop_download: StopDownload | None = None self._download_maxsize = self._request.meta.get( "download_maxsize", download_maxsize @@ -338,6 +349,13 @@ class Stream: self._response["body"].write(data) self._response["flow_controlled_size"] += flow_controlled_length + if stop_download := check_stop_download( + signals.bytes_received, self._crawler, self._request, data=data + ): + self._stop_download = stop_download + self.reset_stream(StreamCloseReason.STOP_DOWNLOAD) + return + # We check maxsize here in case the Content-Length header was not received if ( self._download_maxsize @@ -369,8 +387,20 @@ class Stream: else: self._response["headers"].appendlist(name, value) - # Check if we exceed the allowed max data size which can be received expected_size = int(self._response["headers"].get(b"Content-Length", -1)) + + if stop_download := check_stop_download( + signals.headers_received, + self._crawler, + self._request, + headers=self._response["headers"], + body_length=expected_size if expected_size >= 0 else None, + ): + self._stop_download = stop_download + self.reset_stream(StreamCloseReason.STOP_DOWNLOAD) + return + + # Check if we exceed the allowed max data size which can be received if self._download_maxsize and expected_size > self._download_maxsize: self.reset_stream(StreamCloseReason.MAXSIZE_EXCEEDED) return @@ -387,11 +417,18 @@ class Stream: if self.metadata["stream_closed_local"]: raise StreamClosedError(self.stream_id) - # Clear buffer earlier to avoid keeping data in memory for a long time - self._response["body"].truncate(0) + # The data received so far is the body of the response built for a + # stopped download, otherwise the buffer is cleared early to avoid + # keeping data in memory for a long time + if reason is not StreamCloseReason.STOP_DOWNLOAD: + self._response["body"].truncate(0) self.metadata["stream_closed_local"] = True - self._protocol.conn.reset_stream(self.stream_id, ErrorCodes.REFUSED_STREAM) + # The remote peer may have ended the stream already, e.g. because the + # whole response arrived within the data that triggered this reset, in + # which case there is nothing left to reset + with suppress(StreamClosedError): + self._protocol.conn.reset_stream(self.stream_id, ErrorCodes.REFUSED_STREAM) self.close(reason) def close( @@ -444,7 +481,7 @@ class Stream: logger.error(error_msg) self._deferred_response.errback(DownloadCancelledError(error_msg)) - elif reason is StreamCloseReason.ENDED: + elif reason in {StreamCloseReason.ENDED, StreamCloseReason.STOP_DOWNLOAD}: self._fire_response_deferred() # Stream was abruptly ended here @@ -495,13 +532,18 @@ class Stream: and fires the response deferred callback with the generated response instance""" - response = make_response( - url=self._request.url, - status=self._response["status"], - headers=self._response["headers"], - body=self._response["body"].getvalue(), - certificate=self._protocol.metadata["certificate"], - ip_address=self._protocol.metadata["ip_address"], - protocol="h2", - ) - self._deferred_response.callback(response) + try: + response = make_response( + url=self._request.url, + status=self._response["status"], + headers=self._response["headers"], + body=self._response["body"].getvalue(), + certificate=self._protocol.metadata["certificate"], + ip_address=self._protocol.metadata["ip_address"], + protocol="h2", + stop_download=self._stop_download, + ) + except StopDownload as exc: + self._deferred_response.errback(exc) + else: + self._deferred_response.callback(response) diff --git a/tests/test_downloader_handler_twisted_http2.py b/tests/test_downloader_handler_twisted_http2.py index 449f2d635..9d4e161f3 100644 --- a/tests/test_downloader_handler_twisted_http2.py +++ b/tests/test_downloader_handler_twisted_http2.py @@ -186,18 +186,6 @@ class TestHttp2TLSVersion(H2DownloadHandlerMixin, TestHttpsTLSVersionBase): class TestHttp2WithCrawler(H2DownloadHandlerMixin, TestHttpWithCrawlerBase): is_secure = True - def test_bytes_received_stop_download_callback(self) -> None: # type: ignore[override] - pytest.skip("bytes_received support is not implemented") - - def test_bytes_received_stop_download_errback(self) -> None: # type: ignore[override] - pytest.skip("bytes_received support is not implemented") - - def test_headers_received_stop_download_callback(self) -> None: # type: ignore[override] - pytest.skip("headers_received support is not implemented") - - def test_headers_received_stop_download_errback(self) -> None: # type: ignore[override] - pytest.skip("headers_received support is not implemented") - @pytest.mark.skip(reason="Proxy support is not implemented yet") class TestHttp2Proxy(H2DownloadHandlerMixin, TestHttpProxyBase): diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index b8586d1ca..431b7a458 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -23,13 +23,13 @@ from twisted.web.static import File from scrapy.exceptions import DownloadCancelledError, DownloadTimeoutError from scrapy.http import JsonRequest, Request, Response -from scrapy.settings import Settings from scrapy.spiders import Spider from scrapy.utils.defer import ( deferred_f_from_coro_f, deferred_from_coro, maybe_deferred_to_future, ) +from scrapy.utils.test import get_crawler from tests.mockserver.http_resources import LeafResource, Status, put_child from tests.mockserver.utils import ssl_context_factory @@ -250,7 +250,7 @@ class TestHttps2ClientProtocol: acceptableProtocols=[b"h2"], ) uri = URI.fromBytes(bytes(self.get_url(server_port, "/"), "utf-8")) - h2_client_factory = H2ClientFactory(uri, Settings(), Deferred()) + h2_client_factory = H2ClientFactory(uri, get_crawler(), Deferred()) client_endpoint = SSL4ClientEndpoint( reactor, self.host, server_port, client_options )