Application cleanup

Imrpove application cleanup. This might need another pass
after #112 is merged

Issue #91 #92

Pipfile

@@ -8,10 +8,12 @@ multidict = ">=2.0"
 pyquery = "*"
 aiodns = "*"
 websockets = "*"
+async-timeout = "*"
 
 [dev-packages]
 twine = "*"
 pytest-asyncio = "*"
+"flake8" = "*"
 
 [requires]
 python_version = "3.6"

aiosip/application.py

@@ -167,11 +167,12 @@ async def reply(*args, **kwargs):
                 method=msg.method,
                 from_details=Contact.from_header(msg.headers['To']),
                 to_details=Contact.from_header(msg.headers['From']),
-                call_id=call_id
+                call_id=call_id,
+                inbound=True
             )
 
             await dialog.reply(*args, **kwargs)
-            await dialog.close(fast=True)
+            await dialog.close()
 
         try:
             route = await self.dialplan.resolve(
@@ -226,7 +227,12 @@ def finish(self):
     def register_on_finish(self, func, *args, **kwargs):
         self._finish_callbacks.insert(0, (func, args, kwargs))
 
-    async def close(self):
+    async def close(self, timeout=5):
+        for dialog in set(self._dialogs.values()):
+            try:
+                await dialog.close(timeout=timeout)
+            except asyncio.TimeoutError:
+                pass
         for connector in self._connectors.values():
             await connector.close()
         for task in self._tasks:

aiosip/dialog.py

@@ -4,6 +4,7 @@
 
 from collections import defaultdict
 from multidict import CIMultiDict
+from async_timeout import timeout as Timeout
 
 from . import utils
 from .message import Request, Response
@@ -151,10 +152,19 @@ def _maybe_close(self, msg):
 
     def _close(self):
         LOG.debug('Closing: %s', self)
+        if self._closing:
+            self._closing.cancel()
+
         for transactions in self.transactions.values():
             for transaction in transactions.values():
                 transaction.close()
 
+        # Should not be necessary once dialog are correctly tracked
+        try:
+            del self.app._dialogs[self.dialog_id]
+        except KeyError as e:
+            pass
+
     def _connection_lost(self):
         for transactions in self.transactions.values():
             for transaction in transactions.values():
@@ -187,10 +197,11 @@ def end_transaction(self, transaction):
         for item in to_delete:
             del self.transactions[item[0]][item[1]]
 
-    async def request(self, method, contact_details=None, headers=None, payload=None):
+    async def request(self, method, contact_details=None, headers=None, payload=None, timeout=None):
         msg = self._prepare_request(method, contact_details, headers, payload)
         if msg.method != 'ACK':
-            return await self.start_unreliable_transaction(msg)
+            async with Timeout(timeout):
+                return await self.start_unreliable_transaction(msg)
         else:
             self.peer.send_message(msg)
 
@@ -274,15 +285,19 @@ async def refresh(self, headers=None, expires=1800, *args, **kwargs):
             headers['Expires'] = int(expires)
         return await self.request(self.original_msg.method, headers=headers, *args, **kwargs)
 
-    async def close(self, fast=False, headers=None, *args, **kwargs):
+    async def close(self, headers=None, *args, **kwargs):
         if not self._closed:
             self._closed = True
             result = None
             if not self.inbound and self.original_msg.method in ('REGISTER', 'SUBSCRIBE'):
                 headers = CIMultiDict(headers or {})
                 if 'Expires' not in headers:
                     headers['Expires'] = 0
-                result = await self.request(self.original_msg.method, headers=headers, *args, **kwargs)
+                try:
+                    result = await self.request(self.original_msg.method, headers=headers, *args, **kwargs)
+                finally:
+                    self._close()
+
             self._close()
             return result
 
@@ -409,7 +424,7 @@ def end_transaction(self, transaction):
         for item in to_delete:
             del self.transactions[item[0]][item[1]]
 
-    async def close(self):
+    async def close(self, timeout=None):
         if not self._closed:
             self._closed = True
 
@@ -422,9 +437,11 @@ async def close(self):
             if msg:
                 transaction = UnreliableTransaction(self, original_msg=msg, loop=self.app.loop)
                 self.transactions[msg.method][msg.cseq] = transaction
-                await transaction.start()
 
-        self._close()
+                try:
+                    async with Timeout(timeout):
+                        await transaction.start()
+                finally:
+                    self._close()
 
-    def _close(self):
-        pass
+        self._close()

setup.py

@@ -14,7 +14,8 @@
    'multidict>=2.0',
    'pyquery',
    'aiodns',
-   'websockets'
+   'websockets',
+   'async_timeout'
 ]
 
 test_requirements = [

tests/conftest.py

@@ -1,8 +1,7 @@
-import asyncio
-
 import aiosip
 import pytest
-
+import asyncio
+import itertools
 
 pytest_plugins = ['aiosip.pytest_plugin']
 
@@ -119,3 +118,8 @@ def to_details(request):
 @pytest.fixture
 def loop(event_loop):
     return event_loop
+
+
+@pytest.fixture(params=itertools.permutations(('client', 'server')))
+def close_order(request):
+    return request.param

tests/test_sip_proxy.py

@@ -1,8 +1,11 @@
-import asyncio
 import aiosip
+import pytest
+import asyncio
+import itertools
 
 
-async def test_proxy_subscribe(test_server, test_proxy, protocol, loop, from_details, to_details):
+@pytest.mark.parametrize('close_order', itertools.permutations(('client', 'server', 'proxy')))  # noQa C901: too complex
+async def test_proxy_subscribe(test_server, test_proxy, protocol, loop, from_details, to_details, close_order):
     callback_complete = loop.create_future()
     callback_complete_proxy = loop.create_future()
 
@@ -59,12 +62,19 @@ async def proxy_subscribe(self, request, message):
     assert received_request_server.payload == received_request_proxy.payload
     assert received_request_server.headers == received_request_proxy.headers
 
-    await server_app.close()
-    await proxy_app.close()
-    await app.close()
+    for item in close_order:
+        if item == 'client':
+            await app.close()
+        elif item == 'server':
+            await server_app.close()
+        elif item == 'proxy':
+            await proxy_app.close()
+        else:
+            raise ValueError('Invalid close_order')
 
 
-async def test_proxy_notify(test_server, test_proxy, protocol, loop, from_details, to_details):  # noQa: C901
+@pytest.mark.parametrize('close_order', itertools.permutations(('client', 'server', 'proxy')))  # noQa C901: too complex
+async def test_proxy_notify(test_server, test_proxy, protocol, loop, from_details, to_details, close_order):
     callback_complete = loop.create_future()
     callback_complete_proxy = loop.create_future()
 
@@ -140,6 +150,12 @@ async def proxy_subscribe(self, request, message):
     assert received_notify_server.payload == received_notify_proxy.payload
     assert received_notify_server.headers == received_notify_proxy.headers
 
-    await server_app.close()
-    await proxy_app.close()
-    await app.close()
+    for item in close_order:
+        if item == 'client':
+            await app.close()
+        elif item == 'server':
+            await server_app.close()
+        elif item == 'proxy':
+            await proxy_app.close()
+        else:
+            raise ValueError('Invalid close_order')