processor.py

   1 import json
   2 import socket
   3 import threading
   4 import time
   5 import traceback, sys
   6 import Queue as queue
   7
   8 def random_string(N):
   9     import random, string
  10     return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
  11
  12 def timestr():
  13     return time.strftime("[%d/%m/%Y-%H:%M:%S]")
  14
  15
  16 class Shared:
  17
  18     def __init__(self):
  19         self.lock = threading.Lock()
  20         self._stopped = False
  21
  22     def stop(self):
  23         print "Stopping Stratum"
  24         with self.lock:
  25             self._stopped = True
  26
  27     def stopped(self):
  28         with self.lock:
  29             return self._stopped
  30
  31
  32 class Processor(threading.Thread):
  33
  34     def __init__(self):
  35         threading.Thread.__init__(self)
  36         self.daemon = True
  37         self.dispatcher = None
  38
  39     def process(self, request):
  40         pass
  41
  42     def push_response(self, response):
  43         #print "response", response
  44         self.dispatcher.request_dispatcher.push_response(response)
  45
  46
  47
  48 class Dispatcher:
  49
  50     def __init__(self):
  51         self.shared = Shared()
  52         self.request_dispatcher = RequestDispatcher(self.shared)
  53         self.request_dispatcher.start()
  54         self.response_dispatcher = \
  55             ResponseDispatcher(self.shared, self.request_dispatcher)
  56         self.response_dispatcher.start()
  57
  58     def register(self, prefix, processor):
  59         processor.dispatcher = self
  60         processor.shared = self.shared
  61         processor.start()
  62         self.request_dispatcher.processors[prefix] = processor
  63
  64
  65
  66 class RequestDispatcher(threading.Thread):
  67
  68     def __init__(self, shared):
  69         self.shared = shared
  70         threading.Thread.__init__(self)
  71         self.daemon = True
  72         self.request_queue = queue.Queue()
  73         self.response_queue = queue.Queue()
  74         self.internal_ids = {}
  75         self.internal_id = 1
  76         self.lock = threading.Lock()
  77         self.sessions = []
  78         self.processors = {}
  79
  80     def push_response(self, item):
  81         self.response_queue.put(item)
  82
  83     def pop_response(self):
  84         return self.response_queue.get()
  85
  86     def push_request(self, session, item):
  87         self.request_queue.put((session,item))
  88
  89     def pop_request(self):
  90         return self.request_queue.get()
  91
  92     def get_session_by_address(self, address):
  93         for x in self.sessions:
  94             if x.address == address:
  95                 return x
  96
  97     def get_session_id(self, internal_id):
  98         with self.lock:
  99             return self.internal_ids.pop(internal_id)
 100
 101     def store_session_id(self, session, msgid):
 102         with self.lock:
 103             self.internal_ids[self.internal_id] = session, msgid
 104             r = self.internal_id
 105             self.internal_id += 1
 106             return r
 107
 108     def run(self):
 109         if self.shared is None:
 110             raise TypeError("self.shared not set in Processor")
 111         while not self.shared.stopped():
 112             session, request = self.pop_request()
 113             self.process(session, request)
 114
 115         self.stop()
 116
 117     def stop(self):
 118         pass
 119
 120     def process(self, session, request):
 121         method = request['method']
 122         params = request.get('params',[])
 123
 124         suffix = method.split('.')[-1]
 125         if suffix == 'subscribe':
 126             session.subscribe_to_service(method, params)
 127
 128         # store session and id locally
 129         request['id'] = self.store_session_id(session, request['id'])
 130
 131         # dispatch request to the relevant module..
 132         prefix = request['method'].split('.')[0]
 133         try:
 134             p = self.processors[prefix]
 135         except:
 136             print "error: no processor for", prefix
 137             return
 138         try:
 139             p.process(request)
 140         except:
 141             traceback.print_exc(file=sys.stdout)
 142
 143         if method in ['server.version']:
 144             session.version = params[0]
 145
 146
 147     def add_session(self, session):
 148         with self.lock:
 149             self.sessions.append(session)
 150
 151     def collect_garbage(self):
 152         # Deep copy entire sessions list and blank it
 153         # This is done to minimise lock contention
 154         with self.lock:
 155             sessions = self.sessions[:]
 156             self.sessions = []
 157         for session in sessions:
 158             if not session.stopped():
 159                 # If session is still alive then re-add it back
 160                 # to our internal register
 161                 self.add_session(session)
 162
 163
 164 class Session:
 165
 166     def __init__(self):
 167         self._stopped = False
 168         self.lock = threading.Lock()
 169         self.subscriptions = []
 170         self.address = ''
 171         self.name = ''
 172         self.version = 'unknown'
 173         threading.Timer(2, self.info).start()
 174
 175     # Debugging method. Doesn't need to be threadsafe.
 176     def info(self):
 177         for sub in self.subscriptions:
 178             #print sub
 179             method = sub[0]
 180             if method == 'blockchain.address.subscribe':
 181                 addr = sub[1]
 182                 break
 183         else:
 184             addr = None
 185
 186         if self.subscriptions:
 187             print timestr(), self.name, self.address, addr,\
 188                 len(self.subscriptions), self.version
 189
 190     def stopped(self):
 191         with self.lock:
 192             return self._stopped
 193
 194     def subscribe_to_service(self, method, params):
 195         subdesc = self.build_subdesc(method, params)
 196         with self.lock:
 197             if subdesc is not None:
 198                 self.subscriptions.append(subdesc)
 199
 200     # subdesc = A subscription description
 201     @staticmethod
 202     def build_subdesc(method, params):
 203         if method == "blockchain.numblocks.subscribe":
 204             return method,
 205         elif method == "blockchain.address.subscribe":
 206             if not params:
 207                 return None
 208             else:
 209                 return method, params[0]
 210         else:
 211             return None
 212
 213     def contains_subscription(self, subdesc):
 214         with self.lock:
 215             return subdesc in self.subscriptions
 216
 217
 218 class ResponseDispatcher(threading.Thread):
 219
 220     def __init__(self, shared, processor):
 221         self.shared = shared
 222         self.processor = processor
 223         threading.Thread.__init__(self)
 224         self.daemon = True
 225
 226     def run(self):
 227         while not self.shared.stopped():
 228             self.update()
 229
 230     def update(self):
 231         response = self.processor.pop_response()
 232         #print "pop response", response
 233         internal_id = response.get('id')
 234         method = response.get('method')
 235         params = response.get('params')
 236
 237         # A notification
 238         if internal_id is None: # and method is not None and params is not None:
 239             self.notification(method, params, response)
 240         # A response
 241         elif internal_id is not None: # and method is None and params is None:
 242             self.send_response(internal_id, response)
 243         else:
 244             print "no method", response
 245
 246     def notification(self, method, params, response):
 247         subdesc = Session.build_subdesc(method, params)
 248         for session in self.processor.sessions:
 249             if session.stopped():
 250                 continue
 251             if session.contains_subscription(subdesc):
 252                 session.send_response(response)
 253
 254     def send_response(self, internal_id, response):
 255         session, message_id = self.processor.get_session_id(internal_id)
 256         if session:
 257             response['id'] = message_id
 258             session.send_response(response)
 259         else:
 260             print "send_response: no session", message_id, internal_id, response
 261