From d09ccf8d3b2932d9393e6ce4a20c46befdd43acf Mon Sep 17 00:00:00 2001 From: Aditya Date: Sat, 13 Jun 2020 20:40:01 +0530 Subject: [PATCH] feat(http2): support for POST requests BREAKING CHANGES - Request is sent successfully with its Response received as well. However, the StreamEnded event is not received which do not fires the response deferred --- scrapy/core/http2/protocol.py | 108 +++++++++++++++++----------------- scrapy/core/http2/stream.py | 79 +++++++++++++++++++++++-- 2 files changed, 127 insertions(+), 60 deletions(-) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 847e74f97..6d926b100 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -1,4 +1,5 @@ import logging +from typing import Dict, List from h2.config import H2Configuration from h2.connection import H2Connection @@ -7,45 +8,19 @@ from h2.events import ( StreamEnded, StreamReset, TrailersReceived, WindowUpdated ) from twisted.internet.protocol import connectionDone, Protocol -from zope.interface import implementer, Interface from scrapy.core.http2.stream import Stream from scrapy.http import Request LOGGER = logging.getLogger(__name__) +LOGGER.debug = print -class IH2EventsHandler(Interface): - def connection_terminated(event: ConnectionTerminated): - pass - - def data_received(event: DataReceived): - pass - - def response_received(event: ResponseReceived): - pass - - def remote_settings_changed(event: RemoteSettingsChanged): - pass - - def stream_ended(event: StreamEnded): - pass - - def stream_reset(event: StreamReset): - pass - - def trailers_received(event: TrailersReceived): - pass - - def window_updated(event: WindowUpdated): - pass - - -@implementer(IH2EventsHandler) class H2ClientProtocol(Protocol): - # TODO: Check for user-agent while testing - # TODO: Add support for cookies - # TODO: Handle priority updates + # TODO: + # 1. Check for user-agent while testing + # 2. Add support for cookies + # 3. Handle priority updates def __init__(self): config = H2Configuration(client_side=True, header_encoding='utf-8') @@ -57,14 +32,14 @@ class H2ClientProtocol(Protocol): self.next_stream_id = 1 # Streams are stored in a dictionary keyed off their stream IDs - self.streams = {} + self.streams: Dict[int, Stream] = {} # Boolean to keep track the connection is made # If requests are received before connection is made # we keep all requests in a pool and send them as the connection # is made self.is_connection_made = False - self._pending_request_stream_pool = [] + self._pending_request_stream_pool: List[Stream] = [] def _new_stream(self, request: Request): """Instantiates a new Stream object @@ -100,29 +75,27 @@ 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.info("Connection made to {}".format(self.transport)) + LOGGER.debug("Connection made to {}".format(self.transport)) self.conn.initiate_connection() self._write_to_transport() self.is_connection_made = True - # Initiate all pending requests - for stream in self._pending_request_stream_pool: - assert isinstance(stream, Stream) - stream.initiate_request() - - self._pending_request_stream_pool.clear() - def dataReceived(self, data): events = self.conn.receive_data(data) self._handle_events(events) self._write_to_transport() def connectionLost(self, reason=connectionDone): - """Called by Twisted when the transport connection is lost. """ - for stream_id in self.streams.keys(): + LOGGER.debug(f"connectionLost {reason}") + stream_ids = list(self.streams.keys()) + + for stream in self._pending_request_stream_pool: + stream_ids.remove(stream.stream_id) + + for stream_id in stream_ids: # TODO: Close each Stream instance in a clean manner self.conn.end_stream(stream_id) @@ -135,6 +108,7 @@ class H2ClientProtocol(Protocol): triggered by sending data """ for event in events: + LOGGER.debug(event) if isinstance(event, ConnectionTerminated): self.connection_terminated(event) elif isinstance(event, DataReceived): @@ -153,38 +127,62 @@ class H2ClientProtocol(Protocol): self.remote_settings_changed(event) def send_headers(self, stream_id, headers): - """ Send the headers for a given stream to the resource + """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 """ - if stream_id in self.streams: - self.conn.send_headers(stream_id, headers, end_stream=True) - self._write_to_transport() - else: - pass + 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) + + # Event handler functions starts here def connection_terminated(self, event: ConnectionTerminated): pass def data_received(self, event: DataReceived): stream_id = event.stream_id - # TODO: Stream do not exist in self.streams dict self.streams[stream_id].receive_data(event.data) def response_received(self, event: ResponseReceived): stream_id = event.stream_id - # TODO: Stream do not exist in self.streams dict self.streams[stream_id].receive_headers(event.headers) def remote_settings_changed(self, event: RemoteSettingsChanged): - pass + # 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 - # TODO: Stream do not exist in self.streams dict self.streams[stream_id].end_stream() def stream_reset(self, event: StreamReset): @@ -194,4 +192,6 @@ class H2ClientProtocol(Protocol): pass def window_updated(self, event: WindowUpdated): - pass + stream_id = event.stream_id + if stream_id != 0: + self.streams[stream_id].window_updated() diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index d2a9f02fa..ab5a0fc88 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -23,13 +23,23 @@ class Stream: stream_id {int} -- For one HTTP/2 connection each stream is uniquely identified by a single integer request {Request} -- HTTP request - connection {H2ClientProtocol} -- HTTP/2 connection this stream belongs to + connection {H2Connection} -- HTTP/2 connection this stream belongs to. """ - self.stream_id = stream_id self._request = request - self._conn = connection + self._client_protocol = connection + self._request_body = self._request.body + self.content_length = 0 if self._request_body is None else len(self._request_body) + + # Each time we send a data frame, we will decrease value by the amount send. + self.remaining_content_length = self.content_length + + # Flag to keep track whether we have ended this stream + self.stream_ended = True + + # Data received frame by frame from the server is appended + # and passed to the response Deferred when completely received. self._response_data = b"" # Headers received after sending the request @@ -59,10 +69,67 @@ class Stream: # TODO: Check if scheme can be "http" for HTTP/2 ? (":scheme", "https"), - (":path", url.path) + (":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._conn.send_headers(self.stream_id, http2_request_headers) + self._client_protocol.send_headers(self.stream_id, http2_request_headers) + self.send_data() + + def send_data(self): + """Called immediately after the headers are sent. Here we send all the + data as part of the request. + + If the content length is 0 initially then we end the stream immediately and + wait for response data. + """ + + # TODO: + # 1. Add test for sending very large data + # 2. Add test for small data + # 3. Both (1) and (2) should be tested for + # 3.1 Large number of request + # 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) + + # Next, check what the maximum frame size is. + max_frame_size = self._client_protocol.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. + bytes_to_send = min(window_size, self.remaining_content_length) + + # We now need to send a number of data frames. + while bytes_to_send > 0: + chunk_size = min(bytes_to_send, max_frame_size) + + 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) + + 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) + else: + # TODO: Continue from here :) + pass + + def window_updated(self): + """Flow control window size was changed. + Send data that earlier could not be sent as we were + blocked behind the flow control. + """ + if self.remaining_content_length > 0 and not self.stream_ended: + self.send_data() def receive_data(self, data: bytes): self._response_data += data @@ -72,7 +139,7 @@ class Stream: self._response_headers[name] = value def end_stream(self): - """Stream is ended by the resource hence no further + """Stream is ended by the server hence no further data or headers should be expected on this stream. We will call the response deferred callback passing