Merge 2b6389b into b2138da

CHANGES.rst

@@ -63,6 +63,8 @@ CHANGES
 
 - Link header for 451 status code is mandatory
 
+- Implement lingering on server-side trasport closing #1050
+
 
 0.22.5 (08-02-2016)
 -------------------

CONTRIBUTORS.txt

@@ -70,6 +70,7 @@ Marco Paolini
 Mariano Anaya
 Martin Richard
 Mathias Fröjdman
+Mathieu Sornay
 Matthieu Hauglustaine
 Michael Ihnatenko
 Mikhail Lukyanchenko

aiohttp/server.py

@@ -4,12 +4,13 @@
 import http.server
 import socket
 import traceback
+from contextlib import suppress
 from html import escape as html_escape
 from math import ceil
 
 import aiohttp
 from aiohttp import errors, hdrs, helpers, streams
-from aiohttp.helpers import ensure_future
+from aiohttp.helpers import Timeout, ensure_future
 from aiohttp.log import access_logger, server_logger
 
 __all__ = ('ServerHttpProtocol',)
@@ -57,6 +58,14 @@ class ServerHttpProtocol(aiohttp.StreamProtocol):
 
     :param int timeout: slow request timeout
 
+    :param float lingering_time: maximum time during which the server
+        reads and ignore additionnal data comming from the client when
+        lingering close is on Use 0 for disabling lintering on server
+        channel closing.
+
+    :param float lingering_timeout: maximum waiting time for more
+        client data to arrive when lingering close is in effect
+
     :param allowed_methods: (optional) List of allowed request methods.
                             Set to empty list to allow all methods.
     :type allowed_methods: tuple
@@ -78,6 +87,7 @@ class ServerHttpProtocol(aiohttp.StreamProtocol):
     :param int max_field_size: Optional maximum header field size
 
     :param int max_headers: Optional maximum header size
+
     """
     _request_count = 0
     _request_handler = None
@@ -90,6 +100,8 @@ def __init__(self, *, loop=None,
                  keep_alive=75,  # NGINX default value is 75 secs
                  keep_alive_on=True,
                  timeout=0,
+                 lingering_time=30,    # NGINX default value is 30 secs
+                 lingering_timeout=5,  # NGINX default value is 5 secs
                  logger=server_logger,
                  access_log=access_logger,
                  access_log_format=helpers.AccessLogger.LOG_FORMAT,
@@ -106,6 +118,8 @@ def __init__(self, *, loop=None,
         self._keep_alive_on = keep_alive_on
         self._keep_alive_period = keep_alive  # number of seconds to keep alive
         self._timeout = timeout  # slow request timeout
+        self._lingering_time = float(lingering_time)
+        self._lingering_timeout = float(lingering_timeout)
         self._loop = loop if loop is not None else asyncio.get_event_loop()
 
         self._request_prefix = aiohttp.HttpPrefixParser()
@@ -307,7 +321,26 @@ def start(self):
                 if payload and not payload.is_eof():
                     self.log_debug('Uncompleted request.')
                     self._request_handler = None
-                    self.transport.close()
+
+                    if self._lingering_time:
+                        self.transport.write_eof()
+                        self.log_debug(
+                            'Start lingering close timer for %s sec.',
+                            self._lingering_time)
+
+                        end_time = self._loop.time() + self._lingering_time
+
+                        with suppress(asyncio.TimeoutError,
+                                      errors.ClientDisconnectedError):
+                            while self._loop.time() < end_time:
+                                with Timeout(self._lingering_timeout,
+                                             loop=self._loop):
+                                    # read and ignore
+                                    yield from payload.readany()
+
+                    if self.transport is not None:
+                        self.transport.close()
+
                     return
                 else:
                     reader.unset_parser()

tests/test_server.py

@@ -351,8 +351,17 @@ def mock_coro(*args, **kwargs):
     assert transport.close.called
 
 
-def test_handle_uncompleted(srv, loop):
+def test_handle_uncompleted(make_srv, loop):
     transport = mock.Mock()
+    closed = False
+
+    def close():
+        nonlocal closed
+        closed = True
+
+    transport.close = close
+
+    srv = make_srv(lingering_timeout=0)
     srv.connection_made(transport)
     srv.logger.exception = mock.Mock()
 
@@ -366,10 +375,31 @@ def test_handle_uncompleted(srv, loop):
 
     loop.run_until_complete(srv._request_handler)
     assert handle.called
-    assert transport.close.called
+    assert closed
     srv.logger.exception.assert_called_with("Error handling request")
 
 
+@pytest.mark.run_loop
+def test_lingering(srv, loop):
+
+    transport = mock.Mock()
+    srv.connection_made(transport)
+
+    srv.reader.feed_data(
+        b'GET / HTTP/1.0\r\n'
+        b'Host: example.com\r\n'
+        b'Content-Length: 0\r\n\r\n')
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+
+    srv.reader.feed_data(b'123')
+    srv.reader.feed_eof()
+
+    yield from asyncio.sleep(0, loop=loop)
+    transport.close.assert_called_with()
+
+
 def test_handle_coro(srv, loop):
     transport = mock.Mock()