mirror of https://github.com/scrapy/scrapy.git
refactor: move H2Connection instance to stream
- Remove all wrapper funtions made such that stream can send header/data
to H2Connection as they were not necessary
BREAKING CHANGES
Looks like, for small set of response data the StreamEnded event is
emitted and everything works well -- tested for both GET & POST request.
Maybe some issue with window size and/or flow control as when the
response data needs to be broken into separate chunks -- not all chunks
are received everytime which leads to indefinite waiting for next data
chunk and the connection is lost due to timeout. 😥
Working on setting up testing environment now. After testing is setup
I'll debug the above bug furthur.
This commit is contained in:
Aditya
parent
d09ccf8d3b
commit
d06bb12e35
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue