<s> import time <EOL> import socket <EOL> from itertools import count <EOL> from stompy import Client <EOL> from stompy import Empty as QueueEmpty <EOL> from carrot . backends . base import BaseMessage , BaseBackend <EOL> DEFAULT_PORT = <NUM_LIT> <EOL> class Message ( BaseMessage ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , backend , frame , ** kwargs ) : <EOL> self . _frame = frame <EOL> self . backend = backend <EOL> kwargs [ "<STR_LIT:body>" ] = frame . body <EOL> kwargs [ "<STR_LIT>" ] = frame . headers [ "<STR_LIT>" ] <EOL> kwargs [ "<STR_LIT>" ] = frame . headers . get ( "<STR_LIT>" ) <EOL> kwargs [ "<STR_LIT>" ] = frame . headers . get ( "<STR_LIT>" ) <EOL> kwargs [ "<STR_LIT>" ] = frame . headers . get ( "<STR_LIT>" ) <EOL> super ( Message , self ) . __init__ ( backend , ** kwargs ) <EOL> def ack ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . acknowledged : <EOL> raise self . MessageStateError ( <EOL> "<STR_LIT>" % self . _state ) <EOL> self . backend . ack ( self . _frame ) <EOL> self . _state = "<STR_LIT>" <EOL> def reject ( self ) : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" ) <EOL> def requeue ( self ) : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" ) <EOL> class Backend ( BaseBackend ) : <EOL> Stomp = Client <EOL> Message = Message <EOL> default_port = DEFAULT_PORT <EOL> def __init__ ( self , connection , ** kwargs ) : <EOL> self . connection = connection <EOL> self . default_port = kwargs . get ( "<STR_LIT>" , self . default_port ) <EOL> self . _channel = None <EOL> self . _consumers = { } <EOL> self . _callbacks = { } <EOL> def establish_connection ( self ) : <EOL> conninfo = self . connection <EOL> if not conninfo . port : <EOL> conninfo . port = self . default_port <EOL> stomp = self . Stomp ( conninfo . hostname , conninfo . port ) <EOL> stomp . connect ( username = conninfo . userid , password = conninfo . password ) <EOL> stomp . drain_events = self . drain_events <EOL> return stomp <EOL> def close_connection ( self , connection ) : <EOL> try : <EOL> connection . disconnect ( ) <EOL> except socket . error : <EOL> pass <EOL> def queue_exists ( self , queue ) : <EOL> return True <EOL> def queue_purge ( self , queue , ** kwargs ) : <EOL> for purge_count in count ( <NUM_LIT:0> ) : <EOL> try : <EOL> frame = self . channel . get_nowait ( ) <EOL> except QueueEmpty : <EOL> return purge_count <EOL> else : <EOL> self . channel . ack ( frame ) <EOL> def declare_consumer ( self , queue , no_ack , callback , consumer_tag , <EOL> ** kwargs ) : <EOL> ack = no_ack and "<STR_LIT>" or "<STR_LIT>" <EOL> self . channel . subscribe ( queue , ack = ack ) <EOL> self . _consumers [ consumer_tag ] = queue <EOL> self . _callbacks [ queue ] = callback <EOL> def drain_events ( self , timeout = None ) : <EOL> start_time = time . time ( ) <EOL> while True : <EOL> frame = self . channel . get ( ) <EOL> if frame : <EOL> break <EOL> if time . time ( ) - time_start > timeout : <EOL> raise socket . timeout ( "<STR_LIT>" ) <EOL> queue = frame . headers . get ( "<STR_LIT>" ) <EOL> if not queue or queue not in self . _callbacks : <EOL> return <EOL> self . _callbacks [ queue ] ( frame ) <EOL> def consume ( self , limit = None ) : <EOL> """<STR_LIT>""" <EOL> for total_message_count in count ( ) : <EOL> if limit and total_message_count >= limit : <EOL> raise StopIteration <EOL> self . drain_events ( ) <EOL> yield True <EOL> def queue_declare ( self , queue , * args , ** kwargs ) : <EOL> self . channel . subscribe ( queue , ack = "<STR_LIT>" ) <EOL> def get ( self , queue , no_ack = False ) : <EOL> try : <EOL> frame = self . channel . get_nowait ( ) <EOL> except QueueEmpty : <EOL> return None <EOL> else : <EOL> return self . message_to_python ( frame ) <EOL> def ack ( self , frame ) : <EOL> self . channel . ack ( frame ) <EOL> def message_to_python ( self , raw_message ) : <EOL> """<STR_LIT>""" <EOL> return self . Message ( backend = self , frame = raw_message ) <EOL> def prepare_message ( self , message_data , delivery_mode , priority = <NUM_LIT:0> , <EOL> content_type = None , content_encoding = None ) : <EOL> persistent = "<STR_LIT:false>" <EOL> if delivery_mode == <NUM_LIT:2> : <EOL> persistent = "<STR_LIT:true>" <EOL> priority = priority or <NUM_LIT:0> <EOL> return { "<STR_LIT:body>" : message_data , <EOL> "<STR_LIT>" : persistent , <EOL> "<STR_LIT>" : priority , <EOL> "<STR_LIT>" : content_encoding , <EOL> "<STR_LIT>" : content_type } <EOL> def publish ( self , message , exchange , routing_key , ** kwargs ) : <EOL> message [ "<STR_LIT>" ] = exchange <EOL> self . channel . stomp . send ( message ) <EOL> def cancel ( self , consumer_tag ) : <EOL> if not self . _channel or consumer_tag not in self . _consumers : <EOL> return <EOL> queue = self . _consumers . pop ( consumer_tag ) <EOL> self . channel . unsubscribe ( queue ) <EOL> def close ( self ) : <EOL> for consumer_tag in self . _consumers . keys ( ) : <EOL> self . cancel ( consumer_tag ) <EOL> if self . _channel : <EOL> try : <EOL> self . _channel . disconnect ( ) <EOL> except socket . error : <EOL> pass <EOL> @ property <EOL> def channel ( self ) : <EOL> if not self . _channel : <EOL> self . _channel = self . establish_connection ( ) <EOL> return self . _channel </s>
