abstract services from processor; define server.peers as a service
[electrum-server.git] / stratum.py
index 6643815..4a0a542 100644 (file)
@@ -8,40 +8,64 @@ class Processor(threading.Thread):
 
     def __init__(self):
         self.shared = None
-        self.lock = threading.Lock()
-        self.sessions = []
         threading.Thread.__init__(self)
         self.daemon = True
+        self.request_queue = queue.Queue()
+        self.response_queue = queue.Queue()
+        self.internal_ids = {}
+        self.internal_id = 0
+        self.lock = threading.Lock()
 
-    def add_session(self, session):
+    def push_response(self, item):
+        self.response_queue.put(item)
+
+    def pop_response(self):
+        return self.response_queue.get()
+
+    def push_request(self, session, item):
+        self.request_queue.put((session,item))
+
+    def pop_request(self):
+        return self.request_queue.get()
+
+    def get_session_id(self, internal_id):
         with self.lock:
-            self.sessions.append(session)
+            return session_ids.pop(internal_id)
+
+    def store_session_id(self, session, msgid):
+        with self.lock:
+            self.internal_ids[self.internal_id] = session, msgid
+            self.internal_id += 1
 
     def run(self):
         if self.shared is None:
             raise TypeError("self.shared not set in Processor")
         while not self.shared.stopped():
-            # Deep copy entire sessions list and blank it
-            # This is done to minimise lock contention
-            with self.lock:
-                sessions = self.sessions[:]
-                self.sessions = []
-            for session in sessions:
-                if not session.stopped():
-                    # If session is still alive then re-add it back
-                    # to our internal register
-                    self.add_session(session)
-                    self.process(session)
+            session, request = self.pop_request()
+
+            method = request['method']
+            params = request.get('params',[])
+
+            if method in [ 'numblocks.subscribe', 'address.subscribe', 'server.peers']:
+                session.subscribe_to_service(method, params)
+
+            # store session and id locally
+            request['id'] = self.store_session_id(session, request['id'])
+            self.process(request)
+
         self.stop()
 
     def stop(self):
         pass
 
-    def process(self, session):
-        request = session.pop_request()
+    def process(self, request):
         print "New request", request
-        # Execute and when ready, you call
-        # session.push_response(response)
+        # Do stuff...
+        # response = request
+        # When ready, you call
+        # self.push_response(response)
+
+
 
 class Session:
 
@@ -50,9 +74,8 @@ class Session:
         self.address = address
         self._stopped = False
         self.lock = threading.Lock()
-
-        self.request_queue = queue.Queue()
-        self.response_queue = queue.Queue()
+        self.subscriptions = []
+        print "new session", address
 
     def stop(self):
         self._connection.close()
@@ -70,42 +93,63 @@ class Session:
         else:
             return self._connection
 
-    def push_request(self, item):
-        self.request_queue.put(item)
-
-    def pop_request(self):
-        return self.request_queue.get()
-
-    def push_response(self, item):
-        self.response_queue.put(item)
+    def subscribe_to_service(self, method, params):
+        with self.lock:
+            self.subscriptions.append((method, params))
+    
 
-    def pop_response(self):
-        return self.response_queue.get()
 
-class TcpClientResponder(threading.Thread):
+class TcpResponder(threading.Thread):
 
-    def __init__(self, shared, session):
+    def __init__(self, shared, processor, server):
         self.shared = shared
-        self.session = session
+        self.processor = processor
+        self.server = server
         threading.Thread.__init__(self)
 
+
     def run(self):
-        while not self.shared.stopped() or self.session.stopped():
-            response = self.session.pop_response()
-            raw_response = json.dumps(response)
-            # Possible race condition here by having session
-            # close connection?
-            # I assume Python connections are thread safe interfaces
-            connection = self.session.connection()
+        while not self.shared.stopped():
+            response = self.processor.pop_response()
+
+            internal_id = response.get('id')
+            params = response.get('params',[])
             try:
-                connection.send(raw_response + "\n")
+                method = response['method']
             except:
-                self.session.stop()
+                print "no method", response
+                continue
+
+            if internal_id:
+                session, message_id = self.processor.get_session_id(internal_id)
+                if message_id:
+                    response['id'] = message_id
+                self.send_response(response, session)
+
+            else:
+                for session in self.server.sessions:
+                    if not session.stopped():
+                        if (method,params) in session.subscriptions:
+                            self.send_response(response, session)
+
+
+
+    def send_response(self, response, session):
+        raw_response = json.dumps(response)
+        # Possible race condition here by having session
+        # close connection?
+        # I assume Python connections are thread safe interfaces
+        try:
+            connection = session.connection()
+            connection.send(raw_response + "\n")
+        except:
+            session.stop()
 
 class TcpClientRequestor(threading.Thread):
 
-    def __init__(self, shared, session):
+    def __init__(self, shared, processor, session):
         self.shared = shared
+        self.processor = processor
         self.message = ""
         self.session = session
         threading.Thread.__init__(self)
@@ -113,42 +157,42 @@ class TcpClientRequestor(threading.Thread):
     def run(self):
         while not self.shared.stopped():
             if not self.update():
-                self.session.stop()
-                return
+                break
+
+            while self.parse():
+                pass
 
     def update(self):
         data = self.receive()
-        if data is None:
+        if not data:
             # close_session
-            self.stop()
+            self.session.stop()
             return False
 
         self.message += data
-        if not self.parse():
-            return False
         return True
 
     def receive(self):
         try:
             return self.session.connection().recv(1024)
         except socket.error:
-            return None
+            return ''
 
     def parse(self):
         raw_buffer = self.message.find('\n')
         if raw_buffer == -1:
-            return True
+            return False
 
         raw_command = self.message[0:raw_buffer].strip()
         self.message = self.message[raw_buffer + 1:]
         if raw_command == 'quit': 
+            self.session.stop()
             return False
 
         try:
             command = json.loads(raw_command)
         except:
-            self.session.push_response(
-                {"error": "bad JSON", "request": raw_command})
+            self.processor.push_response({"error": "bad JSON", "request": raw_command})
             return True
 
         try:
@@ -158,35 +202,55 @@ class TcpClientRequestor(threading.Thread):
             method = command['method']
         except KeyError:
             # Return an error JSON in response.
-            self.session.push_response(
-                {"error": "syntax error", "request": raw_command})
+            self.processor.push_response({"error": "syntax error", "request": raw_command})
         else:
-            self.session.push_request(command)
+            self.processor.push_request(self.session,command)
 
         return True
 
 class TcpServer(threading.Thread):
 
-    def __init__(self, shared, processor):
+    def __init__(self, shared, processor, host, port):
         self.shared = shared
         self.processor = processor
-        self.clients = []
+        self.sessions = []
         threading.Thread.__init__(self)
         self.daemon = True
+        self.host = host
+        self.port = port
+        self.lock = threading.Lock()
 
     def run(self):
         print "TCP server started."
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        sock.bind(("176.31.24.241", 50001))
+        sock.bind((self.host, self.port))
         sock.listen(1)
+        responder = TcpResponder(self.shared, self.processor, self)
+        responder.start()
         while not self.shared.stopped():
             session = Session(*sock.accept())
-            client_req = TcpClientRequestor(self.shared, session)
+            client_req = TcpClientRequestor(self.shared, self.processor, session)
             client_req.start()
-            client_res = TcpClientResponder(self.shared, session)
-            client_res.start()
-            self.processor.add_session(session)
+            self.add_session(session)
+            self.collect_garbage()
+
+    def add_session(self, session):
+        with self.lock:
+            self.sessions.append(session)
+
+    def collect_garbage(self):
+        # Deep copy entire sessions list and blank it
+        # This is done to minimise lock contention
+        with self.lock:
+            sessions = self.sessions[:]
+            self.sessions = []
+        for session in sessions:
+            if not session.stopped():
+                # If session is still alive then re-add it back
+                # to our internal register
+                self.add_session(session)
+
 
 class Shared:
 
@@ -211,7 +275,7 @@ class Stratum:
         processor.shared = shared
         processor.start()
         # Create various transports we need
-        transports = TcpServer(shared, processor),
+        transports = TcpServer(shared, processor, "176.31.24.241", 50001),
         for server in transports:
             server.start()
         while not shared.stopped():