Changeset 10

Timestamp:

01/15/09 13:56:19 (20 months ago)

Author:

Denis Bilenko <denis@ag-projects.com>

Hashname:

20090115125619-b2873-fe426e55ab44d12f778abe0d8e0e4fd2a991dfd0

Message:

added Reception class and rudimentary room support

Location:

sipchatserver

Files:

• chatroom.py (modified) (5 diffs)
• server.py (modified) (2 diffs)

sipchatserver/chatroom.py

@@ -49 +49 @@
 class ChatRoom:
 
-    def __init__(self, traffic_logger):
+    def __init__(self, room_id, reception, traffic_logger):
+        self.room_id = room_id
+        self.reception = reception
         self.traffic_logger = traffic_logger
         self.sessions = []
         self.accept_incoming_job = None
         self.incoming_queue = queue()
-        self.message_dispatcher_job = proc.spawn_link(self._message_dispatcher)
+        self.message_dispatcher_job = proc.spawn_link_exception(self._message_dispatcher)
 
     def _message_dispatcher(self):
         """Read from self.incoming_queue and dispatch the messages to other participants"""
-        while True:
-            session, message = self.incoming_queue.wait()
-            self._dispatch_message(session, message)
+        try:
+            while True:
+                session, message = self.incoming_queue.wait()
+                self._dispatch_message(session, message)
+        except proc.ProcExit, ex:
+            return ex
 
     def _dispatch_message(self, session, message):
@@ -71 +76 @@
                     self.remove_session(session)
                 except:
+                    self.remove_session(session)
                     import traceback
                     traceback.print_exc()
@@ -81 +87 @@
         self.sessions = []
 
-    def start_accept_incoming(self, *args, **kwargs):
-        assert not self.accept_incoming_job, self.accept_incoming_job
-        self.accept_incoming_job = proc.spawn(self._accept_incoming, *args, **kwargs)
-
-    def stop_accept_incoming(self):
-        if self.accept_incoming_job:
-            self.accept_incoming_job.kill()
-            self.accept_incoming_job = None
-
-    def add_session(self, session, msrp):
-        proc.spawn(self._forwarder, msrp, session)
+    def add_session(self, session):
+        proc.spawn(self._forwarder, session)
         self.sessions.append(session)
 
     def remove_session(self, session):
         try:
-            #session.end()
+            session.end()
             self.sessions.remove(session)
         except ValueError:
             pass
+        if not self.sessions:
+            self.reception.remove_room(self)
 
-    def _forwarder(self, msrp, session):
+    def _forwarder(self, session):
         while True:
             try:
-                chunk = msrp.receive_chunk()
+                chunk = session.msrp.receive_chunk()
             except ConnectionClosed:
                 self.remove_session(session)
@@ -114 +113 @@
         def new_session(sip, msrp):
             session = MSRPSession(sip, msrp)
-            self.add_session(session, msrp)
+            self.add_session(session)
         acceptor = MSRPAcceptFactory.new(relay, self.traffic_logger)
         handler1 = JoinHandler(acceptor, new_session)
@@ -125 +124 @@
                 print ex
                 sleep(1)
+
+
+class Reception:
+
+    def __init__(self, traffic_logger):
+        self.traffic_logger = traffic_logger
+        self.rooms = {} # maps room_id -> ChatRoom
+
+    def add_session(self, session):
+        room_id = '%s@%s' % (session.local_uri.user, session.local_uri.host)
+        room = self.rooms.get(room_id)
+        if room is None:
+            print 'Creating a room %s' % room_id
+            room = ChatRoom(room_id, self, self.traffic_logger)
+            self.rooms[room_id] = room
+        room.add_session(session)
+
+    def remove_room(self, room):
+        self.rooms.pop(room.room_id, None)
+
+    def accept_incoming_loop(self, e, relay=None, local_uri=None):
+        def new_session(sip, msrp):
+            session = MSRPSession(sip, msrp)
+            self.add_session(session)
+        acceptor = MSRPAcceptFactory.new(relay, self.traffic_logger)
+        handler1 = JoinHandler(acceptor, new_session)
+        handler = IncomingSessionHandler()
+        handler.add_handler(handler1)
+        while True:
+            try:
+                handler.wait_and_handle(e, local_uri)
+            except MSRPSessionErrors, ex:
+                print ex
+                sleep(1)
+            except proc.ProcExit:
+                return
+

sipchatserver/server.py

@@ -10 +10 @@
 
 import sipchatserver
-from sipchatserver.chatroom import ChatRoom
+from sipchatserver.chatroom import Reception
 from sipchatserver.tls import Certificate, PrivateKey
 
@@ -59 +59 @@
         reactor.addSystemEventTrigger('before', 'shutdown', shutdown)
         logger = TrafficLogger.to_file(is_enabled_func = lambda: options.trace_msrp)
-        room = ChatRoom(logger)
-        room.start_accept_incoming(e, options.relay, server_uri)
+        reception = Reception(logger)
+        proc.spawn_link_exception(reception.accept_incoming_loop, e, options.relay, server_uri)
 
     def run_reactor(self):