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 = ResponseDispatcher(self.shared, self.request_dispatcher)
  55         self.response_dispatcher.start()
  56
  57     def register(self, prefix, processor):
  58         processor.dispatcher = self
  59         processor.shared = self.shared
  60         processor.start()
  61         self.request_dispatcher.processors[prefix] = processor
  62
  63
  64
  65 class RequestDispatcher(threading.Thread):
  66
  67     def __init__(self, shared):
  68         self.shared = shared
  69         threading.Thread.__init__(self)
  70         self.daemon = True
  71         self.request_queue = queue.Queue()
  72         self.response_queue = queue.Queue()
  73         self.internal_ids = {}
  74         self.internal_id = 1
  75         self.lock = threading.Lock()
  76         self.sessions = []
  77         self.processors = {}
  78
  79     def push_response(self, item):
  80         self.response_queue.put(item)
  81
  82     def pop_response(self):
  83         return self.response_queue.get()
  84
  85     def push_request(self, session, item):
  86         self.request_queue.put((session,item))
  87
  88     def pop_request(self):
  89         return self.request_queue.get()
  90
  91     def get_session_id(self, internal_id):
  92         with self.lock:
  93             return self.internal_ids.pop(internal_id)
  94
  95     def store_session_id(self, session, msgid):
  96         with self.lock:
  97             self.internal_ids[self.internal_id] = session, msgid
  98             r = self.internal_id
  99             self.internal_id += 1
 100             return r
 101
 102     def run(self):
 103         if self.shared is None:
 104             raise TypeError("self.shared not set in Processor")
 105         while not self.shared.stopped():
 106             session, request = self.pop_request()
 107             self.process(session, request)
 108
 109         self.stop()
 110
 111     def stop(self):
 112         pass
 113
 114     def process(self, session, request):
 115         method = request['method']
 116         params = request.get('params',[])
 117
 118         suffix = method.split('.')[-1]
 119         if suffix == 'subscribe':
 120             session.subscribe_to_service(method, params)
 121
 122         # store session and id locally
 123         request['id'] = self.store_session_id(session, request['id'])
 124
 125         # dispatch request to the relevant module..
 126         prefix = request['method'].split('.')[0]
 127         try:
 128             p = self.processors[prefix]
 129         except:
 130             print "error: no processor for", prefix
 131             return
 132         try:
 133             p.process(request)
 134         except:
 135             traceback.print_exc(file=sys.stdout)
 136
 137         if method in ['server.version']:
 138             session.version = params[0]
 139
 140
 141     def add_session(self, session):
 142         with self.lock:
 143             self.sessions.append(session)
 144
 145     def collect_garbage(self):
 146         # Deep copy entire sessions list and blank it
 147         # This is done to minimise lock contention
 148         with self.lock:
 149             sessions = self.sessions[:]
 150             self.sessions = []
 151         for session in sessions:
 152             if not session.stopped():
 153                 # If session is still alive then re-add it back
 154                 # to our internal register
 155                 self.add_session(session)
 156
 157
 158 class Session:
 159
 160     def __init__(self):
 161         self._stopped = False
 162         self.lock = threading.Lock()
 163         self.subscriptions = []
 164         self.address = ''
 165         self.name = ''
 166         threading.Timer(2, self.info).start()
 167
 168     def info(self):
 169         for s in self.subscriptions:
 170             m, p = s
 171             if m == 'blockchain.address.subscribe':
 172                 addr = p[0]
 173                 break
 174         else:
 175             addr = None
 176         print timestr(), self.name, self.address, addr, len(self.subscriptions), self.version
 177
 178     def stopped(self):
 179         with self.lock:
 180             return self._stopped
 181
 182     def subscribe_to_service(self, method, params):
 183         with self.lock:
 184             self.subscriptions.append((method, params))
 185
 186
 187 class ResponseDispatcher(threading.Thread):
 188
 189     def __init__(self, shared, processor):
 190         self.shared = shared
 191         self.processor = processor
 192         threading.Thread.__init__(self)
 193         self.daemon = True
 194
 195     def run(self):
 196         while not self.shared.stopped():
 197             response = self.processor.pop_response()
 198             #print "pop response", response
 199             internal_id = response.get('id')
 200             params = response.get('params', [])
 201
 202             # This is wrong. "params" and "method" should never
 203             # be in a response.
 204             if internal_id is None:
 205                 method = response.get('method')
 206                 if method is None:
 207                     print "no method", response
 208                     continue
 209                 for session in self.processor.sessions:
 210                     if not session.stopped():
 211                         if (method,params) in session.subscriptions:
 212                             session.send_response(response)
 213                 continue
 214
 215             session, message_id = self.processor.get_session_id(internal_id)
 216             response['id'] = message_id
 217             session.send_response(response)
 218