<s> import copy <EOL> from kombu import Connection <EOL> from kombu . messaging import Producer <EOL> from st2common import log as logging <EOL> from st2common . transport . connection_retry_wrapper import ConnectionRetryWrapper <EOL> ANY_RK = '<STR_LIT:*>' <EOL> CREATE_RK = '<STR_LIT>' <EOL> UPDATE_RK = '<STR_LIT>' <EOL> DELETE_RK = '<STR_LIT>' <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class PoolPublisher ( object ) : <EOL> def __init__ ( self , urls ) : <EOL> self . pool = Connection ( urls , failover_strategy = '<STR_LIT>' ) . Pool ( limit = <NUM_LIT:10> ) <EOL> self . cluster_size = len ( urls ) <EOL> def errback ( self , exc , interval ) : <EOL> LOG . error ( '<STR_LIT>' , exc . message , exc_info = False ) <EOL> def publish ( self , payload , exchange , routing_key = '<STR_LIT>' ) : <EOL> with self . pool . acquire ( block = True ) as connection : <EOL> retry_wrapper = ConnectionRetryWrapper ( cluster_size = self . cluster_size , logger = LOG ) <EOL> def do_publish ( connection , channel ) : <EOL> producer = Producer ( channel ) <EOL> kwargs = { <EOL> '<STR_LIT:body>' : payload , <EOL> '<STR_LIT>' : exchange , <EOL> '<STR_LIT>' : routing_key , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> retry_wrapper . ensured ( connection = connection , <EOL> obj = producer , <EOL> to_ensure_func = producer . publish , <EOL> ** kwargs ) <EOL> retry_wrapper . run ( connection = connection , wrapped_callback = do_publish ) <EOL> class SharedPoolPublishers ( object ) : <EOL> """<STR_LIT>""" <EOL> shared_publishers = { } <EOL> def get_publisher ( self , urls ) : <EOL> urls_copy = copy . copy ( urls ) <EOL> urls_copy . sort ( ) <EOL> publisher_key = '<STR_LIT>' . join ( urls_copy ) <EOL> publisher = self . shared_publishers . get ( publisher_key , None ) <EOL> if not publisher : <EOL> publisher = PoolPublisher ( urls = urls ) <EOL> self . shared_publishers [ publisher_key ] = publisher <EOL> return publisher <EOL> class CUDPublisher ( object ) : <EOL> def __init__ ( self , urls , exchange ) : <EOL> self . _publisher = SharedPoolPublishers ( ) . get_publisher ( urls = urls ) <EOL> self . _exchange = exchange <EOL> def publish_create ( self , payload ) : <EOL> self . _publisher . publish ( payload , self . _exchange , CREATE_RK ) <EOL> def publish_update ( self , payload ) : <EOL> self . _publisher . publish ( payload , self . _exchange , UPDATE_RK ) <EOL> def publish_delete ( self , payload ) : <EOL> self . _publisher . publish ( payload , self . _exchange , DELETE_RK ) <EOL> class StatePublisherMixin ( object ) : <EOL> def __init__ ( self , urls , exchange ) : <EOL> self . _state_publisher = SharedPoolPublishers ( ) . get_publisher ( urls = urls ) <EOL> self . _state_exchange = exchange <EOL> def publish_state ( self , payload , state ) : <EOL> if not state : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> self . _state_publisher . publish ( payload , self . _state_exchange , state ) </s>
