change abe backend to print_log

[electrum-server.git] / processor.py

diff --git a/processor.py b/processor.py
index e7a340c..314f4c5 100644 (file)
--- a/processor.py
+++ b/processor.py
@@ -17,16 +17,17 @@ print_lock = threading.Lock()
 def print_log(*args):
     args = [str(item) for item in args]
     with print_lock:
-        sys.stderr.write(" ".join(args) + "\n")
+        sys.stderr.write(timestr() + " " + " ".join(args) + "\n")
         sys.stderr.flush()
 
 
 
 class Shared:
 
-    def __init__(self):
+    def __init__(self, config):
         self.lock = threading.Lock()
         self._stopped = False
+        self.config = config
 
     def stop(self):
         print_log( "Stopping Stratum" )
@@ -70,8 +71,8 @@ class Processor(threading.Thread):
 
 class Dispatcher:
 
-    def __init__(self):
-        self.shared = Shared()
+    def __init__(self, config):
+        self.shared = Shared(config)
         self.request_dispatcher = RequestDispatcher(self.shared)
         self.request_dispatcher.start()
         self.response_dispatcher = \
@@ -151,18 +152,10 @@ class RequestDispatcher(threading.Thread):
         params = request.get('params',[])
         suffix = method.split('.')[-1]
 
-        is_new = session.protocol_version >= 0.5
-
-        if is_new and method == 'blockchain.address.get_history': 
-            method = 'blockchain.address.get_history2'
-            request['method'] = method
-
-        if suffix == 'subscribe':
-            if is_new and method == 'blockchain.address.subscribe': 
-                method = 'blockchain.address.subscribe2'
-                request['method'] = method
-
-            session.subscribe_to_service(method, params)
+        if session is not None:
+            is_new = session.protocol_version >= 0.5
+            if suffix == 'subscribe':
+                session.subscribe_to_service(method, params)
 
         # store session and id locally
         request['id'] = self.store_session_id(session, request['id'])
@@ -183,9 +176,9 @@ class RequestDispatcher(threading.Thread):
             except:
                 pass
 
-        if session.protocol_version < 0.6:
-            print_log("stopping session from old client")
-            session.stop()
+            #if session.protocol_version < 0.6:
+            #    print_log("stopping session from old client", session.protocol_version)
+            #    session.stop()
 
     def get_sessions(self):
         with self.lock:
@@ -234,7 +227,7 @@ class Session:
             addr = None
 
         if self.subscriptions:
-            print_log( timestr(), self.name, self.address, addr, len(self.subscriptions), self.version )
+            print_log( "%4s"%self.name, "%15s"%self.address, "%35s"%addr, "%3d"%len(self.subscriptions), self.version )
 
     def stopped(self):
         with self.lock:
@@ -253,7 +246,7 @@ class Session:
             return method,
         elif method == "blockchain.headers.subscribe":
             return method,
-        elif method in ["blockchain.address.subscribe", "blockchain.address.subscribe2"]:
+        elif method in ["blockchain.address.subscribe"]:
             if not params:
                 return None
             else:
@@ -268,9 +261,9 @@ class Session:
 
 class ResponseDispatcher(threading.Thread):
 
-    def __init__(self, shared, processor):
+    def __init__(self, shared, request_dispatcher):
         self.shared = shared
-        self.processor = processor
+        self.request_dispatcher = request_dispatcher
         threading.Thread.__init__(self)
         self.daemon = True
 
@@ -279,7 +272,7 @@ class ResponseDispatcher(threading.Thread):
             self.update()
 
     def update(self):
-        response = self.processor.pop_response()
+        response = self.request_dispatcher.pop_response()
         #print "pop response", response
         internal_id = response.get('id')
         method = response.get('method')
@@ -287,28 +280,34 @@ class ResponseDispatcher(threading.Thread):
 
         # A notification
         if internal_id is None: # and method is not None and params is not None:
-            self.notification(method, params, response)
+            found = self.notification(method, params, response)
+            if not found and method == 'blockchain.address.subscribe':
+                params2 = [self.shared.config.get('server','password')] + params
+                self.request_dispatcher.push_request(None,{'method':method.replace('.subscribe', '.unsubscribe'), 'params':params2, 'id':None})
+
         # A response
-        elif internal_id is not None: # and method is None and params is None:
+        elif internal_id is not None: 
             self.send_response(internal_id, response)
         else:
             print_log( "no method", response)
 
     def notification(self, method, params, response):
         subdesc = Session.build_subdesc(method, params)
-        for session in self.processor.sessions:
+        found = False
+        for session in self.request_dispatcher.sessions:
             if session.stopped():
                 continue
             if session.contains_subscription(subdesc):
-                if response.get('method') == "blockchain.address.subscribe2":
-                    response['method'] = "blockchain.address.subscribe"
                 session.send_response(response)
+                found = True
+        # if not found: print_log( "no subscriber for", subdesc)
+        return found
 
     def send_response(self, internal_id, response):
-        session, message_id = self.processor.get_session_id(internal_id)
+        session, message_id = self.request_dispatcher.get_session_id(internal_id)
         if session:
             response['id'] = message_id
             session.send_response(response)
-        else:
-            print_log( "send_response: no session", message_id, internal_id, response )
+        #else:
+        #    print_log( "send_response: no session", message_id, internal_id, response )