X-Git-Url: https://git.novaco.in/?a=blobdiff_plain;f=processor.py;h=314f4c5a8f6c6aa67436638080e5ccfa0bf698cd;hb=e16fd4a0d1bb692d2fa8c6edaffcb5a7a427f3d4;hp=dfe61285ba17a63206aad8821086aa87284cade4;hpb=827d26f24ef2148c4eb22bab764a7b2798233334;p=electrum-server.git

diff --git a/processor.py b/processor.py
index dfe6128..314f4c5 100644
--- a/processor.py
+++ b/processor.py
@@ -13,14 +13,24 @@ def timestr():
     return time.strftime("[%d/%m/%Y-%H:%M:%S]")
 
 
+print_lock = threading.Lock()
+def print_log(*args):
+    args = [str(item) for item in args]
+    with print_lock:
+        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 "Stopping Stratum"
+        print_log( "Stopping Stratum" )
         with self.lock:
             self._stopped = True
 
@@ -55,14 +65,14 @@ class Processor(threading.Thread):
             except:
                 traceback.print_exc(file=sys.stdout)
 
-        print "processor terminating"
+        print_log( "processor terminating")
             
 
 
 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 = \
@@ -142,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'])
@@ -162,7 +164,7 @@ class RequestDispatcher(threading.Thread):
         try:
             p = self.processors[prefix]
         except:
-            print "error: no processor for", prefix
+            print_log( "error: no processor for", prefix)
             return
 
         p.add_request(request)
@@ -174,6 +176,10 @@ class RequestDispatcher(threading.Thread):
             except:
                 pass
 
+            #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:
             r = self.sessions[:]
@@ -221,8 +227,7 @@ class Session:
             addr = None
 
         if self.subscriptions:
-            print 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:
@@ -241,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:
@@ -256,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
 
@@ -267,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')
@@ -275,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 "no method", response
+            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 "send_response: no session", message_id, internal_id, response
+        #else:
+        #    print_log( "send_response: no session", message_id, internal_id, response )