compat.py 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213
  1. """Carrot compatibility interface.
  2. See https://pypi.org/project/carrot/ for documentation.
  3. """
  4. from __future__ import absolute_import, unicode_literals
  5. from itertools import count
  6. from . import messaging
  7. from .entity import Exchange, Queue
  8. from .five import items
  9. __all__ = ('Publisher', 'Consumer')
  10. # XXX compat attribute
  11. entry_to_queue = Queue.from_dict
  12. def _iterconsume(connection, consumer, no_ack=False, limit=None):
  13. consumer.consume(no_ack=no_ack)
  14. for iteration in count(0): # for infinity
  15. if limit and iteration >= limit:
  16. break
  17. yield connection.drain_events()
  18. class Publisher(messaging.Producer):
  19. """Carrot compatible producer."""
  20. exchange = ''
  21. exchange_type = 'direct'
  22. routing_key = ''
  23. durable = True
  24. auto_delete = False
  25. _closed = False
  26. def __init__(self, connection, exchange=None, routing_key=None,
  27. exchange_type=None, durable=None, auto_delete=None,
  28. channel=None, **kwargs):
  29. if channel:
  30. connection = channel
  31. self.exchange = exchange or self.exchange
  32. self.exchange_type = exchange_type or self.exchange_type
  33. self.routing_key = routing_key or self.routing_key
  34. if auto_delete is not None:
  35. self.auto_delete = auto_delete
  36. if durable is not None:
  37. self.durable = durable
  38. if not isinstance(self.exchange, Exchange):
  39. self.exchange = Exchange(name=self.exchange,
  40. type=self.exchange_type,
  41. routing_key=self.routing_key,
  42. auto_delete=self.auto_delete,
  43. durable=self.durable)
  44. super(Publisher, self).__init__(connection, self.exchange, **kwargs)
  45. def send(self, *args, **kwargs):
  46. return self.publish(*args, **kwargs)
  47. def close(self):
  48. super(Publisher, self).close()
  49. self._closed = True
  50. def __enter__(self):
  51. return self
  52. def __exit__(self, *exc_info):
  53. self.close()
  54. @property
  55. def backend(self):
  56. return self.channel
  57. class Consumer(messaging.Consumer):
  58. """Carrot compatible consumer."""
  59. queue = ''
  60. exchange = ''
  61. routing_key = ''
  62. exchange_type = 'direct'
  63. durable = True
  64. exclusive = False
  65. auto_delete = False
  66. _closed = False
  67. def __init__(self, connection, queue=None, exchange=None,
  68. routing_key=None, exchange_type=None, durable=None,
  69. exclusive=None, auto_delete=None, **kwargs):
  70. self.backend = connection.channel()
  71. if durable is not None:
  72. self.durable = durable
  73. if exclusive is not None:
  74. self.exclusive = exclusive
  75. if auto_delete is not None:
  76. self.auto_delete = auto_delete
  77. self.queue = queue or self.queue
  78. self.exchange = exchange or self.exchange
  79. self.exchange_type = exchange_type or self.exchange_type
  80. self.routing_key = routing_key or self.routing_key
  81. exchange = Exchange(self.exchange,
  82. type=self.exchange_type,
  83. routing_key=self.routing_key,
  84. auto_delete=self.auto_delete,
  85. durable=self.durable)
  86. queue = Queue(self.queue,
  87. exchange=exchange,
  88. routing_key=self.routing_key,
  89. durable=self.durable,
  90. exclusive=self.exclusive,
  91. auto_delete=self.auto_delete)
  92. super(Consumer, self).__init__(self.backend, queue, **kwargs)
  93. def revive(self, channel):
  94. self.backend = channel
  95. super(Consumer, self).revive(channel)
  96. def close(self):
  97. self.cancel()
  98. self.backend.close()
  99. self._closed = True
  100. def __enter__(self):
  101. return self
  102. def __exit__(self, *exc_info):
  103. self.close()
  104. def __iter__(self):
  105. return self.iterqueue(infinite=True)
  106. def fetch(self, no_ack=None, enable_callbacks=False):
  107. if no_ack is None:
  108. no_ack = self.no_ack
  109. message = self.queues[0].get(no_ack)
  110. if message:
  111. if enable_callbacks:
  112. self.receive(message.payload, message)
  113. return message
  114. def process_next(self):
  115. raise NotImplementedError('Use fetch(enable_callbacks=True)')
  116. def discard_all(self, filterfunc=None):
  117. if filterfunc is not None:
  118. raise NotImplementedError(
  119. 'discard_all does not implement filters')
  120. return self.purge()
  121. def iterconsume(self, limit=None, no_ack=None):
  122. return _iterconsume(self.connection, self, no_ack, limit)
  123. def wait(self, limit=None):
  124. it = self.iterconsume(limit)
  125. return list(it)
  126. def iterqueue(self, limit=None, infinite=False):
  127. for items_since_start in count(): # for infinity
  128. item = self.fetch()
  129. if (not infinite and item is None) or \
  130. (limit and items_since_start >= limit):
  131. break
  132. yield item
  133. class ConsumerSet(messaging.Consumer):
  134. def __init__(self, connection, from_dict=None, consumers=None,
  135. channel=None, **kwargs):
  136. if channel:
  137. self._provided_channel = True
  138. self.backend = channel
  139. else:
  140. self._provided_channel = False
  141. self.backend = connection.channel()
  142. queues = []
  143. if consumers:
  144. for consumer in consumers:
  145. queues.extend(consumer.queues)
  146. if from_dict:
  147. for queue_name, queue_options in items(from_dict):
  148. queues.append(Queue.from_dict(queue_name, **queue_options))
  149. super(ConsumerSet, self).__init__(self.backend, queues, **kwargs)
  150. def iterconsume(self, limit=None, no_ack=False):
  151. return _iterconsume(self.connection, self, no_ack, limit)
  152. def discard_all(self):
  153. return self.purge()
  154. def add_consumer_from_dict(self, queue, **options):
  155. return self.add_queue(Queue.from_dict(queue, **options))
  156. def add_consumer(self, consumer):
  157. for queue in consumer.queues:
  158. self.add_queue(queue)
  159. def revive(self, channel):
  160. self.backend = channel
  161. super(ConsumerSet, self).revive(channel)
  162. def close(self):
  163. self.cancel()
  164. if not self._provided_channel:
  165. self.channel.close()