optimize keepalive, clean up

Author: nggit

tremolo/lib/http_protocol.py

@@ -296,7 +296,7 @@ async def _receive_data(self):
             elif self.request.body_size >= self.request.content_length > -1:
                 self.queue[0].put_nowait(None)
             elif self.request.body_size < self.options['client_max_body_size']:
-                if self.request.has_body:
+                if self.request.has_body and not self.request.eof():
                     self.transport.resume_reading()
             else:
                 self.close(BadRequest('payload too large'))
@@ -412,17 +412,17 @@ async def _send_data(self):
             self.close(exc)
 
     async def _handle_keepalive(self):
-        if 'receive' in self.events:
-            # waits for all incoming data to enter the queue
-            await self.events['receive']
-
         if self.request.has_body:
             if not self.request.eof():
                 self.logger.info('request body was not fully consumed')
                 self.request.clear()
                 self.close()
                 return
 
+        if 'receive' in self.events:
+            # waits for all incoming data to enter the queue
+            await self.events['receive']
+
         self.events['request'] = self.loop.create_future()
 
         self.add_task(self.app.create_task(

tremolo/lib/http_request.py

@@ -170,15 +170,12 @@ async def recv(self, size=-1, *, timeout=None, raw=True):
         if size == -1:
             return await self._stream.__anext__()
 
-        if len(self._read_buf) < size:
-            async for data in self._stream:
-                self._read_buf.extend(data)
-
-                if len(self._read_buf) >= size:
-                    break
+        while len(self._read_buf) < size:
+            self._read_buf.extend(await self._stream.__anext__())
 
         data = bytes(self._read_buf[:size])
         del self._read_buf[:size]
+
         return data
 
     async def stream(self, timeout=None, raw=False):
@@ -201,10 +198,11 @@ async def stream(self, timeout=None, raw=False):
 
                 if bytes_unread > 2:
                     data = bytes(buf[:bytes_unread - 2])
-                    yield data
-
                     del buf[:bytes_unread - 2]
-                    bytes_unread -= len(data)
+
+                    if data:
+                        yield data
+                        bytes_unread -= len(data)
 
                     paused = False
                     continue
@@ -245,11 +243,12 @@ async def stream(self, timeout=None, raw=False):
                     raise BadRequest('bad chunked encoding') from exc
 
                 data = bytes(buf[i + 2:i + 2 + chunk_size])
-                bytes_unread = chunk_size - len(data) + 2
-
                 del buf[:i + 2 + chunk_size]
 
-                yield data
+                if data:
+                    yield data
+
+                bytes_unread = chunk_size - len(data) + 2
                 paused = True
         else:
             if (self.content_length >

tremolo/lib/request.py

@@ -56,7 +56,7 @@ async def recv(self, timeout=None):
             except asyncio.CancelledError as exc:
                 raise TimeoutError('recv timeout') from exc
 
-            if data is None:
+            if data is None:  # only occurs on a request with Content-Length
                 break
 
             yield data