import traceback, sys
import Queue as queue
+def random_string(N):
+ import random, string
+ return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
+
+def timestr():
+ return time.strftime("[%d/%m/%Y-%H:%M:%S]")
+
+
class Shared:
def __init__(self):
class Processor(threading.Thread):
def __init__(self):
- self.shared = None
+ threading.Thread.__init__(self)
+ self.daemon = True
+ self.dispatcher = None
+
+ def process(self, request):
+ pass
+
+ def push_response(self, response):
+ print "response", response
+ #self.dispatcher.request_dispatcher.push_response(response)
+
+
+
+class Dispatcher:
+
+ def __init__(self):
+ self.shared = Shared()
+ self.request_dispatcher = RequestDispatcher(self.shared)
+ self.request_dispatcher.start()
+ self.response_dispatcher = ResponseDispatcher(self.shared, self.request_dispatcher)
+ self.response_dispatcher.start()
+
+ def register(self, prefix, processor):
+ processor.dispatcher = self
+ processor.shared = self.shared
+ processor.start()
+ self.request_dispatcher.processors[prefix] = processor
+
+
+
+class RequestDispatcher(threading.Thread):
+
+ def __init__(self, shared):
+ self.shared = shared
threading.Thread.__init__(self)
self.daemon = True
self.request_queue = queue.Queue()
self.internal_id += 1
return r
- def register(self, prefix, function):
- self.processors[prefix] = function
-
def run(self):
if self.shared is None:
raise TypeError("self.shared not set in Processor")
# dispatch request to the relevant module..
prefix = request['method'].split('.')[0]
try:
- func = self.processors[prefix]
+ p = self.processors[prefix]
except:
print "error: no processor for", prefix
return
try:
- func(request,self.response_queue)
+ p.process(request)
except:
traceback.print_exc(file=sys.stdout)
+ if method in ['server.version']:
+ session.version = params[0]
+
def add_session(self, session):
with self.lock:
self._stopped = False
self.lock = threading.Lock()
self.subscriptions = []
+ self.address = ''
+ self.name = ''
+ threading.Timer(2, self.info).start()
+
+ def info(self):
+ for s in self.subscriptions:
+ m, p = s
+ if m == 'blockchain.address.subscribe':
+ addr = p[0]
+ break
+ else:
+ addr = None
+ print timestr(), self.name, self.address, addr, len(self.subscriptions), self.version
def stopped(self):
with self.lock:
self.subscriptions.append((method, params))
-class Dispatcher(threading.Thread):
+class ResponseDispatcher(threading.Thread):
def __init__(self, shared, processor):
self.shared = shared
response = self.processor.pop_response()
#print "pop response", response
internal_id = response.get('id')
- params = response.get('params',[])
- try:
- method = response['method']
- except:
- print "no method", response
- continue
-
- if internal_id:
- session, message_id = self.processor.get_session_id(internal_id)
- response['id'] = message_id
- session.send_response(response)
-
- else:
+ params = response.get('params', [])
+
+ # This is wrong. "params" and "method" should never
+ # be in a response.
+ if internal_id is None:
+ method = response.get('method')
+ if method is None:
+ print "no method", response
+ continue
for session in self.processor.sessions:
if not session.stopped():
if (method,params) in session.subscriptions:
session.send_response(response)
+ continue
+
+ session, message_id = self.processor.get_session_id(internal_id)
+ response['id'] = message_id
+ session.send_response(response)