p2p: correctly clean up everything on logout

5 files changed, 30 insertions, 5 deletions

diff --git a/papyon/client.py b/papyon/client.py
index 70cebb7..ba5cf29 100644
--- a/papyon/client.py
+++ b/papyon/client.py
@@ -289,6 +289,7 @@ class Client(EventsDispatcher):
             return
         self.__die = True
         self._call_manager.close()
+        self._p2p_session_manager.close()
         self._switchboard_manager.close()
         self._protocol.signoff()
         self.__state = ClientState.CLOSED
diff --git a/papyon/msnp2p/session.py b/papyon/msnp2p/session.py
index 76c2dd0..c44b39b 100644
--- a/papyon/msnp2p/session.py
+++ b/papyon/msnp2p/session.py
@@ -24,6 +24,7 @@ from papyon.msnp2p.constants import *
 from papyon.msnp2p.SLP import *
 from papyon.msnp2p.transport import *
 from papyon.util.parsing import build_account
+from papyon.util.timer import Timer
 import papyon.util.element_tree as ElementTree
 
 import gobject
@@ -41,7 +42,7 @@ MAX_INT32 = 0x7fffffff
 MAX_INT16 = 0x7fff
 
 
-class P2PSession(gobject.GObject, EventsDispatcher):
+class P2PSession(gobject.GObject, EventsDispatcher, Timer):
 
     __gsignals__ = {
             "accepted" : (gobject.SIGNAL_RUN_FIRST,
@@ -68,6 +69,7 @@ class P2PSession(gobject.GObject, EventsDispatcher):
             application_id=0, message=None):
         gobject.GObject.__init__(self)
         EventsDispatcher.__init__(self)
+        Timer.__init__(self)
         self._session_manager = session_manager
         self._transport_manager = session_manager._transport_manager
         self._client = session_manager._client
@@ -160,6 +162,7 @@ class P2PSession(gobject.GObject, EventsDispatcher):
                 call_id=self._call_id)
         message.body = body
         self._send_slp_message(message)
+        self.start_timeout("response", 60)
 
     def _transreq(self):
         self._cseq = 0
@@ -266,6 +269,7 @@ class P2PSession(gobject.GObject, EventsDispatcher):
 
     def _dispose(self):
         logger.info("Session %s disposed" % self._id)
+        self.stop_all_timeout()
         self._session_manager._transport_manager.cleanup(self.peer,
                 self.peer_guid, self._id)
         self._session_manager._unregister_session(self)
@@ -289,6 +293,7 @@ class P2PSession(gobject.GObject, EventsDispatcher):
                 print "Unhandled signaling blob :", message
         elif isinstance(message, SLPResponseMessage):
             if isinstance(message.body, SLPSessionRequestBody):
+                self.stop_timeout("response")
                 if message.status == 200:
                     self._emit("accepted")
                     self._on_session_accepted()
@@ -303,6 +308,7 @@ class P2PSession(gobject.GObject, EventsDispatcher):
         data.seek(0, os.SEEK_SET)
         self._completed = True
         self._emit("completed", data)
+        self.start_timeout("bye", 5)
 
     def _on_data_received(self, data):
         logger.info("Session data transfer completed")
@@ -314,6 +320,12 @@ class P2PSession(gobject.GObject, EventsDispatcher):
     def _on_data_transferred(self, size):
         self._emit("progressed", size)
 
+    def on_response_timeout(self):
+        self._close()
+
+    def on_bye_timeout(self):
+        self._dispose()
+
     # Methods to implement in different P2P applications
 
     def _on_invite_received(self, message):
diff --git a/papyon/msnp2p/session_manager.py b/papyon/msnp2p/session_manager.py
index b3c50fd..35f8d11 100644
--- a/papyon/msnp2p/session_manager.py
+++ b/papyon/msnp2p/session_manager.py
@@ -65,6 +65,11 @@ class P2PSessionManager(gobject.GObject):
     def register_handler(self, handler_class):
         self._handlers.append(handler_class)
 
+    def close(self):
+        for session in self._sessions.values():
+            session._close()
+        self._transport_manager.close()
+
     def _register_session(self, session):
         self._sessions[session.id] = session
         self._transport_manager.remove_from_blacklist(session.peer,
diff --git a/papyon/msnp2p/transport/base.py b/papyon/msnp2p/transport/base.py
index e55f31a..f223a8c 100644
--- a/papyon/msnp2p/transport/base.py
+++ b/papyon/msnp2p/transport/base.py
@@ -111,6 +111,8 @@ class BaseP2PTransport(gobject.GObject):
         self._queue_lock.release()
 
     def close(self):
+        self._stop_processing()
+        self._reset()
         self._transport_manager._unregister_transport(self)
 
     def _ready_to_send(self):
@@ -181,10 +183,6 @@ class BaseP2PTransport(gobject.GObject):
         if not self.has_data_to_send():
             self._queue_lock.release()
             return False
-        if not self._ready_to_send():
-            logger.info("Transport is not ready to send, bail out")
-            self._queue_lock.release()
-            return False
 
         # FIXME find a better algorithm to choose session
         if 0 in self._data_blob_queue:
@@ -192,6 +190,11 @@ class BaseP2PTransport(gobject.GObject):
         else:
             session_id = self._data_blob_queue.keys()[0]
 
+        if session_id != 0 and not self._ready_to_send():
+            logger.info("Transport is not ready to send, bail out")
+            self._queue_lock.release()
+            return False
+
         sync = self._first
         self._first = False
         (peer, peer_guid, blob) = self._data_blob_queue[session_id][0]
diff --git a/papyon/msnp2p/transport/transport_manager.py b/papyon/msnp2p/transport/transport_manager.py
index c5eb541..28b3d18 100644
--- a/papyon/msnp2p/transport/transport_manager.py
+++ b/papyon/msnp2p/transport/transport_manager.py
@@ -88,6 +88,10 @@ class P2PTransportManager(gobject.GObject):
         blob = MessageBlob(0, buffer, size, session_id)
         self._data_blobs[(peer, peer_guid, session_id)] = blob
 
+    def close(self):
+        for transport in self._transports.copy():
+            transport.close()
+
     def cleanup(self, peer, peer_guid, session_id):
         logger.info("Cleaning up session %s" % session_id)
         if (peer, peer_guid, session_id) in self._data_blobs: