introduce a pubsub system for communication

4 files changed, 43 insertions, 23 deletions

diff --git a/pubsub.py b/pubsub.py
new file mode 100644
index 0000000..6b860e0
--- /dev/null
+++ b/pubsub.py
@@ -0,0 +1,17 @@
+class Publisher(object):
+    def __init__(self):
+        self.subscribers = []
+
+    def subscribe(self, who):
+        if who not in self.subscribers:
+            self.subscribers.append(who)
+
+    def unsubscribe(self, who):
+        if who in self.subscribers:
+            self.subscribers.remove(who)
+
+    def publish(self, message, *args):
+        for subscriber in self.subscribers:
+            method = "msg_" + message
+            if hasattr(subscriber, method):
+                getattr(subscriber, method)(*args)
diff --git a/server.py b/server.py
index 3df5c92..8eb5a7e 100644
--- a/server.py
+++ b/server.py
@@ -2,13 +2,13 @@ import socket
 import threading
 import uuid
 
+import pubsub
 import ssh
 import termcast
 
 class Server(object):
     def __init__(self):
-        self.termcast_connections = {}
-        self.ssh_connections = {}
+        self.publisher = pubsub.Publisher()
 
     def listen(self):
         ssh_sock = self._open_socket(2200)
@@ -36,17 +36,13 @@ class Server(object):
     def handle_ssh_connection(self, client):
         self._handle_connection(
             client,
-            self.ssh_connections,
-            self.termcast_connections,
-            lambda client, connection_id: ssh.Connection(client, connection_id)
+            lambda client, connection_id: ssh.Connection(client, connection_id, self.publisher)
         )
 
     def handle_termcast_connection(self, client):
         self._handle_connection(
             client,
-            self.termcast_connections,
-            self.ssh_connections,
-            lambda client, connection_id: termcast.Connection(client, connection_id)
+            lambda client, connection_id: termcast.Connection(client, connection_id, self.publisher)
         )
 
     def _wait_for_connection(self, sock, cb):
@@ -61,12 +57,12 @@ class Server(object):
 
             threading.Thread(target=cb, args=(client,)).start()
 
-    def _handle_connection(self, client, connection_store, other_store, cb):
+    def _handle_connection(self, client, cb):
         connection_id = uuid.uuid4().hex
         connection = cb(client, connection_id)
-        connection_store[connection_id] = connection
-        connection.run(other_store)
-        del connection_store[connection_id]
+        self.publisher.subscribe(connection)
+        connection.run()
+        self.publisher.unsubscribe(connection)
 
     def _open_socket(self, port):
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
diff --git a/ssh.py b/ssh.py
index 7beb1f1..52c9056 100644
--- a/ssh.py
+++ b/ssh.py
@@ -11,20 +11,19 @@ class Handler(object):
         pass
 
 class Connection(object):
-    def __init__(self, client, connection_id):
+    def __init__(self, client, connection_id, publisher):
         self.transport = paramiko.Transport(client)
         self.transport.add_server_key(paramiko.RSAKey(filename='test_rsa.key'))
+        self.connection_id = connection_id
+        self.publisher = publisher
 
-    def run(self, termcast_connections):
+    def run(self):
         self.transport.start_server(server=Server())
         chan = self.transport.accept(None)
 
-        if len(termcast_connections) > 0:
-            connection = termcast_connections.values().__iter__().__next__()
-            term_contents = connection.handler.get_term()
-            chan.send(term_contents.replace("\n", "\r\n"))
-        else:
-            chan.send("no data for doy\r\n")
+        # XXX need to have the user select a stream, and then pass the stream's
+        # id in here
+        self.publisher.publish("new_viewer", chan, "some-random-id")
 
         time.sleep(5)
         chan.close()
@@ -46,4 +45,3 @@ class Server(paramiko.ServerInterface):
 
     def get_allowed_auths(self, username):
         return "password"
-
diff --git a/termcast.py b/termcast.py
index aea056c..c701f6a 100644
--- a/termcast.py
+++ b/termcast.py
@@ -24,10 +24,12 @@ class Handler(object):
         return term[:-1]
 
 class Connection(object):
-    def __init__(self, client, connection_id):
+    def __init__(self, client, connection_id, publisher):
         self.client = client
+        self.connection_id = connection_id
+        self.publisher = publisher
 
-    def run(self, ssh_connections):
+    def run(self):
         buf = b''
         while len(buf) < 1024 and b"\n" not in buf:
             buf += self.client.recv(1024)
@@ -69,3 +71,10 @@ class Connection(object):
                 self.handler.process(buf)
             else:
                 return
+
+    def msg_new_viewer(self, sock, connection_id):
+        # XXX restore this once we start passing in meaningful connection ids
+        # if connection_id != self.connection_id:
+        #     return
+        term_contents = self.handler.get_term()
+        sock.send(term_contents.replace("\n", "\r\n"))