<s> """<STR_LIT>""" <EOL> import Queue as queue <EOL> from carrot . backends import base <EOL> from eventlet import greenthread <EOL> from nova import log as logging <EOL> LOG = logging . getLogger ( "<STR_LIT>" ) <EOL> EXCHANGES = { } <EOL> QUEUES = { } <EOL> CONSUMERS = { } <EOL> class Message ( base . BaseMessage ) : <EOL> pass <EOL> class Exchange ( object ) : <EOL> def __init__ ( self , name , exchange_type ) : <EOL> self . name = name <EOL> self . exchange_type = exchange_type <EOL> self . _queue = queue . Queue ( ) <EOL> self . _routes = { } <EOL> def publish ( self , message , routing_key = None ) : <EOL> nm = self . name <EOL> LOG . debug ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % locals ( ) ) <EOL> if routing_key in self . _routes : <EOL> for f in self . _routes [ routing_key ] : <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , f ) <EOL> f ( message , routing_key = routing_key ) <EOL> def bind ( self , callback , routing_key ) : <EOL> self . _routes . setdefault ( routing_key , [ ] ) <EOL> self . _routes [ routing_key ] . append ( callback ) <EOL> class Queue ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> self . _queue = queue . Queue ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . name <EOL> def push ( self , message , routing_key = None ) : <EOL> self . _queue . put ( message ) <EOL> def size ( self ) : <EOL> return self . _queue . qsize ( ) <EOL> def pop ( self ) : <EOL> return self . _queue . get ( ) <EOL> class Backend ( base . BaseBackend ) : <EOL> def queue_declare ( self , queue , ** kwargs ) : <EOL> global QUEUES <EOL> if queue not in QUEUES : <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , queue ) <EOL> QUEUES [ queue ] = Queue ( queue ) <EOL> def exchange_declare ( self , exchange , type , * args , ** kwargs ) : <EOL> global EXCHANGES <EOL> if exchange not in EXCHANGES : <EOL> LOG . debug ( _ ( '<STR_LIT>' ) , exchange ) <EOL> EXCHANGES [ exchange ] = Exchange ( exchange , type ) <EOL> def queue_bind ( self , queue , exchange , routing_key , ** kwargs ) : <EOL> global EXCHANGES <EOL> global QUEUES <EOL> LOG . debug ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % locals ( ) ) <EOL> EXCHANGES [ exchange ] . bind ( QUEUES [ queue ] . push , routing_key ) <EOL> def declare_consumer ( self , queue , callback , consumer_tag , * args , ** kwargs ) : <EOL> global CONSUMERS <EOL> LOG . debug ( "<STR_LIT>" , consumer_tag ) <EOL> CONSUMERS [ consumer_tag ] = ( queue , callback ) <EOL> def cancel ( self , consumer_tag ) : <EOL> global CONSUMERS <EOL> LOG . debug ( "<STR_LIT>" , consumer_tag ) <EOL> del CONSUMERS [ consumer_tag ] <EOL> def consume ( self , limit = None ) : <EOL> global CONSUMERS <EOL> num = <NUM_LIT:0> <EOL> while True : <EOL> for ( queue , callback ) in CONSUMERS . itervalues ( ) : <EOL> item = self . get ( queue ) <EOL> if item : <EOL> callback ( item ) <EOL> num += <NUM_LIT:1> <EOL> yield <EOL> if limit and num == limit : <EOL> raise StopIteration ( ) <EOL> greenthread . sleep ( <NUM_LIT:0.1> ) <EOL> def get ( self , queue , no_ack = False ) : <EOL> global QUEUES <EOL> if not queue in QUEUES or not QUEUES [ queue ] . size ( ) : <EOL> return None <EOL> ( message_data , content_type , content_encoding ) = QUEUES [ queue ] . pop ( ) <EOL> message = Message ( backend = self , body = message_data , <EOL> content_type = content_type , <EOL> content_encoding = content_encoding ) <EOL> message . result = True <EOL> LOG . debug ( _ ( '<STR_LIT>' ) % locals ( ) ) <EOL> return message <EOL> def prepare_message ( self , message_data , delivery_mode , <EOL> content_type , content_encoding , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return ( message_data , content_type , content_encoding ) <EOL> def publish ( self , message , exchange , routing_key , ** kwargs ) : <EOL> global EXCHANGES <EOL> if exchange in EXCHANGES : <EOL> EXCHANGES [ exchange ] . publish ( message , routing_key = routing_key ) <EOL> def reset_all ( ) : <EOL> global EXCHANGES <EOL> global QUEUES <EOL> global CONSUMERS <EOL> EXCHANGES = { } <EOL> QUEUES = { } <EOL> CONSUMERS = { } </s>
