Merge pull request #2622 from rolando-contrib/download-maxsize-abort

[MRG+1] Abort connection earlier and avoid to buffer data when max size limit is reached

scrapy/core/downloader/handlers/http11.py

@@ -348,7 +348,8 @@ class ScrapyAgent(object):
                            {'size': expected_size, 'warnsize': warnsize})
 
         def _cancel(_):
-            txresponse._transport._producer.loseConnection()
+            # Abort connection inmediately.
+            txresponse._transport._producer.abortConnection()
 
         d = defer.Deferred(_cancel)
         txresponse.deliverBody(_ResponseReader(
@@ -401,6 +402,11 @@ class _ResponseReader(protocol.Protocol):
         self._bytes_received = 0
 
     def dataReceived(self, bodyBytes):
+        # This maybe called several times after cancel was called with buffered
+        # data.
+        if self._finished.called:
+            return
+
         self._bodybuf.write(bodyBytes)
         self._bytes_received += len(bodyBytes)
 
@@ -409,6 +415,9 @@ class _ResponseReader(protocol.Protocol):
                          "max size (%(maxsize)s).",
                          {'bytes': self._bytes_received,
                           'maxsize': self._maxsize})
+            # Clear buffer earlier to avoid keeping data in memory for a long
+            # time.
+            self._bodybuf.truncate(0)
             self._finished.cancel()
 
         if self._warnsize and self._bytes_received > self._warnsize and not self._reached_warnsize:

tests/test_downloader_handlers.py

@@ -177,6 +177,16 @@ class EmptyContentTypeHeaderResource(resource.Resource):
         return request.content.read()
 
 
+class LargeChunkedFileResource(resource.Resource):
+    def render(self, request):
+        def response():
+            for i in range(1024):
+                request.write(b"x" * 1024)
+            request.finish()
+        reactor.callLater(0, response)
+        return server.NOT_DONE_YET
+
+
 class HttpTestCase(unittest.TestCase):
 
     scheme = 'http'
@@ -202,6 +212,7 @@ class HttpTestCase(unittest.TestCase):
         r.putChild(b"broken-chunked", BrokenChunkedResource())
         r.putChild(b"contentlength", ContentLengthHeaderResource())
         r.putChild(b"nocontenttype", EmptyContentTypeHeaderResource())
+        r.putChild(b"largechunkedfile", LargeChunkedFileResource())
         r.putChild(b"echo", Echo())
         self.site = server.Site(r, timeout=None)
         self.wrapper = WrappingFactory(self.site)
@@ -384,6 +395,25 @@ class Http11TestCase(HttpTestCase):
         d = self.download_request(request, Spider('foo', download_maxsize=9))
         yield self.assertFailure(d, defer.CancelledError, error.ConnectionAborted)
 
+    @defer.inlineCallbacks
+    def test_download_with_maxsize_very_large_file(self):
+        with mock.patch('scrapy.core.downloader.handlers.http11.logger') as logger:
+            request = Request(self.getURL('largechunkedfile'))
+
+            def check(logger):
+                logger.error.assert_called_once_with(mock.ANY, mock.ANY)
+
+            d = self.download_request(request, Spider('foo', download_maxsize=1500))
+            yield self.assertFailure(d, defer.CancelledError, error.ConnectionAborted)
+
+            # As the error message is logged in the dataReceived callback, we
+            # have to give a bit of time to the reactor to process the queue
+            # after closing the connection.
+            d = defer.Deferred()
+            d.addCallback(check)
+            reactor.callLater(.1, d.callback, logger)
+            yield d
+
     @defer.inlineCallbacks
     def test_download_with_maxsize_per_req(self):
         meta = {'download_maxsize': 2}