<s> """<STR_LIT>""" <EOL> import socket <EOL> import warnings <EOL> from collections import deque <EOL> from copy import copy <EOL> from Queue import Queue , Empty as QueueEmpty <EOL> from amqplib . client_0_8 . connection import AMQPConnectionException <EOL> from carrot . backends import get_backend_cls <EOL> from carrot . utils import retry_over_time <EOL> DEFAULT_CONNECT_TIMEOUT = <NUM_LIT:5> <EOL> SETTING_PREFIX = "<STR_LIT>" <EOL> COMPAT_SETTING_PREFIX = "<STR_LIT>" <EOL> ARG_TO_DJANGO_SETTING = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:password>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:port>" : "<STR_LIT>" , <EOL> } <EOL> SETTING_DEPRECATED_FMT = "<STR_LIT>" "<STR_LIT>" <EOL> class BrokerConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> virtual_host = "<STR_LIT:/>" <EOL> port = None <EOL> insist = False <EOL> connect_timeout = DEFAULT_CONNECT_TIMEOUT <EOL> ssl = False <EOL> _closed = True <EOL> backend_cls = None <EOL> ConnectionException = AMQPConnectionException <EOL> @ property <EOL> def host ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT::>" . join ( [ self . hostname , str ( self . port ) ] ) <EOL> def __init__ ( self , hostname = None , userid = None , password = None , <EOL> virtual_host = None , port = None , pool = None , ** kwargs ) : <EOL> self . hostname = hostname <EOL> self . userid = userid <EOL> self . password = password <EOL> self . virtual_host = virtual_host or self . virtual_host <EOL> self . port = port or self . port <EOL> self . insist = kwargs . get ( "<STR_LIT>" , self . insist ) <EOL> self . pool = pool <EOL> self . connect_timeout = kwargs . get ( "<STR_LIT>" , <EOL> self . connect_timeout ) <EOL> self . ssl = kwargs . get ( "<STR_LIT>" , self . ssl ) <EOL> self . backend_cls = ( kwargs . get ( "<STR_LIT>" ) or <EOL> kwargs . get ( "<STR_LIT>" ) ) <EOL> self . _closed = None <EOL> self . _connection = None <EOL> def __copy__ ( self ) : <EOL> return self . __class__ ( self . hostname , self . userid , self . password , <EOL> self . virtual_host , self . port , <EOL> insist = self . insist , <EOL> connect_timeout = self . connect_timeout , <EOL> ssl = self . ssl , <EOL> backend_cls = self . backend_cls , <EOL> pool = self . pool ) <EOL> @ property <EOL> def connection ( self ) : <EOL> if self . _closed == True : <EOL> return <EOL> if not self . _connection : <EOL> self . _connection = self . _establish_connection ( ) <EOL> self . _closed = False <EOL> return self . _connection <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , e_type , e_value , e_trace ) : <EOL> if e_type : <EOL> raise e_type ( e_value ) <EOL> self . close ( ) <EOL> def _establish_connection ( self ) : <EOL> return self . create_backend ( ) . establish_connection ( ) <EOL> def get_backend_cls ( self ) : <EOL> """<STR_LIT>""" <EOL> backend_cls = self . backend_cls <EOL> if not backend_cls or isinstance ( backend_cls , basestring ) : <EOL> backend_cls = get_backend_cls ( backend_cls ) <EOL> return backend_cls <EOL> def create_backend ( self ) : <EOL> """<STR_LIT>""" <EOL> backend_cls = self . get_backend_cls ( ) <EOL> return backend_cls ( connection = self ) <EOL> def channel ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . create_backend ( ) <EOL> def get_channel ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . connection . channel ( ) <EOL> def connect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _closed = False <EOL> return self . connection <EOL> def drain_events ( self , ** kwargs ) : <EOL> return self . connection . drain_events ( ** kwargs ) <EOL> def ensure_connection ( self , errback = None , max_retries = None , <EOL> interval_start = <NUM_LIT:2> , interval_step = <NUM_LIT:2> , interval_max = <NUM_LIT:30> ) : <EOL> """<STR_LIT>""" <EOL> retry_over_time ( self . connect , self . connection_errors , ( ) , { } , <EOL> errback , max_retries , <EOL> interval_start , interval_step , interval_max ) <EOL> return self <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if self . _connection : <EOL> backend = self . create_backend ( ) <EOL> backend . close_connection ( self . _connection ) <EOL> except socket . error : <EOL> pass <EOL> self . _closed = True <EOL> def release ( self ) : <EOL> if not self . pool : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" ) <EOL> self . pool . release ( self ) <EOL> def info ( self ) : <EOL> """<STR_LIT>""" <EOL> backend_cls = self . backend_cls or "<STR_LIT>" <EOL> port = self . port or self . create_backend ( ) . default_port <EOL> return { "<STR_LIT>" : self . hostname , <EOL> "<STR_LIT>" : self . userid , <EOL> "<STR_LIT:password>" : self . password , <EOL> "<STR_LIT>" : self . virtual_host , <EOL> "<STR_LIT:port>" : port , <EOL> "<STR_LIT>" : self . insist , <EOL> "<STR_LIT>" : self . ssl , <EOL> "<STR_LIT>" : backend_cls , <EOL> "<STR_LIT>" : backend_cls , <EOL> "<STR_LIT>" : self . connect_timeout } <EOL> @ property <EOL> def connection_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . create_backend ( ) . connection_errors <EOL> @ property <EOL> def channel_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . create_backend ( ) . channel_errors <EOL> AMQPConnection = BrokerConnection <EOL> class ConnectionLimitExceeded ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class ConnectionPool ( object ) : <EOL> def __init__ ( self , source_connection , min = <NUM_LIT:2> , max = None , preload = True ) : <EOL> self . source_connection = source_connection <EOL> self . min = min <EOL> self . max = max <EOL> self . preload = preload <EOL> self . source_connection . pool = self <EOL> self . _connections = Queue ( ) <EOL> self . _dirty = deque ( ) <EOL> self . _connections . put ( self . source_connection ) <EOL> for i in range ( min - <NUM_LIT:1> ) : <EOL> self . _connections . put_nowait ( self . _new_connection ( ) ) <EOL> def acquire ( self , block = False , timeout = None , connect_timeout = None ) : <EOL> try : <EOL> conn = self . _connections . get ( block = block , timeout = timeout ) <EOL> except QueueEmpty : <EOL> conn = self . _new_connection ( ) <EOL> self . _dirty . append ( conn ) <EOL> if connect_timeout is not None : <EOL> conn . connect_timeout = connect_timeout <EOL> return conn <EOL> def release ( self , connection ) : <EOL> self . _dirty . remove ( connection ) <EOL> self . _connections . put_nowait ( connection ) <EOL> def _new_connection ( self ) : <EOL> if len ( self . _dirty ) >= self . max : <EOL> raise ConnectionLimitExceeded ( self . max ) <EOL> return copy ( self . source_connection ) <EOL> def get_django_conninfo ( settings = None ) : <EOL> ci = { } <EOL> if settings is None : <EOL> from django . conf import settings <EOL> ci [ "<STR_LIT>" ] = getattr ( settings , "<STR_LIT>" , None ) <EOL> for arg_name , setting_name in ARG_TO_DJANGO_SETTING . items ( ) : <EOL> setting = "<STR_LIT>" % ( SETTING_PREFIX , setting_name ) <EOL> compat_setting = "<STR_LIT>" % ( COMPAT_SETTING_PREFIX , setting_name ) <EOL> if hasattr ( settings , setting ) : <EOL> ci [ arg_name ] = getattr ( settings , setting , None ) <EOL> elif hasattr ( settings , compat_setting ) : <EOL> ci [ arg_name ] = getattr ( settings , compat_setting , None ) <EOL> warnings . warn ( DeprecationWarning ( SETTING_DEPRECATED_FMT % ( <EOL> compat_setting , setting ) ) ) <EOL> if "<STR_LIT>" not in ci : <EOL> if hasattr ( settings , "<STR_LIT>" ) : <EOL> ci [ "<STR_LIT>" ] = settings . AMQP_SERVER <EOL> warnings . warn ( DeprecationWarning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> return ci <EOL> class DjangoBrokerConnection ( BrokerConnection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> settings = kwargs . pop ( "<STR_LIT>" , None ) <EOL> kwargs = dict ( get_django_conninfo ( settings ) , ** kwargs ) <EOL> super ( DjangoBrokerConnection , self ) . __init__ ( * args , ** kwargs ) <EOL> DjangoAMQPConnection = DjangoBrokerConnection </s>
