Merge branch 'release/2.2.1'

.travis.yml

@@ -1,9 +1,10 @@
 language: python
 python:
-  - "3.5"
   - "3.6"
+  - "3.7"
+  - "3.8"
 install:
-  - pip install twisted autobahn coveralls
+  - pip install twisted autobahn coveralls asgiref
 script:
   - nosetests --with-coverage --cover-package=txasgiresource
 after_success:

CHANGELOG.rst

@@ -1,6 +1,13 @@
 Changelog
 =========
 
+Version 2.2.1 (07-05-2020)
+-----------------------------------------------------------
+
+*   Fixed memoryleak regarding websocket timeout and
+    lack of application cleanup
+*   Fixed bug related to shown exception on sendfile
+
 Version 2.2.0 (17-02-2020)
 -----------------------------------------------------------
 

setup.py

@@ -25,16 +25,17 @@
         'asgiref>=2.3.2'
     ],
     classifiers=[
-        'Development Status :: 3 - Alpha',
+        'Development Status :: 4 - Beta',
         'Environment :: Web Environment',
         'Framework :: Twisted',
         'Intended Audience :: Developers',
         'License :: OSI Approved :: MIT License',
         'Operating System :: OS Independent',
         'Programming Language :: Python',
         'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.5',
         'Programming Language :: Python :: 3.6',
+        'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
         'Topic :: Internet :: WWW/HTTP',
         'Topic :: Software Development :: Libraries :: Python Modules',
     ],

txasgiresource/__init__.py

@@ -8,4 +8,4 @@
     asyncioreactor.install()
 
 
-__version__ = "2.2.0"
+__version__ = "2.2.1"

txasgiresource/application.py

@@ -12,15 +12,13 @@ def __init__(self, application):
     @defer.inlineCallbacks
     def stop(self):
         wait_for = []
-        for protocol, application_instance in list(self.application_instances.items()):
-            if protocol.do_cleanup():
-                wait_for.append(application_instance)
+        for protocol in list(self.application_instances.keys()):
+            promise = protocol.do_cleanup()
+            if promise:
+                wait_for.append(promise)
 
         for d in wait_for:
-            try:
-                yield defer.Deferred.fromFuture(d)
-            except CancelledError:
-                pass
+            yield defer.Deferred.fromFuture(d)
 
     def create_application_instance(self, protocol, scope):
         async def handle_reply(msg):
@@ -35,10 +33,10 @@ async def handle_reply(msg):
         return queue
 
     def finish_protocol(self, protocol):
-        wait_for = False
+        wait_for = None
         if protocol in self.application_instances:
             if not self.application_instances[protocol].done():
-                if self.application_instances[protocol].cancel():
+                if not self.application_instances[protocol].cancelled():
 
                     def handle_cancel_exception(f):
                         try:
@@ -49,6 +47,7 @@ def handle_cancel_exception(f):
                     self.application_instances[protocol].add_done_callback(
                         handle_cancel_exception
                     )
-                wait_for = True
+                    self.application_instances[protocol].cancel()
+                wait_for = self.application_instances[protocol]
             del self.application_instances[protocol]
         return wait_for

txasgiresource/http.py

@@ -2,7 +2,7 @@
 import logging
 import os
 
-from twisted.internet import defer, reactor
+from twisted.internet import defer, error, reactor
 from twisted.web import http, resource, server, static
 
 from .utils import send_error_page
@@ -160,7 +160,10 @@ def do_sendfile(self, request, path):
         if request.setETag(etag) != http.CACHED:
             finished_defer = request.notifyFinish()
             static.File(path).render(request)
-            yield finished_defer
+            try:
+                yield finished_defer
+            except error.ConnectionDone:
+                logger.debug("sendfile done")
 
     def do_cleanup(self, is_finished=False):
         logger.debug(

txasgiresource/ws.py

@@ -18,7 +18,6 @@ class ASGIWebSocketServerProtocol(WebSocketServerProtocol, policies.TimeoutMixin
     accept_promise = None
     queue = None
 
-    @defer.inlineCallbacks
     def _onConnect(self, request):
         scope = dict(self.factory.base_scope)
         scope["type"] = "websocket"
@@ -34,8 +33,8 @@ def _onConnect(self, request):
         scope["subprotocols"] = subprotocols
 
         try:
-            self.queue = yield defer.maybeDeferred(
-                self.factory.application.create_application_instance, self, scope
+            self.queue = self.factory.application.create_application_instance(
+                self, scope
             )
             self.opened = True
         except Exception:
@@ -64,10 +63,10 @@ def send_replies(self):
             except defer.TimeoutError:
                 logger.debug("We hit a timeout")
                 self.dropConnection(abort=True)
-                break
+                return
             except defer.CancelledError:
                 self.dropConnection(abort=True)
-                break
+                return
 
             if not self.accepted:
                 if reply["type"] == "websocket.accept":
@@ -79,7 +78,7 @@ def send_replies(self):
                         ConnectionDeny(code=403, reason="Denied")
                     )
                     self.dropConnection(abort=True)
-                    break
+                    return
                 else:
                     continue
 
@@ -108,12 +107,12 @@ def onMessage(self, payload, isBinary):
             )
 
     def onClose(self, wasClean, code, reason):
-        if not self.opened:
-            return
+        if self.opened:
+            logger.info("Called onClose")
 
-        logger.info("Called onClose")
+            self.queue.put_nowait({"type": "websocket.disconnect", "code": code})
 
-        self.queue.put_nowait({"type": "websocket.disconnect", "code": code})
+        self.do_cleanup()
 
     def timeoutConnection(self):
         logger.debug("Timeout from mixin")
@@ -125,6 +124,7 @@ def handle_reply(self, msg):
         d.callback(msg)
 
     def do_cleanup(self):
+        self.setTimeout(None)
         return self.factory.application.finish_protocol(self)