From 0dc337197311d0fec373bf479f8ee0ac26b4cad6 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期三, 15 六月 2022 16:33:47 +0800
Subject: [PATCH] client enhanced&camera contorller
---
src/engine/MessageThreads.py | 20 ++++++++++++--------
1 files changed, 12 insertions(+), 8 deletions(-)
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 04d86e2..98e7cc0 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -50,7 +50,7 @@
def stop(self):
"""stop the thread"""
self.is_stop = True
- self.connection.close()
+ # self.connection.close()
def run(self):
@@ -62,7 +62,7 @@
while True:
try:
- if not self.message_queue.empty():
+ if not self.message_queue.empty() and not self.is_stop:
# decode the routing key and message body from tuple list
message = self.message_queue.get()
@@ -115,7 +115,7 @@
def stop(self):
"""stop the thread"""
self.is_stop = True
- self.connection.close()
+ # self.connection.close()
def run(self):
@@ -130,6 +130,7 @@
channel.queue_declare(queue=self.MQqueue_name, durable=True)
for binding_key in self.binding_keys:
channel.queue_bind(exchange=self.exchange_name, queue=self.MQqueue_name, routing_key=binding_key)
+ print(f"[*] Using [{binding_key}] bind [{self.MQqueue_name}] on [{self.exchange_name}]")
# consume the message
channel.basic_qos(prefetch_count=1)
@@ -140,6 +141,9 @@
channel.start_consuming()
def callback(self,ch, method, properties, body):
+
+ if self.is_stop:
+ ch.close()
self.message_queue.put(body)
ch.basic_ack(delivery_tag = method.delivery_tag)
@@ -191,7 +195,7 @@
queue = self.queue_list[i_tag]
# make sure the queue is not empty
- while not queue.empty():
+ while not queue.empty() and not self.is_stop:
# decode the msg using cooresponding method
queue_tag = self.queue_tags[i_tag]
decode_method = self.decode_select(queue_tag)
@@ -214,7 +218,7 @@
while not queue.empty() and count < nums_limit:
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.LogsMessage()
msg_pb.ParseFromString(msg)
@@ -241,7 +245,7 @@
while not queue.empty():
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.ImagesMessage()
msg_pb.ParseFromString(msg)
@@ -268,7 +272,7 @@
filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
filepath = os.path.join(self.imagesmsg_folder,filename)
with open(filepath, 'wb') as f:
- f.write(image)
+ f.write(image_byte)
# display the info to the console
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
@@ -282,7 +286,7 @@
while not queue.empty():
# extracting the msg from queue
- msg = queue.pop()
+ msg = queue.get()
msg_pb = pb.PlyMessage()
msg_pb.ParseFromString(msg)
--
Gitblit v1.9.1