bynder / qpid-bow Goto Github PK

async def consumer(message: Message) -> bool:
   print(f'Received message: {message.body}')
   return True

class QpidClient:
   def __init__(self, queue_name : str, reply_queue_name : str, server_address = '127.0.0.1'):
      if queue_name is None or len(queue_name) == 0 or reply_queue_name is None or len(reply_queue_name) == 0:
         raise Exception('QpidClient: queue_name and reply_queue_name must be strings.')
      self.server_url = server_address
      self.queue_name = queue_name
      self.timeout = timedelta(seconds=5)
      self.reply_sender = Sender(server_url=server_address)
      self.reply_queue_name = reply_queue_name
      if not self.try_create_queues():
         raise Exception('QpidClient: could not create queues ', self.queue_name, ' and ', reply_queue_name)

   def try_create_queues(self):
      rtn = True
      try:
         create_queue(self.queue_name,       priorities=10, server_url=self.server_url)
      except QMF2ObjectExists:
         print('Queue already exists:', self.queue_name)
         # pass  # Queue already exists
      except QMF2Exception:
         print("QMF2 error during queue creation")
         rtn = False
      try:
         create_queue(self.reply_queue_name, priorities=10, server_url=self.server_url)
      except QMF2ObjectExists:
         print('Queue already exists:', self.reply_queue_name)
         # pass  # Queue already exists
      except QMF2Exception:
         print("QMF2 error during queue creation")
         rtn = False
      return rtn

   def send_messages(self, message_to_send = []):
      sender = Sender(self.queue_name, self.server_url)
      messages = []
      for msg in message_to_send:
         message = create_message(msg)
         message.address = self.queue_name
         message.reply_to = self.reply_queue_name
         messages.append(message)
         print('message', msg)
      sender.queue(messages)
      sender.send()

   # This should? spend timeout seconds trying to get messages.
   def recv_messages_async(self, messages_recieved = [], message_count_limit=1):
      # >>>> These three lines receive all the queued messages, but then execution is blocked in loop.run_forever()
      #receiver.run()
      #loop = asyncio.get_event_loop()
      #loop.run_forever()

      # >>>> These two lines receive nothing even when there are enqueued messages.
      receiver = Receiver(consumer, self.queue_name, self.server_url, container_class=AsyncioContainer)
      receiver.receive( timeout=timedelta(0, 1, 0, 0, 0, 0, 0,) )