Merge pull request #90 from sangoma/bug-fixes

Slew of bug fixes

aiosip/application.py

@@ -110,7 +110,8 @@ def _create_dialog(self):
                         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
                     )
                 return self.dialog
 
@@ -137,6 +138,10 @@ async def _dispatch(self, protocol, msg, addr):
             await dialog.receive_message(msg)
             return
 
+        # If we got an ACK, but nowhere to deliver it, drop it
+        if msg.method == 'ACK':
+            return
+
         router = await self.dialplan.resolve(
             username=msg.from_details['uri']['user'],
             protocol=peer.protocol,

aiosip/dialog.py

@@ -32,8 +32,11 @@ def __init__(self,
                  peer,
                  contact_details,
                  *,
+                 headers=None,
+                 payload=None,
                  password=None,
-                 cseq=0):
+                 cseq=0,
+                 inbound=False):
 
         self.app = app
         self.from_details = from_details
@@ -43,12 +46,17 @@ def __init__(self,
         self.peer = peer
         self.password = password
         self.cseq = cseq
+        self.inbound = inbound
         self.transactions = defaultdict(dict)
 
         # TODO: Needs to be last because we need the above attributes set
-        self.original_msg = self._prepare_request(method)
+        self.original_msg = self._prepare_request(method, headers=headers, payload=payload)
+
+        self._closed = False
+        self._closing = None
 
     def _receive_response(self, msg):
+
         try:
             transaction = self.transactions[msg.method][msg.cseq]
             transaction._incoming(msg)
@@ -87,10 +95,17 @@ def _prepare_request(self, method, contact_details=None, headers=None, payload=N
 
     async def start(self, *, expires=None):
         # TODO: this is a hack
-        headers = {}
-        if expires:
+        headers = self.original_msg.headers
+        if expires is not None:
             headers['Expires'] = expires
-        return await self.request(self.original_msg.method, headers=headers)
+        return await self.request(self.original_msg.method, headers=headers, payload=self.original_msg.payload)
+
+    def ack(self, msg, headers=None, *args, **kwargs):
+        headers = CIMultiDict(headers or {})
+
+        headers['Via'] = msg.headers['Via']
+        ack = self._prepare_request('ACK', cseq=msg.cseq, to_details=msg.to_details, headers=headers, *args, **kwargs)
+        self.peer.send_message(ack)
 
     async def unauthorized(self, msg):
         self._nonce = utils.gen_str(10)
@@ -107,10 +122,6 @@ def validate_auth(self, msg, password):
         else:
             return False
 
-    def close(self, *, fast=False):
-        self.peer._close_dialog(self.call_id)
-        self._close()
-
     def close_later(self, delay=None):
         if delay is None:
             delay = self.app.defaults['dialog_closing_delay']
@@ -124,7 +135,7 @@ async def closure():
         self._closing = asyncio.ensure_future(closure())
 
     def _maybe_close(self, msg):
-        if msg.method in ('REGISTER', 'SUBSCRIBE'):
+        if msg.method in ('REGISTER', 'SUBSCRIBE') and not self.inbound:
             expire = int(msg.headers.get('Expires', 0))
             delay = int(expire * 1.1) if expire else None
             self.close_later(delay)
@@ -231,7 +242,6 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
         self._nonce = None
-        self._closing = None
         self._incoming = asyncio.Queue()
 
     async def receive_message(self, msg):
@@ -262,10 +272,17 @@ async def refresh(self, headers=None, expires=1800, *args, **kwargs):
         return await self.request(self.original_msg.method, headers=headers, *args, **kwargs)
 
     async def close(self, fast=False, headers=None, *args, **kwargs):
-        headers = CIMultiDict(headers or {})
-        if 'Expires' not in headers:
-            headers['Expires'] = 0
-        return await self.request(self.original_msg.method, headers=headers, *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)
+            self.peer._close_dialog(self.call_id)
+            self._close()
+            return result
 
     async def notify(self, *args, headers=None, **kwargs):
         headers = CIMultiDict(headers or {})
@@ -350,6 +367,17 @@ async def handle_completed_state(msg):
             else:
                 return await self._receive_request(msg)
 
+    async def _receive_request(self, msg):
+        self.peer._bookkeeping(msg, self.call_id)
+
+        if 'tag' in msg.from_details['params']:
+            self.to_details['params']['tag'] = msg.from_details['params']['tag']
+
+        if msg.method == 'BYE':
+            self._closed = True
+
+        self._maybe_close(msg)
+
     @property
     def state(self):
         return self._state
@@ -358,6 +386,9 @@ async def start(self, *, expires=None):
         # TODO: this is a hack
         self.peer.send_message(self.original_msg)
 
+    async def recv(self):
+        return await self._queue.get()
+
     async def wait_for_terminate(self):
         while not self._waiter.done():
             yield await self._queue.get()
@@ -367,13 +398,6 @@ async def ready(self):
         if msg.status_code != 200:
             raise RuntimeError("INVITE failed with {}".format(msg.status_code))
 
-    def ack(self, msg, headers=None, *args, **kwargs):
-        headers = CIMultiDict(headers or {})
-
-        headers['Via'] = msg.headers['Via']
-        ack = self._prepare_request('ACK', cseq=msg.cseq, to_details=msg.to_details, headers=headers, *args, **kwargs)
-        self.peer.send_message(ack)
-
     def end_transaction(self, transaction):
         to_delete = list()
         for method, values in self.transactions.items():
@@ -386,10 +410,22 @@ def end_transaction(self, transaction):
             del self.transactions[item[0]][item[1]]
 
     async def close(self):
-        msg = self._prepare_request('BYE')
-        transaction = UnreliableTransaction(self, original_msg=msg, loop=self.app.loop)
-        self.transactions[msg.method][msg.cseq] = transaction
-        return await transaction.start()
+        if not self._closed:
+            self._closed = True
+
+            msg = None
+            if self._state == CallState.Terminated:
+                msg = self._prepare_request('BYE')
+            elif self._state != CallState.Completed:
+                msg = self._prepare_request('CANCEL')
+
+            if msg:
+                transaction = UnreliableTransaction(self, original_msg=msg, loop=self.app.loop)
+                self.transactions[msg.method][msg.cseq] = transaction
+                await transaction.start()
+
+        self.peer._close_dialog(self.call_id)
+        self._close()
 
     def _close(self):
         pass

aiosip/peers.py

@@ -39,7 +39,7 @@ def send_message(self, msg):
         self._protocol.send_message(msg, addr=self.peer_addr)
 
     def _create_dialog(self, method, from_details, to_details, contact_details=None, password=None, call_id=None,
-                       cseq=0):
+                       headers=None, payload=None, cseq=0, inbound=False):
 
         if not call_id:
             call_id = str(uuid.uuid4())
@@ -73,12 +73,34 @@ def _create_dialog(self, method, from_details, to_details, contact_details=None,
             call_id=call_id,
             peer=self,
             password=password,
+            headers=headers,
+            payload=payload,
             cseq=cseq,
+            inbound=inbound,
         )
         LOG.debug('Creating: %s', dialog)
         self._dialogs[call_id] = dialog
         return dialog
 
+    async def request(self, method, from_details, to_details, contact_details=None, password=None, call_id=None,
+                      headers=None, payload=None, cseq=0):
+        dialog = self._create_dialog(method=method,
+                                     from_details=from_details,
+                                     to_details=to_details,
+                                     contact_details=contact_details,
+                                     headers=headers,
+                                     payload=payload,
+                                     password=password,
+                                     call_id=call_id, cseq=cseq)
+        try:
+            response = await dialog.start()
+            dialog.status_code = response.status_code
+            dialog.status_message = response.status_message
+            return dialog
+        except asyncio.CancelledError:
+            dialog.cancel()
+            raise
+
     async def subscribe(self, from_details, to_details, contact_details=None, password=None, call_id=None, cseq=0,
                         expires=3600):
         dialog = self._create_dialog(method="SUBSCRIBE",
@@ -113,7 +135,8 @@ async def register(self, from_details, to_details, contact_details=None, passwor
             dialog.cancel()
             raise
 
-    async def invite(self, from_details, to_details, contact_details=None, password=None, call_id=None, cseq=0):
+    async def invite(self, from_details, to_details, contact_details=None, password=None, call_id=None, headers=None,
+                     payload=None, cseq=0):
 
         if not call_id:
             call_id = str(uuid.uuid4())
@@ -146,6 +169,8 @@ async def invite(self, from_details, to_details, contact_details=None, password=
             call_id=call_id,
             peer=self,
             contact_details=contact_details,
+            headers=headers,
+            payload=payload,
             password=None,
             cseq=cseq
         )