old_contents: from kombu import Connection
  from kombu.pools import producers
  
  CREATE_RK = 'create'
  UPDATE_RK = 'update'
  DELETE_RK = 'delete'
  
  
  class PoolPublisher(object):
      def __init__(self, url):
          self.pool = Connection(url).Pool(10)
  
      def publish(self, payload, exchange, routing_key=''):
          # pickling the payload for now. Better serialization mechanism is essential.
          with self.pool.acquire(block=True) as connection:
              with producers[connection].acquire(block=True) as producer:
                  producer.publish(payload, exchange=exchange, routing_key=routing_key,
                                   serializer='pickle')
  
  
  class CUDPublisher(object):
      def __init__(self, url, exchange):
          self._publisher = PoolPublisher(url)
          self._exchange = exchange
  
      def publish_create(self, payload):
          self._publisher.publish(payload, self._exchange, CREATE_RK)
  
      def publish_update(self, payload):
          self._publisher.publish(payload, self._exchange, UPDATE_RK)
  
      def publish_delete(self, payload):
          self._publisher.publish(payload, self._exchange, DELETE_RK)
  
new_contents: from kombu import Connection
  from kombu.pools import producers
  
  from st2common import log as logging
  
  CREATE_RK = 'create'
  UPDATE_RK = 'update'
  DELETE_RK = 'delete'
  
  LOG = logging.getLogger(__name__)
  
  
  class PoolPublisher(object):
      def __init__(self, url):
          self.pool = Connection(url).Pool(limit=10, preload=1)
  
      def errback(self, exc, interval):
          LOG.error('Rabbitmq connection error: %r', exc, exc_info=True)
  
      def publish(self, payload, exchange, routing_key=''):
          # pickling the payload for now. Better serialization mechanism is essential.
          with self.pool.acquire(block=True) as connection:
              with producers[connection].acquire(block=True) as producer:
                  try:
                      publish = connection.ensure(producer, producer.publish, errback=self.errback,
                                                  max_retries=3)
                      publish(payload, exchange=exchange, routing_key=routing_key,
                              serializer='pickle')
                  except:
                      LOG.exception('Connections to rabbitmq cannot be re-established.')
  
  
  class CUDPublisher(object):
      def __init__(self, url, exchange):
          self._publisher = PoolPublisher(url)
          self._exchange = exchange
  
      def publish_create(self, payload):
          self._publisher.publish(payload, self._exchange, CREATE_RK)
  
      def publish_update(self, payload):
          self._publisher.publish(payload, self._exchange, UPDATE_RK)
  
      def publish_delete(self, payload):
          self._publisher.publish(payload, self._exchange, DELETE_RK)
  
subject: Handle rabbitmq connection closes with retries Refer: https://github.com/celery/kombu/commit/1c5f0ecf9f79e0c3f8f689057541295d3bf7a336#diff-9fa13492ded708799168a1c48c81584cR104 * Fabric uses multiprocessing for running tasks in parallel. Kombu wants to reset connections on multiprocessing forks. So you see abrupt rabbitmq channel closes. There also seems to be a bug in connection status not being synced to channel status in Kombu. * Fix: Add retries to publish loop and call ensure_connection to make sure connection can be re-established. This is needed anyways. It just happens to be a work around for the above bug.