diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 6d926b100..4036dfb3e 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -4,8 +4,7 @@ from typing import Dict, List from h2.config import H2Configuration from h2.connection import H2Connection from h2.events import ( - ConnectionTerminated, DataReceived, ResponseReceived, RemoteSettingsChanged, - StreamEnded, StreamReset, TrailersReceived, WindowUpdated + ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded, StreamReset, WindowUpdated ) from twisted.internet.protocol import connectionDone, Protocol @@ -13,7 +12,6 @@ from scrapy.core.http2.stream import Stream from scrapy.http import Request LOGGER = logging.getLogger(__name__) -LOGGER.debug = print class H2ClientProtocol(Protocol): @@ -44,19 +42,27 @@ class H2ClientProtocol(Protocol): def _new_stream(self, request: Request): """Instantiates a new Stream object """ - stream = Stream(self.next_stream_id, request, self) + stream = Stream(self.next_stream_id, request, self.conn) self.next_stream_id += 2 self.streams[stream.stream_id] = stream return stream + def _send_pending_requests(self): + # TODO: handle MAX_CONCURRENT_STREAMS + # Initiate all pending requests + for stream in self._pending_request_stream_pool: + stream.initiate_request() + self._write_to_transport() + + self._pending_request_stream_pool.clear() + def _write_to_transport(self): """ Write data to the underlying transport connection from the HTTP2 connection instance if any """ data = self.conn.data_to_send() - if data: - self.transport.write(data) + self.transport.write(data) def request(self, _request: Request): stream = self._new_stream(_request) @@ -75,10 +81,11 @@ class H2ClientProtocol(Protocol): """Called by Twisted when the connection is established. We can start sending some data now: we should open with the connection preamble. """ - LOGGER.debug("Connection made to {}".format(self.transport)) self.conn.initiate_connection() self._write_to_transport() + self._send_pending_requests() + self.is_connection_made = True def dataReceived(self, data): @@ -89,7 +96,6 @@ class H2ClientProtocol(Protocol): def connectionLost(self, reason=connectionDone): """Called by Twisted when the transport connection is lost. """ - LOGGER.debug(f"connectionLost {reason}") stream_ids = list(self.streams.keys()) for stream in self._pending_request_stream_pool: @@ -119,47 +125,10 @@ class H2ClientProtocol(Protocol): self.stream_ended(event) elif isinstance(event, StreamReset): self.stream_reset(event) - elif isinstance(event, TrailersReceived): - self.trailers_received(event) elif isinstance(event, WindowUpdated): self.window_updated(event) - elif isinstance(event, RemoteSettingsChanged): - self.remote_settings_changed(event) - - def send_headers(self, stream_id, headers): - """Send the headers for a given stream to the resource - Initiates a new connection hence. - This function is wrapper for :func:`~h2.connection.H2Connection.send_headers` - - Arguments: - stream_id {int} -- Valid stream id - headers {List[Tuple[str, str]]} -- Headers of the request - """ - LOGGER.debug(f'Send Headers: stream_id={stream_id} headers={headers}') - self.conn.send_headers(stream_id, headers, end_stream=False) - - def send_data(self, stream_id, data): - """Send the data for a given stream to the resource. - Requires request headers to be sent at least once before this - function is called. - This function is wrapper for :func:`~h2.connection.H2Connection.send_data` - - Arguments: - stream_id {int} -- Valid stream id - data {bytes} -- The data to send on the stream. - """ - LOGGER.debug(f"Send Data: stream_id={stream_id} data={data}") - self.conn.send_data(stream_id, data, end_stream=False) - - def end_stream(self, stream_id): - """End the given stream. - This function is wrapper for :func:`~h2.connection.H2Connection.end_stream` - - Arguments: - stream_id {int} - Valid stream id - """ - LOGGER.debug(f"End Stream: stream_id={stream_id}") - self.conn.end_stream(stream_id) + else: + LOGGER.info("Received unhandled event {}".format(event)) # Event handler functions starts here def connection_terminated(self, event: ConnectionTerminated): @@ -173,14 +142,6 @@ class H2ClientProtocol(Protocol): stream_id = event.stream_id self.streams[stream_id].receive_headers(event.headers) - def remote_settings_changed(self, event: RemoteSettingsChanged): - # TODO: handle MAX_CONCURRENT_STREAMS - # Initiate all pending requests - for stream in self._pending_request_stream_pool: - stream.initiate_request() - - self._pending_request_stream_pool.clear() - def stream_ended(self, event: StreamEnded): stream_id = event.stream_id self.streams[stream_id].end_stream() @@ -188,9 +149,6 @@ class H2ClientProtocol(Protocol): def stream_reset(self, event: StreamReset): pass - def trailers_received(self, event: TrailersReceived): - pass - def window_updated(self, event: WindowUpdated): stream_id = event.stream_id if stream_id != 0: diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index ab5a0fc88..b37755042 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -1,5 +1,6 @@ from urllib.parse import urlparse +from h2.connection import H2Connection from twisted.internet.defer import Deferred from scrapy.http import Request, Response @@ -17,7 +18,7 @@ class Stream: 1. Combine all the data frames """ - def __init__(self, stream_id: int, request: Request, connection): + def __init__(self, stream_id: int, request: Request, connection: H2Connection): """ Arguments: stream_id {int} -- For one HTTP/2 connection each stream is @@ -27,7 +28,7 @@ class Stream: """ self.stream_id = stream_id self._request = request - self._client_protocol = connection + self._conn = connection self._request_body = self._request.body self.content_length = 0 if self._request_body is None else len(self._request_body) @@ -70,13 +71,12 @@ class Stream: # TODO: Check if scheme can be "http" for HTTP/2 ? (":scheme", "https"), (":path", url.path), - # ("Content-Length", str(self.content_length)) # TODO: Make sure 'Content-Type' and 'Content-Encoding' headers # are sent for request having body ] - self._client_protocol.send_headers(self.stream_id, http2_request_headers) + self._conn.send_headers(self.stream_id, http2_request_headers, end_stream=False) self.send_data() def send_data(self): @@ -95,10 +95,10 @@ class Stream: # 3.2 Small number of requests # Firstly, check what the flow control window is for current stream. - window_size = self._client_protocol.conn.local_flow_control_window(stream_id=self.stream_id) + window_size = self._conn.local_flow_control_window(stream_id=self.stream_id) # Next, check what the maximum frame size is. - max_frame_size = self._client_protocol.conn.max_outbound_frame_size + max_frame_size = self._conn.max_outbound_frame_size # We will send no more than the window size or the remaining file size # of data in this call, whichever is smaller. @@ -111,14 +111,14 @@ class Stream: data_chunk_start = self.content_length - self.remaining_content_length data_chunk = self._request_body[data_chunk_start:data_chunk_start + chunk_size] - self._client_protocol.send_data(self.stream_id, data_chunk, end_stream=False) + self._conn.send_data(self.stream_id, data_chunk, end_stream=False) bytes_to_send = max(0, bytes_to_send - chunk_size) self.remaining_content_length = max(0, self.remaining_content_length - chunk_size) # End the stream if no more data has to be send if self.remaining_content_length == 0: - self._client_protocol.end_stream(self.stream_id) + self._conn.end_stream(self.stream_id) else: # TODO: Continue from here :) pass