<s> import msgpack <EOL> import six <EOL> class MessageType ( object ) : <EOL> REQUEST = <NUM_LIT:0> <EOL> RESPONSE = <NUM_LIT:1> <EOL> NOTIFY = <NUM_LIT:2> <EOL> class MessageEncoder ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( MessageEncoder , self ) . __init__ ( ) <EOL> self . _packer = msgpack . Packer ( encoding = None ) <EOL> self . _unpacker = msgpack . Unpacker ( encoding = None ) <EOL> self . _next_msgid = <NUM_LIT:0> <EOL> def _create_msgid ( self ) : <EOL> this_id = self . _next_msgid <EOL> self . _next_msgid = ( self . _next_msgid + <NUM_LIT:1> ) % <NUM_LIT> <EOL> return this_id <EOL> def create_request ( self , method , params ) : <EOL> assert isinstance ( method , six . binary_type ) <EOL> assert isinstance ( params , list ) <EOL> msgid = self . _create_msgid ( ) <EOL> return ( self . _packer . pack ( [ MessageType . REQUEST , msgid , method , <EOL> params ] ) , msgid ) <EOL> def create_response ( self , msgid , error = None , result = None ) : <EOL> assert isinstance ( msgid , int ) <EOL> assert <NUM_LIT:0> <= msgid and msgid <= <NUM_LIT> <EOL> assert error is None or result is None <EOL> return self . _packer . pack ( [ MessageType . RESPONSE , msgid , error , result ] ) <EOL> def create_notification ( self , method , params ) : <EOL> assert isinstance ( method , six . binary_type ) <EOL> assert isinstance ( params , list ) <EOL> return self . _packer . pack ( [ MessageType . NOTIFY , method , params ] ) <EOL> def get_and_dispatch_messages ( self , data , disp_table ) : <EOL> """<STR_LIT>""" <EOL> self . _unpacker . feed ( data ) <EOL> for m in self . _unpacker : <EOL> self . _dispatch_message ( m , disp_table ) <EOL> def _dispatch_message ( self , m , disp_table ) : <EOL> type = m [ <NUM_LIT:0> ] <EOL> try : <EOL> f = disp_table [ type ] <EOL> except KeyError : <EOL> return <EOL> f ( m [ <NUM_LIT:1> : ] ) <EOL> from collections import deque <EOL> import select <EOL> class EndPoint ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sock , encoder = None , disp_table = None ) : <EOL> if encoder is None : <EOL> encoder = MessageEncoder ( ) <EOL> self . _encoder = encoder <EOL> self . _sock = sock <EOL> if disp_table is None : <EOL> self . _table = { <EOL> MessageType . REQUEST : self . _enqueue_incoming_request , <EOL> MessageType . RESPONSE : self . _enqueue_incoming_response , <EOL> MessageType . NOTIFY : self . _enqueue_incoming_notification <EOL> } <EOL> else : <EOL> self . _table = disp_table <EOL> self . _send_buffer = bytearray ( ) <EOL> self . _pending_requests = set ( ) <EOL> self . _requests = deque ( ) <EOL> self . _notifications = deque ( ) <EOL> self . _responses = { } <EOL> self . _incoming = <NUM_LIT:0> <EOL> self . _closed_by_peer = False <EOL> def selectable ( self ) : <EOL> rlist = [ self . _sock ] <EOL> wlist = [ ] <EOL> if self . _send_buffer : <EOL> wlist . append ( self . _sock ) <EOL> return rlist , wlist <EOL> def process_outgoing ( self ) : <EOL> try : <EOL> sent_bytes = self . _sock . send ( self . _send_buffer ) <EOL> except IOError : <EOL> sent_bytes = <NUM_LIT:0> <EOL> del self . _send_buffer [ : sent_bytes ] <EOL> def process_incoming ( self ) : <EOL> self . receive_messages ( all = True ) <EOL> def process ( self ) : <EOL> self . process_outgoing ( ) <EOL> self . process_incoming ( ) <EOL> def block ( self ) : <EOL> rlist , wlist = self . selectable ( ) <EOL> select . select ( rlist , wlist , rlist + wlist ) <EOL> def serve ( self ) : <EOL> while not self . _closed_by_peer : <EOL> self . block ( ) <EOL> self . process ( ) <EOL> def _send_message ( self , msg ) : <EOL> self . _send_buffer += msg <EOL> self . process_outgoing ( ) <EOL> def send_request ( self , method , params ) : <EOL> """<STR_LIT>""" <EOL> msg , msgid = self . _encoder . create_request ( method , params ) <EOL> self . _send_message ( msg ) <EOL> self . _pending_requests . add ( msgid ) <EOL> return msgid <EOL> def send_response ( self , msgid , error = None , result = None ) : <EOL> """<STR_LIT>""" <EOL> msg = self . _encoder . create_response ( msgid , error , result ) <EOL> self . _send_message ( msg ) <EOL> def send_notification ( self , method , params ) : <EOL> """<STR_LIT>""" <EOL> msg = self . _encoder . create_notification ( method , params ) <EOL> self . _send_message ( msg ) <EOL> def receive_messages ( self , all = False ) : <EOL> """<STR_LIT>""" <EOL> while all or self . _incoming == <NUM_LIT:0> : <EOL> try : <EOL> packet = self . _sock . recv ( <NUM_LIT> ) <EOL> except IOError : <EOL> packet = None <EOL> if not packet : <EOL> if packet is not None : <EOL> self . _closed_by_peer = True <EOL> break <EOL> self . _encoder . get_and_dispatch_messages ( packet , self . _table ) <EOL> return self . _incoming > <NUM_LIT:0> <EOL> def _enqueue_incoming_request ( self , m ) : <EOL> self . _requests . append ( m ) <EOL> self . _incoming += <NUM_LIT:1> <EOL> def _enqueue_incoming_response ( self , m ) : <EOL> msgid , error , result = m <EOL> try : <EOL> self . _pending_requests . remove ( msgid ) <EOL> except KeyError : <EOL> return <EOL> assert msgid not in self . _responses <EOL> self . _responses [ msgid ] = ( error , result ) <EOL> self . _incoming += <NUM_LIT:1> <EOL> def _enqueue_incoming_notification ( self , m ) : <EOL> self . _notifications . append ( m ) <EOL> self . _incoming += <NUM_LIT:1> <EOL> def _get_message ( self , q ) : <EOL> try : <EOL> m = q . popleft ( ) <EOL> assert self . _incoming > <NUM_LIT:0> <EOL> self . _incoming -= <NUM_LIT:1> <EOL> return m <EOL> except IndexError : <EOL> return None <EOL> def get_request ( self ) : <EOL> return self . _get_message ( self . _requests ) <EOL> def get_response ( self , msgid ) : <EOL> try : <EOL> m = self . _responses . pop ( msgid ) <EOL> assert self . _incoming > <NUM_LIT:0> <EOL> self . _incoming -= <NUM_LIT:1> <EOL> except KeyError : <EOL> return None <EOL> error , result = m <EOL> return ( result , error ) <EOL> def get_notification ( self ) : <EOL> return self . _get_message ( self . _notifications ) <EOL> class RPCError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , error ) : <EOL> self . _error = error <EOL> def get_value ( self ) : <EOL> return self . _error <EOL> def __str__ ( self ) : <EOL> return str ( self . _error ) <EOL> class Client ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sock , encoder = None , notification_callback = None ) : <EOL> self . _endpoint = EndPoint ( sock , encoder ) <EOL> if notification_callback is None : <EOL> self . _notification_callback = lambda n : None <EOL> else : <EOL> self . _notification_callback = notification_callback <EOL> def _process_input_notification ( self ) : <EOL> n = self . _endpoint . get_notification ( ) <EOL> if n : <EOL> self . _notification_callback ( n ) <EOL> def _process_input_request ( self ) : <EOL> self . _endpoint . get_request ( ) <EOL> def call ( self , method , params ) : <EOL> """<STR_LIT>""" <EOL> msgid = self . _endpoint . send_request ( method , params ) <EOL> while True : <EOL> if not self . _endpoint . receive_messages ( ) : <EOL> raise EOFError ( "<STR_LIT>" ) <EOL> res = self . _endpoint . get_response ( msgid ) <EOL> if res : <EOL> result , error = res <EOL> if error is None : <EOL> return result <EOL> raise RPCError ( error ) <EOL> self . _process_input_notification ( ) <EOL> self . _process_input_request ( ) <EOL> def send_notification ( self , method , params ) : <EOL> """<STR_LIT>""" <EOL> self . _endpoint . send_notification ( method , params ) <EOL> def receive_notification ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _endpoint . receive_messages ( ) : <EOL> raise EOFError ( "<STR_LIT>" ) <EOL> self . _process_input_notification ( ) <EOL> self . _process_input_request ( ) <EOL> def peek_notification ( self ) : <EOL> while True : <EOL> rlist , _wlist = self . _endpoint . selectable ( ) <EOL> rlist , _wlist , _xlist = select . select ( rlist , [ ] , [ ] , <NUM_LIT:0> ) <EOL> if not rlist : <EOL> break <EOL> self . receive_notification ( ) </s>
