add try/except

Author: twotwo

rabbitmq/rabbitmq.py

@@ -42,6 +42,9 @@ def close(self):
         finally:
             self.is_open = False
 
+    def __repr__(self):
+        return self.amqp_url
+
 
 class MQProducer(RabbitMQ):
 
@@ -58,16 +61,19 @@ def publish(self, exchange: str, routing_key: str, body: str,
 
 class MQConsumer(RabbitMQ):
 
-    def open(self):
+    def open(self, prefetch: int = 200):
+        """
+        :param prefetch: prefetch count of message
+        """
         super(MQConsumer, self).open()
-        self.channel.basic_qos(prefetch_count=1)
+        self.channel.basic_qos(prefetch_count=prefetch)
 
     def consume(self, queue: str):
         for method_frame, properties, body in self.channel.consume(queue):
             return method_frame, properties, body.decode('utf-8')
 
     def basic_consume(self, queue: str, on_message_callback: typing.Callable):
-        self.channel.basic_consume(on_message_callback, queue=queue)
+        self.channel.basic_consume(queue, on_message_callback)
 
     def start_consuming(self):
         self.channel.start_consuming()

rabbitmq/thread_comsume.py

@@ -35,13 +35,18 @@ def __init__(self, conf: dict):
         # self.mq_consumer.open()
 
     def run(self):
-        self.mq_consumer.open()
-        queue_name = f"test-{self.exchange}"
-        self.mq_consumer.bind(self.exchange, self.routing_key, queue_name)
-        self.mq_consumer.basic_consume(queue_name, self.callback_rabbit)
-        logger.warning(
-            f"binding from {self.exchange}, routing_key={self.routing_key}, queue={queue_name}")
-        self.mq_consumer.start_consuming()
+        try:
+            self.mq_consumer.open()
+            queue_name = f"test-{self.exchange}"
+            self.mq_consumer.bind(self.exchange, self.routing_key, queue_name)
+            self.mq_consumer.basic_consume(queue_name, self.callback_rabbit)
+            logger.warning(
+                f"binding from {self.exchange}, routing_key={self.routing_key}, queue={queue_name}")
+            self.mq_consumer.start_consuming()
+        except KeyboardInterrupt:
+            pass
+        finally:
+            self.mq_consumer.close()
 
     @classmethod
     def callback_rabbit(cls, ch, method, properties, body):