<s> import json <EOL> import Queue as queue <EOL> import socket <EOL> import threading <EOL> import time <EOL> import sys <EOL> from utils import random_string , timestr , print_log <EOL> from utils import logger <EOL> class Shared : <EOL> def __init__ ( self , config ) : <EOL> self . lock = threading . Lock ( ) <EOL> self . _stopped = False <EOL> self . config = config <EOL> self . _paused = True <EOL> def paused ( self ) : <EOL> with self . lock : <EOL> return self . _paused <EOL> def pause ( self ) : <EOL> with self . lock : <EOL> self . _paused = True <EOL> def unpause ( self ) : <EOL> with self . lock : <EOL> self . _paused = False <EOL> def stop ( self ) : <EOL> print_log ( "<STR_LIT>" ) <EOL> with self . lock : <EOL> self . _stopped = True <EOL> def stopped ( self ) : <EOL> with self . lock : <EOL> return self . _stopped <EOL> class Processor ( threading . Thread ) : <EOL> def __init__ ( self ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . daemon = True <EOL> self . dispatcher = None <EOL> self . queue = queue . Queue ( ) <EOL> def process ( self , request ) : <EOL> pass <EOL> def add_request ( self , session , request ) : <EOL> self . queue . put ( ( session , request ) ) <EOL> def push_response ( self , session , response ) : <EOL> self . dispatcher . request_dispatcher . push_response ( session , response ) <EOL> def close ( self ) : <EOL> pass <EOL> def run ( self ) : <EOL> while not self . shared . stopped ( ) : <EOL> try : <EOL> session , request = self . queue . get ( True , timeout = <NUM_LIT:1> ) <EOL> msg_id = request . get ( '<STR_LIT:id>' ) <EOL> except : <EOL> continue <EOL> if session . stopped ( ) : <EOL> continue <EOL> try : <EOL> result = self . process ( request ) <EOL> self . push_response ( session , { '<STR_LIT:id>' : msg_id , '<STR_LIT:result>' : result } ) <EOL> except BaseException , e : <EOL> self . push_response ( session , { '<STR_LIT:id>' : msg_id , '<STR_LIT:error>' : str ( e ) } ) <EOL> except : <EOL> logger . error ( "<STR_LIT>" , exc_info = True ) <EOL> self . push_response ( session , { '<STR_LIT:id>' : msg_id , '<STR_LIT:error>' : '<STR_LIT>' } ) <EOL> self . close ( ) <EOL> class Dispatcher : <EOL> def __init__ ( self , config ) : <EOL> self . shared = Shared ( config ) <EOL> self . request_dispatcher = RequestDispatcher ( self . shared ) <EOL> self . request_dispatcher . start ( ) <EOL> self . response_dispatcher = ResponseDispatcher ( self . shared , self . request_dispatcher ) <EOL> self . response_dispatcher . start ( ) <EOL> def register ( self , prefix , processor ) : <EOL> processor . dispatcher = self <EOL> processor . shared = self . shared <EOL> processor . start ( ) <EOL> self . request_dispatcher . processors [ prefix ] = processor <EOL> class RequestDispatcher ( threading . Thread ) : <EOL> def __init__ ( self , shared ) : <EOL> self . shared = shared <EOL> threading . Thread . __init__ ( self ) <EOL> self . daemon = True <EOL> self . request_queue = queue . Queue ( ) <EOL> self . response_queue = queue . Queue ( ) <EOL> self . lock = threading . Lock ( ) <EOL> self . idlock = threading . Lock ( ) <EOL> self . sessions = { } <EOL> self . processors = { } <EOL> self . lastgc = <NUM_LIT:0> <EOL> def push_response ( self , session , item ) : <EOL> self . response_queue . put ( ( session , item ) ) <EOL> def pop_response ( self ) : <EOL> return self . response_queue . get ( ) <EOL> def push_request ( self , session , item ) : <EOL> self . request_queue . put ( ( session , item ) ) <EOL> def pop_request ( self ) : <EOL> return self . request_queue . get ( ) <EOL> def get_session_by_address ( self , address ) : <EOL> for x in self . sessions . values ( ) : <EOL> if x . address == address : <EOL> return x <EOL> def run ( self ) : <EOL> if self . shared is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> while not self . shared . stopped ( ) : <EOL> session , request = self . pop_request ( ) <EOL> try : <EOL> self . do_dispatch ( session , request ) <EOL> except : <EOL> logger . error ( '<STR_LIT>' , exc_info = True ) <EOL> self . stop ( ) <EOL> def stop ( self ) : <EOL> pass <EOL> def do_dispatch ( self , session , request ) : <EOL> """<STR_LIT>""" <EOL> method = request [ '<STR_LIT>' ] <EOL> params = request . get ( '<STR_LIT>' , [ ] ) <EOL> suffix = method . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> if session is not None : <EOL> if suffix == '<STR_LIT>' : <EOL> if not session . subscribe_to_service ( method , params ) : <EOL> return <EOL> prefix = request [ '<STR_LIT>' ] . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] <EOL> try : <EOL> p = self . processors [ prefix ] <EOL> except : <EOL> print_log ( "<STR_LIT>" , prefix ) <EOL> return <EOL> p . add_request ( session , request ) <EOL> if method in [ '<STR_LIT>' ] : <EOL> try : <EOL> session . version = params [ <NUM_LIT:0> ] <EOL> session . protocol_version = float ( params [ <NUM_LIT:1> ] ) <EOL> except : <EOL> pass <EOL> def get_sessions ( self ) : <EOL> with self . lock : <EOL> r = self . sessions . values ( ) <EOL> return r <EOL> def add_session ( self , session ) : <EOL> key = session . key ( ) <EOL> with self . lock : <EOL> self . sessions [ key ] = session <EOL> def remove_session ( self , session ) : <EOL> key = session . key ( ) <EOL> with self . lock : <EOL> del self . sessions [ key ] <EOL> class Session : <EOL> def __init__ ( self , dispatcher ) : <EOL> self . dispatcher = dispatcher <EOL> self . bp = self . dispatcher . processors [ '<STR_LIT>' ] <EOL> self . _stopped = False <EOL> self . lock = threading . Lock ( ) <EOL> self . subscriptions = [ ] <EOL> self . address = '<STR_LIT>' <EOL> self . name = '<STR_LIT>' <EOL> self . version = '<STR_LIT>' <EOL> self . protocol_version = <NUM_LIT:0.> <EOL> self . time = time . time ( ) <EOL> self . max_subscriptions = dispatcher . shared . config . getint ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> threading . Timer ( <NUM_LIT:2> , self . info ) . start ( ) <EOL> def key ( self ) : <EOL> return self . address <EOL> def info ( self ) : <EOL> if self . subscriptions : <EOL> print_log ( "<STR_LIT>" % self . name , <EOL> "<STR_LIT>" % self . address , <EOL> "<STR_LIT>" % len ( self . subscriptions ) , <EOL> self . version ) <EOL> def stop ( self ) : <EOL> with self . lock : <EOL> if self . _stopped : <EOL> return <EOL> self . _stopped = True <EOL> self . shutdown ( ) <EOL> self . dispatcher . remove_session ( self ) <EOL> self . stop_subscriptions ( ) <EOL> def shutdown ( self ) : <EOL> pass <EOL> def stopped ( self ) : <EOL> with self . lock : <EOL> return self . _stopped <EOL> def subscribe_to_service ( self , method , params ) : <EOL> if self . stopped ( ) : <EOL> return False <EOL> if len ( self . subscriptions ) > self . max_subscriptions : <EOL> print_log ( "<STR_LIT>" , self . address ) <EOL> self . stop ( ) <EOL> return False <EOL> self . bp . do_subscribe ( method , params , self ) <EOL> with self . lock : <EOL> if ( method , params ) not in self . subscriptions : <EOL> self . subscriptions . append ( ( method , params ) ) <EOL> return True <EOL> def stop_subscriptions ( self ) : <EOL> with self . lock : <EOL> s = self . subscriptions [ : ] <EOL> for method , params in s : <EOL> self . bp . do_unsubscribe ( method , params , self ) <EOL> with self . lock : <EOL> self . subscriptions = [ ] <EOL> class ResponseDispatcher ( threading . Thread ) : <EOL> def __init__ ( self , shared , request_dispatcher ) : <EOL> self . shared = shared <EOL> self . request_dispatcher = request_dispatcher <EOL> threading . Thread . __init__ ( self ) <EOL> self . daemon = True <EOL> def run ( self ) : <EOL> while not self . shared . stopped ( ) : <EOL> session , response = self . request_dispatcher . pop_response ( ) <EOL> session . send_response ( response ) </s>
