Implement lingered closing

aiohttp/server.py

@@ -94,6 +94,8 @@ def __init__(self, *, loop=None,
                  max_line_size=8190,
                  max_headers=32768,
                  max_field_size=8190,
+                 lingering_time=30,
+                 lingering_timeout=5,
                  **kwargs):
 
         # process deprecated params
@@ -116,6 +118,8 @@ def __init__(self, *, loop=None,
         self._tcp_keepalive = tcp_keepalive
         self._keepalive_timeout = keepalive_timeout
         self._slow_request_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_parser = aiohttp.HttpRequestParser(
@@ -256,9 +260,25 @@ def start(self):
 
                 yield from self.handle_request(message, payload)
 
-                if payload and not payload.is_eof():
+                if not payload.is_eof():
                     self.log_debug('Uncompleted request.')
                     self._closing = True
+
+                    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()
                 else:
                     reader.unset_parser()
                     if not self._keepalive or not self._keepalive_timeout:

docs/web_reference.rst

@@ -1320,6 +1320,14 @@ duplicated like one using :meth:`Application.copy`.
     :param int max_field_size: Optional maximum header field size. Default:
       ``8190``.
 
+    :param float lingering_time: maximum time during which the server
+       reads and ignore additional data coming from the client when
+       lingering close is on.  Use ``0`` for disabling lingering on
+       server channel closing.
+
+    :param float lingering_timeout: maximum waiting time for more
+        client data to arrive when lingering close is in effect
+
 
     You should pass result of the method as *protocol_factory* to
     :meth:`~asyncio.AbstractEventLoop.create_server`, e.g.::

tests/test_server.py

@@ -329,8 +329,18 @@ 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()
 
@@ -344,10 +354,84 @@ 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")
 
 
+@asyncio.coroutine
+def test_lingering(srv, loop):
+
+    transport = mock.Mock()
+    srv.connection_made(transport)
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+
+    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')
+
+    srv.reader.feed_data(b'123')
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+    srv.reader.feed_eof()
+
+    yield from asyncio.sleep(0, loop=loop)
+    transport.close.assert_called_with()
+
+
+@asyncio.coroutine
+def test_lingering_disabled(make_srv, loop):
+    srv = make_srv(lingering_time=0)
+
+    transport = mock.Mock()
+    srv.connection_made(transport)
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+
+    srv.reader.feed_data(
+        b'GET / HTTP/1.0\r\n'
+        b'Host: example.com\r\n'
+        b'Content-Length: 50\r\n\r\n')
+
+    srv.reader.feed_data(b'123')
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+    srv.reader.feed_eof()
+
+    yield from asyncio.sleep(0, loop=loop)
+    transport.close.assert_called_with()
+
+
+@asyncio.coroutine
+def test_lingering_zero_timeout(make_srv, loop):
+    srv = make_srv(lingering_time=1e-30)
+
+    transport = mock.Mock()
+    srv.connection_made(transport)
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+
+    srv.reader.feed_data(
+        b'GET / HTTP/1.0\r\n'
+        b'Host: example.com\r\n'
+        b'Content-Length: 50\r\n\r\n')
+
+    srv.reader.feed_data(b'123')
+
+    yield from asyncio.sleep(0, loop=loop)
+    assert not transport.close.called
+    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()