RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(http2): POST request not sending large body

This commit is contained in:
Aditya 2020-06-21 00:33:34 +05:30
parent 700df3eeb7
commit 303485a9b4
2 changed files with 138 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
scrapy/core/http2/protocol.py
View File

@ -1,3 +1,4 @@
import ipaddress
import itertools
import logging
from collections import deque
@ -5,7 +6,7 @@ from collections import deque
from h2.config import H2Configuration
from h2.connection import H2Connection
from h2.events import (
ConnectionTerminated, DataReceived, ResponseReceived,
DataReceived, ResponseReceived, SettingsAcknowledged,
StreamEnded, StreamReset, WindowUpdated
)
from twisted.internet.protocol import connectionDone, Protocol
@ -49,6 +50,13 @@ class H2ClientProtocol(Protocol):
self.is_connection_made = False
self._pending_request_stream_pool = deque()
# Some meta data of this connection
# initialized when connection is successfully made
self._metadata = {
'certificate': None,
'ip_address': None
}
def _stream_close_cb(self, stream_id: int):
"""Called when stream is closed completely
"""
@ -63,6 +71,7 @@ class H2ClientProtocol(Protocol):
stream_id=stream_id,
request=request,
connection=self.conn,
metadata=self._metadata,
write_to_transport=self._write_to_transport,
cb_close=self._stream_close_cb
)
@ -73,7 +82,7 @@ class H2ClientProtocol(Protocol):
def _send_pending_requests(self):
# TODO: handle MAX_CONCURRENT_STREAMS
# Initiate all pending requests
while len(self._pending_request_stream_pool):
while self._pending_request_stream_pool:
stream = self._pending_request_stream_pool.popleft()
stream.initiate_request()
@ -84,6 +93,8 @@ class H2ClientProtocol(Protocol):
data = self.conn.data_to_send()
self.transport.write(data)
LOGGER.debug("Sent {} bytes to {} via transport".format(len(data), self._metadata['ip_address']))
def request(self, _request: Request):
stream = self._new_stream(_request)
d = stream.get_response()
@ -101,17 +112,16 @@ 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.
"""
self.destination = self.transport.connector.getDestination()
self.destination = self.transport.getPeer()
LOGGER.info('Connection made to {}'.format(self.destination))
self._metadata['certificate'] = self.transport.getPeerCertificate()
self._metadata['ip_address'] = ipaddress.ip_address(self.destination.host)
self.conn.initiate_connection()
self._write_to_transport()
self.is_connection_made = True
# Send off all the pending requests
# as now we have established a proper HTTP/2 connection
self._send_pending_requests()
def dataReceived(self, data):
events = self.conn.receive_data(data)
self._handle_events(events)
@ -123,7 +133,7 @@ class H2ClientProtocol(Protocol):
"""
# Pop all streams which were pending and were not yet started
for stream_id in list(self.streams):
self.streams[stream_id].lost_connection()
self.streams[stream_id].close()
self.conn.close_connection()
@ -139,9 +149,7 @@ class H2ClientProtocol(Protocol):
"""
for event in events:
LOGGER.debug(event)
if isinstance(event, ConnectionTerminated):
self.connection_terminated(event)
elif isinstance(event, DataReceived):
if isinstance(event, DataReceived):
self.data_received(event)
elif isinstance(event, ResponseReceived):
self.response_received(event)
@ -151,13 +159,12 @@ class H2ClientProtocol(Protocol):
self.stream_reset(event)
elif isinstance(event, WindowUpdated):
self.window_updated(event)
elif isinstance(event, SettingsAcknowledged):
self.settings_acknowledged(event)
else:
LOGGER.info("Received unhandled event {}".format(event))
# Event handler functions starts here
def connection_terminated(self, event: ConnectionTerminated):
pass
def data_received(self, event: DataReceived):
stream_id = event.stream_id
self.streams[stream_id].receive_data(event.data, event.flow_controlled_length)
@ -166,20 +173,26 @@ class H2ClientProtocol(Protocol):
stream_id = event.stream_id
self.streams[stream_id].receive_headers(event.headers)
def settings_acknowledged(self, event: SettingsAcknowledged):
# Send off all the pending requests
# as now we have established a proper HTTP/2 connection
self._send_pending_requests()
def stream_ended(self, event: StreamEnded):
stream_id = event.stream_id
self.streams[stream_id].end_stream()
self.streams[stream_id].close()
def stream_reset(self, event: StreamReset):
# TODO: event.stream_id was abruptly closed
# Q. What should be the response? (Failure/Partial/???)
self.streams[event.stream_id].reset()
self.streams[event.stream_id].close(event)
def window_updated(self, event: WindowUpdated):
stream_id = event.stream_id
if stream_id != 0:
self.streams[stream_id].receive_window_update(event.delta)
else:
# TODO:
# Q. What to do when StreamID=0 ?
pass
# Send leftover data for all the streams
for stream in self.streams.values():
if stream.request_sent:
stream.send_data()

148
scrapy/core/http2/stream.py
View File

@ -1,11 +1,15 @@
import logging
from typing import Dict
from urllib.parse import urlparse
from h2.connection import H2Connection
from h2.events import StreamEnded
from h2.exceptions import StreamClosedError
from twisted.internet.defer import Deferred
from scrapy.http import Request, Response
from scrapy.http import Request
from scrapy.http.headers import Headers
from scrapy.responsetypes import responsetypes
LOGGER = logging.getLogger(__name__)
@ -22,12 +26,13 @@ class Stream:
"""
def __init__(
self,
stream_id: int,
request: Request,
connection: H2Connection,
write_to_transport,
cb_close
self,
stream_id: int,
request: Request,
connection: H2Connection,
metadata: Dict,
write_to_transport,
cb_close
):
"""
Arguments:
@ -35,6 +40,7 @@ class Stream:
uniquely identified by a single integer
request {Request} -- HTTP request
connection {H2Connection} -- HTTP/2 connection this stream belongs to.
metadata {Dict} -- Reference to dictionary having metadata of HTTP/2 connection
write_to_transport {callable} -- Method used to write & send data to the server
This method should be used whenever some frame is to be sent to the server.
cb_close {callable} -- Method called when this stream is closed
@ -43,12 +49,16 @@ class Stream:
self.stream_id = stream_id
self._request = request
self._conn = connection
self._metadata = metadata
self._write_to_transport = write_to_transport
self._cb_close = cb_close
self._request_body = self._request.body
self.content_length = 0 if self._request_body is None else len(self._request_body)
# Flag to keep track whether this stream has initiated the request
self.request_sent = False
# Each time we send a data frame, we will decrease value by the amount send.
self.remaining_content_length = self.content_length
@ -58,20 +68,30 @@ class Stream:
# Flag to keep track whether the server has closed the stream
self.stream_closed_server = False
# Data received frame by frame from the server is appended
# and passed to the response Deferred when completely received.
self._response_data = b""
# The amount of data received that counts against the flow control
# window
self._response_flow_controlled_size = 0
# Headers received after sending the request
self._response_headers = Headers({})
# Private variable used to build the response
# this response is then converted to appropriate Response class
# passed to the response deferred callback
self._response = {
# Data received frame by frame from the server is appended
# and passed to the response Deferred when completely received.
'body': b'',
# Headers received after sending the request
'headers': Headers({})
}
# TODO: Add canceller for the Deferred below
self._deferred_response = Deferred()
def __str__(self):
return "Stream(id={})".format(self.stream_id)
__repr__ = __str__
def get_response(self):
"""Simply return a Deferred which fires when response
from the asynchronous request is available
@ -104,6 +124,8 @@ class Stream:
self._conn.send_headers(self.stream_id, headers, end_stream=False)
self._write_to_transport()
self.request_sent = True
self.send_data()
def send_data(self):
@ -112,7 +134,18 @@ class Stream:
If the content length is 0 initially then we end the stream immediately and
wait for response data.
Warning: Only call this method when stream not closed from client side
and has initiated request already by sending HEADER frame. If not then
stream will be closed from client side with 499 response.
TODO: Q. Should we instead raise ProtocolError here with a proper message?
"""
if self.stream_closed_local or self.stream_closed_server:
raise StreamClosedError(self.stream_id)
elif not self.request_sent:
self.close()
return
# TODO:
# 1. Add test for sending very large data
@ -132,7 +165,8 @@ class Stream:
bytes_to_send_size = min(window_size, self.remaining_content_length)
# We now need to send a number of data frames.
while bytes_to_send_size:
data_frames_sent = 0
while bytes_to_send_size > 0:
chunk_size = min(bytes_to_send_size, max_frame_size)
data_chunk_start_id = self.content_length - self.remaining_content_length
@ -140,16 +174,24 @@ class Stream:
self._conn.send_data(self.stream_id, data_chunk, end_stream=False)
data_frames_sent += 1
bytes_to_send_size = bytes_to_send_size - chunk_size
self.remaining_content_length = self.remaining_content_length - chunk_size
# End the stream if no more data has to be send
if self.remaining_content_length <= 0:
self.remaining_content_length = 0
self.remaining_content_length = max(0, self.remaining_content_length)
LOGGER.debug("{} sending {}/{} data bytes ({} frames) to {}".format(
self,
self.content_length - self.remaining_content_length, self.content_length,
data_frames_sent,
self._metadata['ip_address'])
)
# End the stream if no more data needs to be send
if self.remaining_content_length == 0:
self.stream_closed_local = True
self._conn.end_stream(self.stream_id)
# Write data to transport -- Empty the outstanding data
self._write_to_transport()
# Q. What about the rest of the data?
@ -163,11 +205,11 @@ class Stream:
Arguments:
delta -- Window change delta
"""
if self.stream_closed_local is False:
if self.remaining_content_length > 0 and not self.stream_closed_server:
self.send_data()
def receive_data(self, data: bytes, flow_controlled_length: int):
self._response_data += data
self._response['body'] += data
self._response_flow_controlled_size += flow_controlled_length
# Acknowledge the data received
@ -178,47 +220,69 @@ class Stream:
def receive_headers(self, headers):
for name, value in headers:
self._response_headers[name] = value
self._response['headers'][name] = value
def reset(self):
"""Received a RST_STREAM -- forcefully reset"""
# TODO:
# Q1. Do we need to send the request again?
# Q2. What response should we send now?
self.stream_closed_server = True
self._cb_close(self.stream_id)
def close(self, event=None):
"""Based on the event sent we will handle each case.
def lost_connection(self):
# TODO: Same as self.reset
self.stream_closed_server = True
self._cb_close(self.stream_id)
def end_stream(self):
"""Stream is ended by the server hence no further
event: StreamEnded
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
the response object
event: StreamReset
Stream reset via RST_FRAME by the upstream hence forcefully close
this stream and send TODO: ?
event: None
No event is launched -- Hence we will simply close this stream
"""
# TODO: In case of abruptly stream close
# Q1. Do we need to send the request again?
# Q2. What response should we send now?
assert self.stream_closed_server is False
self.stream_closed_server = True
if not isinstance(event, StreamEnded):
# TODO
# Stream was abruptly ended here
# Partial - Content-Length header not provided
pass
self._fire_response_deferred()
self._cb_close(self.stream_id)
def _fire_response_deferred(self):
def _fire_response_deferred(self, flags=None):
"""Builds response from the self._response dict
and fires the response deferred callback with the
generated response instance"""
# TODO:
# 1. Set flags, certificate, ip_address in response
# 2. Should we fire this in case of
# 2.1 StreamReset in between when data is received partially
# 2.2 Forcefully closed the stream
# 3. Update Client Side Status Codes here
# NOTE: Presently on fired with successful response
response = Response(
response_cls = responsetypes.from_args(
headers=self._response['headers'],
url=self._request.url,
status=self._response_headers[":status"],
headers=self._response_headers,
body=self._response_data,
request=self._request
body=self._response['body']
)
# If there is :status in headers then
# HTTP Status Code: 499 - Client Closed Request
status = self._response['headers'].get(':status', '499')
response = response_cls(
url=self._request.url,
status=status,
headers=self._response['headers'],
body=self._response['body'],
request=self._request,
flags=flags,
certificate=self._metadata['certificate'],
ip_address=self._metadata['ip_address']
)
self._deferred_response.callback(response)