import socket
import threading
import time
+import traceback, sys
import Queue as queue
class Shared:
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):
+ 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
self.lock = threading.Lock()
self.sessions = []
+ self.processors = {}
def push_response(self, item):
self.response_queue.put(item)
raise TypeError("self.shared not set in Processor")
while not self.shared.stopped():
session, request = self.pop_request()
-
- method = request['method']
- params = request.get('params',[])
-
- if method in [ 'numblocks.subscribe', 'address.subscribe', 'server.peers']:
- session.subscribe_to_service(method, params)
-
- # store session and id locally
- request['id'] = self.store_session_id(session, request['id'])
- self.process(request)
+ self.process(session, request)
self.stop()
def stop(self):
pass
- def process(self, request):
- print "New request", request
- # Do stuff...
- # response = request
- # When ready, you call
- # self.push_response(response)
+ def process(self, session, request):
+ method = request['method']
+ params = request.get('params',[])
+
+ suffix = method.split('.')[-1]
+ if suffix == 'subscribe':
+ session.subscribe_to_service(method, params)
+
+ # store session and id locally
+ request['id'] = self.store_session_id(session, request['id'])
+
+ # dispatch request to the relevant module..
+ prefix = request['method'].split('.')[0]
+ try:
+ p = self.processors[prefix]
+ except:
+ print "error: no processor for", prefix
+ return
+ try:
+ p.process(request)
+ except:
+ traceback.print_exc(file=sys.stdout)
+
def add_session(self, session):
with self.lock:
self.subscriptions.append((method, params))
-class Dispatcher(threading.Thread):
+class ResponseDispatcher(threading.Thread):
def __init__(self, shared, processor):
self.shared = shared
self.processor = processor
threading.Thread.__init__(self)
+ self.daemon = True
def run(self):
while not self.shared.stopped():