From 9ff9caecadf8215ce75b0ad4231b8289bca168fe Mon Sep 17 00:00:00 2001 From: Aditya Date: Sun, 7 Jun 2020 14:04:53 +0530 Subject: [PATCH] feat(http2): support for GET requests --- scrapy/core/http2/protocol.py | 126 ++++++++++++++++++++++------------ scrapy/core/http2/stream.py | 71 +++++++++++++++---- 2 files changed, 139 insertions(+), 58 deletions(-) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index b95cb05ae..847e74f97 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -1,20 +1,19 @@ -from h2.connection import H2Connection +import logging + from h2.config import H2Configuration +from h2.connection import H2Connection from h2.events import ( - ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded, - StreamReset, TrailersReceived, WindowUpdated + ConnectionTerminated, DataReceived, ResponseReceived, RemoteSettingsChanged, + StreamEnded, StreamReset, TrailersReceived, WindowUpdated ) - -from scrapy.http import Request -from scrapy.core.http2.stream import Stream - -from twisted.internet.defer import Deferred -from twisted.internet.protocol import Protocol - -from urllib.parse import urlparse - +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__) + class IH2EventsHandler(Interface): def connection_terminated(event: ConnectionTerminated): @@ -26,6 +25,9 @@ class IH2EventsHandler(Interface): def response_received(event: ResponseReceived): pass + def remote_settings_changed(event: RemoteSettingsChanged): + pass + def stream_ended(event: StreamEnded): pass @@ -46,7 +48,7 @@ class H2ClientProtocol(Protocol): # TODO: Handle priority updates def __init__(self): - config = H2Configuration(client_side=True) + config = H2Configuration(client_side=True, header_encoding='utf-8') self.conn = H2Connection(config=config) # ID of the next request stream @@ -57,60 +59,69 @@ class H2ClientProtocol(Protocol): # Streams are stored in a dictionary keyed off their stream IDs self.streams = {} - def _new_stream(self, headers): + # 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 = [] + + def _new_stream(self, request: Request): """Instantiates a new Stream object """ - stream = Stream(self.next_stream_id, headers) - + stream = Stream(self.next_stream_id, request, self) self.next_stream_id += 2 + self.streams[stream.stream_id] = stream return stream + 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) + def request(self, _request: Request): - """ - - Arguments: - _request {Request} -- [description] - """ - url = urlparse(_request.url) - - _request[":method"] = _request.method - - # TODO: Make authority private class variable instead - # of parsing it from request url all requests to same - # host are multiplexed into one connection & a connection - # can have only 1 host at a time - _request[":authority"] = url.netloc - - # TODO: Check if scheme can be 'http' for HTTP/2 ? - _request[":scheme"] = "https" - _request[":path"] = url.path - - stream = self._new_stream(_request.headers) + stream = self._new_stream(_request) d = stream.get_response() - return d + # If connection is not yet established then add the + # stream to pool or initiate request + if self.is_connection_made: + stream.initiate_request() + else: + self._pending_request_stream_pool.append(stream) + return d def connectionMade(self): """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)) self.conn.initiate_connection() - self.transport.write(self.conn.data_to_send()) + 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() - _data = self.conn.data_to_send() - if _data: - self.transport.write(data) + def connectionLost(self, reason=connectionDone): - def connectionLost(self, reason): """Called by Twisted when the transport connection is lost. """ - for stream_id in self.streams.keys(): # TODO: Close each Stream instance in a clean manner self.conn.end_stream(stream_id) @@ -138,18 +149,43 @@ class H2ClientProtocol(Protocol): 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. + + 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 def connection_terminated(self, event: ConnectionTerminated): pass def data_received(self, event: DataReceived): - pass + 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 def stream_ended(self, event: StreamEnded): - pass + 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): pass diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index 558a6552c..d2a9f02fa 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -1,3 +1,8 @@ +from urllib.parse import urlparse + +from twisted.internet.defer import Deferred + +from scrapy.http import Request, Response from scrapy.http.headers import Headers @@ -5,24 +10,30 @@ class Stream: """Represents a single HTTP/2 Stream. Stream is a bidirectional flow of bytes within an established connection, - which may carry one or more messages. Handles the tranfer of HTTP Headers + which may carry one or more messages. Handles the transfer of HTTP Headers and Data frames. + + Role of this class is to + 1. Combine all the data frames """ - def __init__(self, stream_id, headers): + def __init__(self, stream_id: int, request: Request, connection): """ Arguments: stream_id {int} -- For one HTTP/2 connection each stream is uniquely identified by a single integer - headers {Headers} -- HTTP request headers + request {Request} -- HTTP request + connection {H2ClientProtocol} -- HTTP/2 connection this stream belongs to """ - # Headers received after sending the request - self.response_headers = Headers({}) + self.stream_id = stream_id + self._request = request + self._conn = connection - # Headers which are send with the request - # These cannot be modified any furthur - self._request_headers = headers + self._response_data = b"" + + # Headers received after sending the request + self._response_headers = Headers({}) # TODO: Add canceller for the Deferred below self._deferred_response = Deferred() @@ -32,13 +43,47 @@ class Stream: from the asynchronous request is available Returns: - Deferred -- Calls the callback when the response is - avaialble + Deferred -- Calls the callback passing the response """ return self._deferred_response - def receive_data(self, data): - pass + def initiate_request(self): + http2_request_headers = [] + for name, value in self._request.headers.items(): + http2_request_headers.append((name, value)) + + url = urlparse(self._request.url) + http2_request_headers += [ + (":method", self._request.method), + (":authority", url.netloc), + + # TODO: Check if scheme can be "http" for HTTP/2 ? + (":scheme", "https"), + (":path", url.path) + ] + + self._conn.send_headers(self.stream_id, http2_request_headers) + + def receive_data(self, data: bytes): + self._response_data += data def receive_headers(self, headers): - pass + for name, value in headers: + self._response_headers[name] = value + + def end_stream(self): + """Stream is ended by the resource hence no further + data or headers should be expected on this stream. + + We will call the response deferred callback passing + the response object + """ + # TODO: Set flags, certificate, ip_address + response = Response( + url=self._request.url, + status=self._response_headers[":status"], + headers=self._response_headers, + body=self._response_data, + request=self._request + ) + self._deferred_response.callback(response)