<s> import sys <EOL> import socket <EOL> import struct <EOL> import logging <EOL> from types import NoneType <EOL> import functools <EOL> from errors import ProgrammingError , IntegrityError , InterfaceError <EOL> import helpers <EOL> import asyncjobs <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> host = None , <EOL> port = None , <EOL> dbuser = None , <EOL> dbpass = None , <EOL> autoreconnect = True , <EOL> pool = None , <EOL> backend = "<STR_LIT>" , <EOL> rs = None , <EOL> seed = None , <EOL> secondary_only = False , <EOL> ** kwargs ) : <EOL> assert isinstance ( autoreconnect , bool ) <EOL> assert isinstance ( dbuser , ( str , unicode , NoneType ) ) <EOL> assert isinstance ( dbpass , ( str , unicode , NoneType ) ) <EOL> assert isinstance ( rs , ( str , NoneType ) ) <EOL> assert pool <EOL> assert isinstance ( secondary_only , bool ) <EOL> if rs : <EOL> assert host is None <EOL> assert port is None <EOL> assert isinstance ( seed , ( set , list ) ) <EOL> else : <EOL> assert isinstance ( host , ( str , unicode ) ) <EOL> assert isinstance ( port , int ) <EOL> assert seed is None <EOL> self . _host = host <EOL> self . _port = port <EOL> self . __rs = rs <EOL> self . __seed = seed <EOL> self . __secondary_only = secondary_only <EOL> self . __dbuser = dbuser <EOL> self . __dbpass = dbpass <EOL> self . __stream = None <EOL> self . __callback = None <EOL> self . __alive = False <EOL> self . __autoreconnect = autoreconnect <EOL> self . __pool = pool <EOL> self . __kwargs = kwargs <EOL> self . __backend = self . __load_backend ( backend ) <EOL> self . __job_queue = [ ] <EOL> self . usage_count = <NUM_LIT:0> <EOL> self . __connect ( self . connection_error ) <EOL> def connection_error ( self , error ) : <EOL> raise error <EOL> def __load_backend ( self , name ) : <EOL> __import__ ( '<STR_LIT>' % name ) <EOL> mod = sys . modules [ '<STR_LIT>' % name ] <EOL> return mod . AsyncBackend ( ) <EOL> def __connect ( self , err_callback ) : <EOL> if self . __dbuser and self . __dbpass : <EOL> self . _put_job ( asyncjobs . AuthorizeJob ( self , self . __dbuser , self . __dbpass , self . __pool , err_callback ) ) <EOL> if self . __rs : <EOL> self . _put_job ( asyncjobs . ConnectRSJob ( self , self . __seed , self . __rs , self . __secondary_only , err_callback ) ) <EOL> self . __alive = True <EOL> else : <EOL> self . _socket_connect ( ) <EOL> def _socket_connect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . usage_count = <NUM_LIT:0> <EOL> try : <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM , <NUM_LIT:0> ) <EOL> s . connect ( ( self . _host , self . _port ) ) <EOL> self . __stream = self . __backend . register_stream ( s , ** self . __kwargs ) <EOL> self . __stream . set_close_callback ( self . _socket_close ) <EOL> self . __alive = True <EOL> except socket . error , error : <EOL> raise InterfaceError ( error ) <EOL> def _socket_close ( self ) : <EOL> """<STR_LIT>""" <EOL> callback = self . __callback <EOL> self . __callback = None <EOL> try : <EOL> if callback : <EOL> callback ( None , InterfaceError ( '<STR_LIT>' ) ) <EOL> finally : <EOL> self . __job_queue = [ ] <EOL> self . __alive = False <EOL> self . __pool . cache ( self ) <EOL> def _close ( self ) : <EOL> """<STR_LIT>""" <EOL> callback = self . __callback <EOL> self . __callback = None <EOL> try : <EOL> if callback : <EOL> callback ( None , InterfaceError ( '<STR_LIT>' ) ) <EOL> finally : <EOL> self . __job_queue = [ ] <EOL> self . __alive = False <EOL> self . __stream . close ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _close ( ) <EOL> finally : <EOL> self . __pool . cache ( self ) <EOL> def send_message ( self , message , callback ) : <EOL> """<STR_LIT>""" <EOL> if self . __callback is not None : <EOL> raise ProgrammingError ( '<STR_LIT>' ) <EOL> if callback : <EOL> err_callback = functools . partial ( callback , None ) <EOL> else : <EOL> err_callback = None <EOL> for job in self . __job_queue : <EOL> if isinstance ( job , asyncjobs . AsyncJob ) : <EOL> job . update_err_callback ( err_callback ) <EOL> if not self . __alive : <EOL> if self . __autoreconnect : <EOL> self . __connect ( err_callback ) <EOL> else : <EOL> raise InterfaceError ( '<STR_LIT>' ) <EOL> self . _put_job ( asyncjobs . AsyncMessage ( self , message , callback ) , <NUM_LIT:0> ) <EOL> self . _next_job ( ) <EOL> def _put_job ( self , job , pos = None ) : <EOL> if pos is None : <EOL> pos = len ( self . __job_queue ) <EOL> self . __job_queue . insert ( pos , job ) <EOL> def _next_job ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __job_queue : <EOL> job = self . __job_queue . pop ( ) <EOL> job . process ( ) <EOL> def _send_message ( self , message , callback ) : <EOL> self . __callback = callback <EOL> self . usage_count += <NUM_LIT:1> <EOL> ( self . __request_id , data ) = message <EOL> try : <EOL> self . __stream . write ( data ) <EOL> if self . __callback : <EOL> self . __stream . read ( <NUM_LIT:16> , callback = self . _parse_header ) <EOL> else : <EOL> self . __request_id = None <EOL> self . __pool . cache ( self ) <EOL> except IOError : <EOL> self . __alive = False <EOL> raise <EOL> def _parse_header ( self , header ) : <EOL> length = int ( struct . unpack ( "<STR_LIT>" , header [ : <NUM_LIT:4> ] ) [ <NUM_LIT:0> ] ) <EOL> request_id = struct . unpack ( "<STR_LIT>" , header [ <NUM_LIT:8> : <NUM_LIT:12> ] ) [ <NUM_LIT:0> ] <EOL> assert request_id == self . __request_id , "<STR_LIT>" % ( self . __request_id , <EOL> request_id ) <EOL> operation = <NUM_LIT:1> <EOL> assert operation == struct . unpack ( "<STR_LIT>" , header [ <NUM_LIT:12> : ] ) [ <NUM_LIT:0> ] <EOL> try : <EOL> self . __stream . read ( length - <NUM_LIT:16> , callback = self . _parse_response ) <EOL> except IOError : <EOL> self . __alive = False <EOL> raise <EOL> def _parse_response ( self , response ) : <EOL> callback = self . __callback <EOL> request_id = self . __request_id <EOL> self . __request_id = None <EOL> self . __callback = None <EOL> if not self . __job_queue : <EOL> self . __pool . cache ( self ) <EOL> try : <EOL> response = helpers . _unpack_response ( response , request_id ) <EOL> except Exception , e : <EOL> logging . debug ( '<STR_LIT>' % e ) <EOL> callback ( None , e ) <EOL> return <EOL> if response and response [ '<STR_LIT:data>' ] and response [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) and response [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] . get ( '<STR_LIT:code>' ) : <EOL> callback ( response , IntegrityError ( response [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , code = response [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:code>' ] ) ) <EOL> return <EOL> callback ( response ) </s>
