<s> """<STR_LIT>""" <EOL> import httplib <EOL> try : <EOL> from cStringIO import StringIO <EOL> except ImportError : <EOL> from StringIO import StringIO <EOL> from avro import io <EOL> from avro import protocol <EOL> from avro import schema <EOL> HANDSHAKE_REQUEST_SCHEMA = schema . parse ( """<STR_LIT>""" ) <EOL> HANDSHAKE_RESPONSE_SCHEMA = schema . parse ( """<STR_LIT>""" ) <EOL> HANDSHAKE_REQUESTOR_WRITER = io . DatumWriter ( HANDSHAKE_REQUEST_SCHEMA ) <EOL> HANDSHAKE_REQUESTOR_READER = io . DatumReader ( HANDSHAKE_RESPONSE_SCHEMA ) <EOL> HANDSHAKE_RESPONDER_WRITER = io . DatumWriter ( HANDSHAKE_RESPONSE_SCHEMA ) <EOL> HANDSHAKE_RESPONDER_READER = io . DatumReader ( HANDSHAKE_REQUEST_SCHEMA ) <EOL> META_SCHEMA = schema . parse ( '<STR_LIT>' ) <EOL> META_WRITER = io . DatumWriter ( META_SCHEMA ) <EOL> META_READER = io . DatumReader ( META_SCHEMA ) <EOL> SYSTEM_ERROR_SCHEMA = schema . parse ( '<STR_LIT>' ) <EOL> REMOTE_HASHES = { } <EOL> REMOTE_PROTOCOLS = { } <EOL> BIG_ENDIAN_INT_STRUCT = io . struct_class ( '<STR_LIT>' ) <EOL> BUFFER_HEADER_LENGTH = <NUM_LIT:4> <EOL> BUFFER_SIZE = <NUM_LIT> <EOL> class AvroRemoteException ( schema . AvroException ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fail_msg = None ) : <EOL> schema . AvroException . __init__ ( self , fail_msg ) <EOL> class ConnectionClosedException ( schema . AvroException ) : <EOL> pass <EOL> class BaseRequestor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , local_protocol , transceiver ) : <EOL> self . _local_protocol = local_protocol <EOL> self . _transceiver = transceiver <EOL> self . _remote_protocol = None <EOL> self . _remote_hash = None <EOL> self . _send_protocol = None <EOL> local_protocol = property ( lambda self : self . _local_protocol ) <EOL> transceiver = property ( lambda self : self . _transceiver ) <EOL> def set_remote_protocol ( self , new_remote_protocol ) : <EOL> self . _remote_protocol = new_remote_protocol <EOL> REMOTE_PROTOCOLS [ self . transceiver . remote_name ] = self . remote_protocol <EOL> remote_protocol = property ( lambda self : self . _remote_protocol , <EOL> set_remote_protocol ) <EOL> def set_remote_hash ( self , new_remote_hash ) : <EOL> self . _remote_hash = new_remote_hash <EOL> REMOTE_HASHES [ self . transceiver . remote_name ] = self . remote_hash <EOL> remote_hash = property ( lambda self : self . _remote_hash , set_remote_hash ) <EOL> def set_send_protocol ( self , new_send_protocol ) : <EOL> self . _send_protocol = new_send_protocol <EOL> send_protocol = property ( lambda self : self . _send_protocol , set_send_protocol ) <EOL> def request ( self , message_name , request_datum ) : <EOL> """<STR_LIT>""" <EOL> buffer_writer = StringIO ( ) <EOL> buffer_encoder = io . BinaryEncoder ( buffer_writer ) <EOL> self . write_handshake_request ( buffer_encoder ) <EOL> self . write_call_request ( message_name , request_datum , buffer_encoder ) <EOL> call_request = buffer_writer . getvalue ( ) <EOL> return self . issue_request ( call_request , message_name , request_datum ) <EOL> def write_handshake_request ( self , encoder ) : <EOL> local_hash = self . local_protocol . md5 <EOL> remote_name = self . transceiver . remote_name <EOL> remote_hash = REMOTE_HASHES . get ( remote_name ) <EOL> if remote_hash is None : <EOL> remote_hash = local_hash <EOL> self . remote_protocol = self . local_protocol <EOL> request_datum = { } <EOL> request_datum [ '<STR_LIT>' ] = local_hash <EOL> request_datum [ '<STR_LIT>' ] = remote_hash <EOL> if self . send_protocol : <EOL> request_datum [ '<STR_LIT>' ] = str ( self . local_protocol ) <EOL> HANDSHAKE_REQUESTOR_WRITER . write ( request_datum , encoder ) <EOL> def write_call_request ( self , message_name , request_datum , encoder ) : <EOL> """<STR_LIT>""" <EOL> request_metadata = { } <EOL> META_WRITER . write ( request_metadata , encoder ) <EOL> message = self . local_protocol . messages . get ( message_name ) <EOL> if message is None : <EOL> raise schema . AvroException ( '<STR_LIT>' % message_name ) <EOL> encoder . write_utf8 ( message . name ) <EOL> self . write_request ( message . request , request_datum , encoder ) <EOL> def write_request ( self , request_schema , request_datum , encoder ) : <EOL> datum_writer = io . DatumWriter ( request_schema ) <EOL> datum_writer . write ( request_datum , encoder ) <EOL> def read_handshake_response ( self , decoder ) : <EOL> handshake_response = HANDSHAKE_REQUESTOR_READER . read ( decoder ) <EOL> match = handshake_response . get ( '<STR_LIT>' ) <EOL> if match == '<STR_LIT>' : <EOL> self . send_protocol = False <EOL> return True <EOL> elif match == '<STR_LIT>' : <EOL> if self . send_protocol : <EOL> raise schema . AvroException ( '<STR_LIT>' ) <EOL> self . remote_protocol = protocol . parse ( <EOL> handshake_response . get ( '<STR_LIT>' ) ) <EOL> self . remote_hash = handshake_response . get ( '<STR_LIT>' ) <EOL> self . send_protocol = False <EOL> return True <EOL> elif match == '<STR_LIT>' : <EOL> if self . send_protocol : <EOL> raise schema . AvroException ( '<STR_LIT>' ) <EOL> self . remote_protocol = protocol . parse ( <EOL> handshake_response . get ( '<STR_LIT>' ) ) <EOL> self . remote_hash = handshake_response . get ( '<STR_LIT>' ) <EOL> self . send_protocol = True <EOL> return False <EOL> else : <EOL> raise schema . AvroException ( '<STR_LIT>' % match ) <EOL> def read_call_response ( self , message_name , decoder ) : <EOL> """<STR_LIT>""" <EOL> response_metadata = META_READER . read ( decoder ) <EOL> remote_message_schema = self . remote_protocol . messages . get ( message_name ) <EOL> if remote_message_schema is None : <EOL> raise schema . AvroException ( '<STR_LIT>' % message_name ) <EOL> local_message_schema = self . local_protocol . messages . get ( message_name ) <EOL> if local_message_schema is None : <EOL> raise schema . AvroException ( '<STR_LIT>' % message_name ) <EOL> if not decoder . read_boolean ( ) : <EOL> writers_schema = remote_message_schema . response <EOL> readers_schema = local_message_schema . response <EOL> return self . read_response ( writers_schema , readers_schema , decoder ) <EOL> else : <EOL> writers_schema = remote_message_schema . errors <EOL> readers_schema = local_message_schema . errors <EOL> raise self . read_error ( writers_schema , readers_schema , decoder ) <EOL> def read_response ( self , writers_schema , readers_schema , decoder ) : <EOL> datum_reader = io . DatumReader ( writers_schema , readers_schema ) <EOL> result = datum_reader . read ( decoder ) <EOL> return result <EOL> def read_error ( self , writers_schema , readers_schema , decoder ) : <EOL> datum_reader = io . DatumReader ( writers_schema , readers_schema ) <EOL> return AvroRemoteException ( datum_reader . read ( decoder ) ) <EOL> class Requestor ( BaseRequestor ) : <EOL> def issue_request ( self , call_request , message_name , request_datum ) : <EOL> call_response = self . transceiver . transceive ( call_request ) <EOL> buffer_decoder = io . BinaryDecoder ( StringIO ( call_response ) ) <EOL> call_response_exists = self . read_handshake_response ( buffer_decoder ) <EOL> if call_response_exists : <EOL> return self . read_call_response ( message_name , buffer_decoder ) <EOL> else : <EOL> return self . request ( message_name , request_datum ) <EOL> class Responder ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , local_protocol ) : <EOL> self . _local_protocol = local_protocol <EOL> self . _local_hash = self . local_protocol . md5 <EOL> self . _protocol_cache = { } <EOL> self . set_protocol_cache ( self . local_hash , self . local_protocol ) <EOL> local_protocol = property ( lambda self : self . _local_protocol ) <EOL> local_hash = property ( lambda self : self . _local_hash ) <EOL> protocol_cache = property ( lambda self : self . _protocol_cache ) <EOL> def get_protocol_cache ( self , hash ) : <EOL> return self . protocol_cache . get ( hash ) <EOL> def set_protocol_cache ( self , hash , protocol ) : <EOL> self . protocol_cache [ hash ] = protocol <EOL> def respond ( self , call_request ) : <EOL> """<STR_LIT>""" <EOL> buffer_reader = StringIO ( call_request ) <EOL> buffer_decoder = io . BinaryDecoder ( buffer_reader ) <EOL> buffer_writer = StringIO ( ) <EOL> buffer_encoder = io . BinaryEncoder ( buffer_writer ) <EOL> error = None <EOL> response_metadata = { } <EOL> try : <EOL> remote_protocol = self . process_handshake ( buffer_decoder , buffer_encoder ) <EOL> if remote_protocol is None : <EOL> return buffer_writer . getvalue ( ) <EOL> request_metadata = META_READER . read ( buffer_decoder ) <EOL> remote_message_name = buffer_decoder . read_utf8 ( ) <EOL> remote_message = remote_protocol . messages . get ( remote_message_name ) <EOL> if remote_message is None : <EOL> fail_msg = '<STR_LIT>' % remote_message_name <EOL> raise schema . AvroException ( fail_msg ) <EOL> local_message = self . local_protocol . messages . get ( remote_message_name ) <EOL> if local_message is None : <EOL> fail_msg = '<STR_LIT>' % remote_message_name <EOL> raise schema . AvroException ( fail_msg ) <EOL> writers_schema = remote_message . request <EOL> readers_schema = local_message . request <EOL> request = self . read_request ( writers_schema , readers_schema , <EOL> buffer_decoder ) <EOL> try : <EOL> response = self . invoke ( local_message , request ) <EOL> except AvroRemoteException , e : <EOL> error = e <EOL> except Exception , e : <EOL> error = AvroRemoteException ( str ( e ) ) <EOL> META_WRITER . write ( response_metadata , buffer_encoder ) <EOL> buffer_encoder . write_boolean ( error is not None ) <EOL> if error is None : <EOL> writers_schema = local_message . response <EOL> self . write_response ( writers_schema , response , buffer_encoder ) <EOL> else : <EOL> writers_schema = local_message . errors <EOL> self . write_error ( writers_schema , error , buffer_encoder ) <EOL> except schema . AvroException , e : <EOL> error = AvroRemoteException ( str ( e ) ) <EOL> buffer_encoder = io . BinaryEncoder ( StringIO ( ) ) <EOL> META_WRITER . write ( response_metadata , buffer_encoder ) <EOL> buffer_encoder . write_boolean ( True ) <EOL> self . write_error ( SYSTEM_ERROR_SCHEMA , error , buffer_encoder ) <EOL> return buffer_writer . getvalue ( ) <EOL> def process_handshake ( self , decoder , encoder ) : <EOL> handshake_request = HANDSHAKE_RESPONDER_READER . read ( decoder ) <EOL> handshake_response = { } <EOL> client_hash = handshake_request . get ( '<STR_LIT>' ) <EOL> client_protocol = handshake_request . get ( '<STR_LIT>' ) <EOL> remote_protocol = self . get_protocol_cache ( client_hash ) <EOL> if remote_protocol is None and client_protocol is not None : <EOL> remote_protocol = protocol . parse ( client_protocol ) <EOL> self . set_protocol_cache ( client_hash , remote_protocol ) <EOL> server_hash = handshake_request . get ( '<STR_LIT>' ) <EOL> if self . local_hash == server_hash : <EOL> if remote_protocol is None : <EOL> handshake_response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> handshake_response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> if remote_protocol is None : <EOL> handshake_response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> handshake_response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if handshake_response [ '<STR_LIT>' ] != '<STR_LIT>' : <EOL> handshake_response [ '<STR_LIT>' ] = str ( self . local_protocol ) <EOL> handshake_response [ '<STR_LIT>' ] = self . local_hash <EOL> HANDSHAKE_RESPONDER_WRITER . write ( handshake_response , encoder ) <EOL> return remote_protocol <EOL> def invoke ( self , local_message , request ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def read_request ( self , writers_schema , readers_schema , decoder ) : <EOL> datum_reader = io . DatumReader ( writers_schema , readers_schema ) <EOL> return datum_reader . read ( decoder ) <EOL> def write_response ( self , writers_schema , response_datum , encoder ) : <EOL> datum_writer = io . DatumWriter ( writers_schema ) <EOL> datum_writer . write ( response_datum , encoder ) <EOL> def write_error ( self , writers_schema , error_exception , encoder ) : <EOL> datum_writer = io . DatumWriter ( writers_schema ) <EOL> datum_writer . write ( str ( error_exception ) , encoder ) <EOL> class FramedReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reader ) : <EOL> self . _reader = reader <EOL> reader = property ( lambda self : self . _reader ) <EOL> def read_framed_message ( self ) : <EOL> message = [ ] <EOL> while True : <EOL> buffer = StringIO ( ) <EOL> buffer_length = self . _read_buffer_length ( ) <EOL> if buffer_length == <NUM_LIT:0> : <EOL> return '<STR_LIT>' . join ( message ) <EOL> while buffer . tell ( ) < buffer_length : <EOL> chunk = self . reader . read ( buffer_length - buffer . tell ( ) ) <EOL> if chunk == '<STR_LIT>' : <EOL> raise ConnectionClosedException ( "<STR_LIT>" ) <EOL> buffer . write ( chunk ) <EOL> message . append ( buffer . getvalue ( ) ) <EOL> def _read_buffer_length ( self ) : <EOL> read = self . reader . read ( BUFFER_HEADER_LENGTH ) <EOL> if read == '<STR_LIT>' : <EOL> raise ConnectionClosedException ( "<STR_LIT>" ) <EOL> return BIG_ENDIAN_INT_STRUCT . unpack ( read ) [ <NUM_LIT:0> ] <EOL> class FramedWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , writer ) : <EOL> self . _writer = writer <EOL> writer = property ( lambda self : self . _writer ) <EOL> def write_framed_message ( self , message ) : <EOL> message_length = len ( message ) <EOL> total_bytes_sent = <NUM_LIT:0> <EOL> while message_length - total_bytes_sent > <NUM_LIT:0> : <EOL> if message_length - total_bytes_sent > BUFFER_SIZE : <EOL> buffer_length = BUFFER_SIZE <EOL> else : <EOL> buffer_length = message_length - total_bytes_sent <EOL> self . write_buffer ( message [ total_bytes_sent : <EOL> ( total_bytes_sent + buffer_length ) ] ) <EOL> total_bytes_sent += buffer_length <EOL> self . write_buffer_length ( <NUM_LIT:0> ) <EOL> def write_buffer ( self , chunk ) : <EOL> buffer_length = len ( chunk ) <EOL> self . write_buffer_length ( buffer_length ) <EOL> self . writer . write ( chunk ) <EOL> def write_buffer_length ( self , n ) : <EOL> self . writer . write ( BIG_ENDIAN_INT_STRUCT . pack ( n ) ) <EOL> class HTTPTransceiver ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , host , port , req_resource = '<STR_LIT:/>' ) : <EOL> self . req_resource = req_resource <EOL> self . conn = httplib . HTTPConnection ( host , port ) <EOL> self . conn . connect ( ) <EOL> sock = property ( lambda self : self . conn . sock ) <EOL> remote_name = property ( lambda self : self . sock . getsockname ( ) ) <EOL> def set_conn ( self , new_conn ) : <EOL> self . _conn = new_conn <EOL> conn = property ( lambda self : self . _conn , set_conn ) <EOL> req_resource = '<STR_LIT:/>' <EOL> def transceive ( self , request ) : <EOL> self . write_framed_message ( request ) <EOL> result = self . read_framed_message ( ) <EOL> return result <EOL> def read_framed_message ( self ) : <EOL> response = self . conn . getresponse ( ) <EOL> response_reader = FramedReader ( response ) <EOL> framed_message = response_reader . read_framed_message ( ) <EOL> response . read ( ) <EOL> return framed_message <EOL> def write_framed_message ( self , message ) : <EOL> req_method = '<STR_LIT:POST>' <EOL> req_headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' } <EOL> req_body_buffer = FramedWriter ( StringIO ( ) ) <EOL> req_body_buffer . write_framed_message ( message ) <EOL> req_body = req_body_buffer . writer . getvalue ( ) <EOL> self . conn . request ( req_method , self . req_resource , req_body , req_headers ) <EOL> def close ( self ) : <EOL> self . conn . close ( ) </s>
