import os
|
import queue
|
import threading
|
import pika
|
import time
|
from time import sleep
|
import sys
|
from datetime import datetime
|
from typing import List
|
|
# # add python path of src to sys.path
|
# src_path = os.path.join(__file__, *(['..'] * 2))
|
# src_path = os.path.abspath(src_path)
|
# sys.path.insert(0, src_path)
|
|
from ..protos import aerial_pb2 as pb
|
|
|
|
# class Message(object):
|
# def __init__(self,sender_id,sender_key,body,routing_key):
|
# self.sender_id = sender_id
|
# self.sender_key = sender_key
|
# self.body = body
|
# self.routing_key = routing_key
|
|
class senderThread(threading.Thread):
|
|
def __init__(self,
|
MQparams: pika.ConnectionParameters,
|
exchange_name: str,
|
exchange_type:str,
|
# routing_key:str,
|
message_queue:queue.Queue,
|
):
|
super(senderThread,self).__init__()
|
|
self.MQparams = MQparams
|
self.exchange_name = exchange_name
|
self.exchange_type = exchange_type
|
# self.routing_key = routing_key
|
self.message_queue = message_queue
|
|
# connect to RabbitMQ
|
self.is_stop = False
|
self.connection = pika.BlockingConnection(self.MQparams)
|
|
def stop(self):
|
"""stop the thread"""
|
self.is_stop = True
|
self.connection.close()
|
|
def run(self):
|
|
while not self.is_stop:
|
# declare a exchange
|
channel = self.connection.channel()
|
channel.exchange_declare(exchange=self.exchange_name,
|
exchange_type=self.exchange_type)
|
|
while True:
|
try:
|
if not self.message_queue.empty():
|
|
# decode the routing key and message body from tuple list
|
message = self.message_queue.get()
|
msg_routing_key = message[0]
|
msg_body = message[1]
|
|
# publish a PERSISTENT message
|
channel.basic_publish(
|
exchange = self.exchange_name,
|
routing_key = msg_routing_key,
|
body = msg_body,
|
properties=pika.BasicProperties(
|
delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
|
)
|
)
|
|
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]")
|
print(caption)
|
|
else:
|
time.sleep(1)
|
|
except Exception as e:
|
print(e)
|
break
|
|
class receiverThread(threading.Thread):
|
def __init__(self,
|
MQparams: pika.ConnectionParameters,
|
exchange_name: str,
|
exchange_type:str,
|
MQqueue_name:str,
|
binding_keys:List(str),
|
message_queue:queue.Queue,
|
):
|
super(senderThread,self).__init__()
|
|
self.MQparams = MQparams
|
self.exchange_name = exchange_name
|
self.exchange_type = exchange_type
|
self.MQqueue_name = MQqueue_name
|
self.binding_keys = binding_keys
|
self.message_queue = message_queue
|
|
# connect to RabbitMQ
|
self.is_stop = False
|
self.connection = pika.BlockingConnection(self.MQparams)
|
|
def stop(self):
|
"""stop the thread"""
|
self.is_stop = True
|
self.connection.close()
|
|
|
def run(self):
|
|
while not self.is_stop:
|
# declare a exchange
|
channel = self.connection.channel()
|
channel.exchange_declare(exchange=self.exchange_name,
|
exchange_type=self.exchange_type)
|
|
# declare the queue and bind it with keys
|
channel.queue_declare(queue=self.MQqueue_name, durable=True)
|
for binding_key in self.binding_keys:
|
channel.queue_bind(exchange=self.exchangeName, queue=self.MQqueue_name, routing_key=binding_key)
|
|
# consume the message
|
channel.basic_qos(prefetch_count=1)
|
channel.basic_consume(
|
queue=self.MQqueue_name,
|
on_message_callback=self.callback
|
)
|
channel.start_consuming()
|
|
def callback(self,ch, method, properties, body):
|
self.message_queue.put(body)
|
ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
class decodeThread(threading.Thread):
|
def __init__(self,
|
queue_list:List(queue.Queue),
|
queue_tags:List(str),
|
save_path:str):
|
super().__init__()
|
|
self.queue_list = queue_list
|
self.queue_tags = queue_tags
|
self.queue_num = 0
|
self.save_path = save_path
|
self.is_stop = False
|
|
assert len(self.queue_list) == len(self.queue_tags)
|
|
|
def stop(self):
|
"""stop the thread"""
|
self.is_stop = True
|
|
def run(self) -> None:
|
|
# creating a path for saving data
|
time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
self.save_root_folder = os.path.join(self.save_path,time_tag)
|
os.makedirs(self.save_root_folder)
|
|
self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg")
|
self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg")
|
self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg")
|
self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg")
|
|
sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder,
|
self.plysmsg_folder,self.commandsmsg_folder]
|
|
for folder in sub_folder_list:
|
os.mkdir(folder)
|
print(f"[NOTICE] Creating folder: {folder}")
|
|
# process the queue list
|
self.queue_num = len(self.queue_list)
|
while self.queue_num:
|
i = 0
|
while True:
|
# select a queue
|
i_tag = i % self.queue_num
|
queue = self.queue_list[i_tag]
|
|
# make sure the queue is not empty
|
while not queue.empty():
|
# decode the msg using cooresponding method
|
queue_tag = self.queue_tags[i_tag]
|
decode_method = self.decode_select(queue_tag)
|
decode_method(queue)
|
# next queue
|
i += 1
|
|
|
def decode_select(self,queue_tags:str):
|
route_dic = {
|
"Logs":self.decode_LogsMsg,
|
"Images":self.decode_ImagesMsg,
|
"Plys":self.decode_PlysMsg,
|
"Commands":self.decode_CommandsMsg
|
}
|
return route_dic[queue_tags]
|
|
def decode_LogsMsg(self,queue):
|
nums_limit = 10
|
count = 0
|
while not queue.empty() and count < nums_limit:
|
|
# extracting the msg from queue
|
msg = queue.pop()
|
msg_pb = pb.LogsMessage()
|
msg_pb.ParseFromString(msg)
|
|
# save the raw pb msg
|
raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg")
|
with open(raw_file_name, 'wb') as file:
|
file.write(msg)
|
|
# process the msg
|
msg_source = msg_pb.source
|
msg_timestamp = msg_pb.timestamp
|
msg_context = msg_pb.context
|
|
# display the info to the console
|
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
print(f"[x] {curr_time}|(From {msg_source}): {msg_context}" )
|
|
# next msg
|
count += 1
|
|
|
def decode_ImagesMsg(self,queue):
|
|
while not queue.empty():
|
|
# extracting the msg from queue
|
msg = queue.pop()
|
msg_pb = pb.ImagesMessage()
|
msg_pb.ParseFromString(msg)
|
|
# save the raw pb msg
|
raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg")
|
with open(raw_file_name, 'wb') as file:
|
file.write(msg)
|
|
# process the msg
|
msg_source = msg_pb.source
|
msg_timestamp = msg_pb.timestamp
|
|
# process the images
|
id = 0
|
for image in msg_pb.Image:
|
imagelegth = image.length
|
image_byte = image.data
|
|
# save the images
|
if (image_byte is None) | (len(image_byte) == 0):
|
pass
|
else:
|
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)
|
|
# display the info to the console
|
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
print(f"[x] {curr_time}|(From {msg_source}): image{id} saved!" )
|
|
# next image item
|
id += 1
|
|
|
def decode_PlysMsg(self,queue):
|
|
|
|
|
pass
|
|
def decode_CommandsMsg(self,queue):
|
|
pass
|
|
|
|
# 根据rabbitmq重新设置消息参数
|
class device_state_senderThread(senderThread):
|
def __init__(self,
|
MQparameters,
|
input_msg_queue,
|
sender_id,
|
):
|
super(device_state_senderThread,self).__init__(
|
MQparameters, \
|
input_msg_queue,
|
sender_id,
|
"device_state",
|
"fanout"
|
)
|
# 根据rabbitmq重新设置消息参数
|
class device_state_receiverThread(receiverThread):
|
def __init__(self,
|
MQparameters,
|
receiver_id):
|
super(device_state_receiverThread,self).__init__(
|
MQparameters,
|
None,
|
None,
|
"device_state",
|
"fanout",
|
receiver_id
|
)
|
|
def callback(self, ch, method, properties, body):
|
|
response_tag = f"{time.asctime()} received: {body}."
|
print(response_tag)
|
ch.basic_ack(delivery_tag = method.delivery_tag)
|
# 根据rabbitmq重新设置消息参数
|
class alarm_state_senderThread(senderThread):
|
def __init__(self,
|
MQparameters,
|
input_msg_queue,
|
sender_id,
|
):
|
super(alarm_state_senderThread,self).__init__(
|
MQparameters, \
|
input_msg_queue,
|
sender_id,
|
"alarm_state",
|
"fanout"
|
)
|
# 根据rabbitmq重新设置消息参数
|
class alarm_state_receiverThread(receiverThread):
|
def __init__(self,
|
MQparameters,
|
receiver_id):
|
super(alarm_state_receiverThread,self).__init__(
|
MQparameters,
|
None,
|
None,
|
"alarm_state",
|
"fanout",
|
receiver_id
|
)
|
|
def callback(self, ch, method, properties, body):
|
|
response_tag = f"{time.asctime()} received: {body}."
|
print(response_tag)
|
ch.basic_ack(delivery_tag = method.delivery_tag)
|
# 根据rabbitmq重新设置消息参数
|
class command_request_senderThread(senderThread):
|
def __init__(self,
|
MQparameters,
|
input_msg_queue,
|
sender_id,
|
):
|
super(command_request_senderThread,self).__init__(
|
MQparameters, \
|
input_msg_queue,
|
sender_id,
|
"command_request",
|
"topic"
|
)
|
# 根据rabbitmq重新设置消息参数
|
class command_request_receiverThread(receiverThread):
|
def __init__(self,
|
MQparameters,
|
received_msg_queue,
|
response_msg_queue,
|
receiver_id):
|
super(command_request_receiverThread,self).__init__(
|
MQparameters,
|
received_msg_queue,
|
response_msg_queue,
|
"command_request",
|
"topic",
|
receiver_id
|
)
|
|
def callback(self, ch, method, properties, body):
|
received_msg = Message(None,None,body,None)
|
self.queue.put(received_msg)
|
|
response_tag = f"{time.asctime()},{self.receiver_id} received {self.exchangeName} message."
|
print(response_tag)
|
self.response_queue.put(Message(self.receiver_id,self.receiver_id,response_tag,''))
|
ch.basic_ack(delivery_tag = method.delivery_tag)
|
# 根据rabbitmq重新设置消息参数
|
class command_response_senderThread(senderThread):
|
def __init__(self,
|
MQparameters,
|
input_msg_queue,
|
sender_id,
|
):
|
super(command_response_senderThread,self).__init__(
|
MQparameters, \
|
input_msg_queue,
|
sender_id,
|
"command_response",
|
"fanout"
|
)
|
# 根据rabbitmq重新设置消息参数
|
class command_response_receiverThread(receiverThread):
|
def __init__(self,
|
MQparameters,
|
receiver_id):
|
super(command_response_receiverThread,self).__init__(
|
MQparameters,
|
None,
|
None,
|
"command_response",
|
"fanout",
|
receiver_id
|
)
|
|
def callback(self, ch, method, properties, body):
|
response_tag = f"{time.asctime()} received: {body}."
|
print(response_tag)
|
ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
|
# TODO
|
class image_data_senderThread(senderThread):
|
def __init__(self,
|
MQparameters,
|
input_msg_queue,
|
sender_id,
|
):
|
super(command_response_senderThread,self).__init__(
|
MQparameters, \
|
input_msg_queue,
|
sender_id,
|
"image_data",
|
"fanout"
|
)
|
|
class image_data_receiverThread(receiverThread):
|
pass
|
|
|
def message_sender(MQparams: pika.ConnectionParameters,
|
exchange_name: str,
|
exchange_type:str,
|
routing_key:str,
|
message:str,
|
):
|
# connect to RabbitMQ
|
connection = pika.BlockingConnection(MQparams)
|
|
# declare a exchange
|
channel = connection.channel()
|
channel.exchange_declare(exchange=exchange_name,
|
exchange_type=exchange_type)
|
# publish a PERSISTENT message
|
channel.basic_publish(
|
exchange=exchange_name,
|
routing_key=routing_key,
|
body=message,
|
properties=pika.BasicProperties(
|
delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
|
)
|
)
|
# close the connection
|
connection.close()
|
|
def message_receiver(MQparams: pika.ConnectionParameters,
|
exchange_name: str,
|
exchange_type:str,
|
queue_name:str,
|
binding_keys:List[str],
|
callback_fn:callable
|
):
|
# connect to RabbitMQ
|
connection = pika.BlockingConnection(MQparams)
|
|
# declare a exchange
|
channel = connection.channel()
|
channel.exchange_declare(exchange=exchange_name,
|
exchange_type=exchange_type)
|
|
# declare a durable queue
|
channel.queue_declare(queue=queue_name,durable=True)
|
|
# literate the binding keys and bind the queue
|
for binding_key in binding_keys:
|
channel.queue_bind(queue=queue_name,exchange=exchange_name,
|
routing_key=binding_key)
|
|
# waiting for consuming the messages
|
channel.basic_qos(prefetch_count=1)
|
channel.basic_consume(queue=queue_name,
|
auto_ack=False,
|
on_message_callback=callback_fn
|
)
|
channel.start_consuming()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# class device_state_up_msg_sender_thread(threading.Thread):
|
# def __init__(self, MQparameters,device_state_msg_queue):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.queue = device_state_msg_queue
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("device_state_up_msg_sender_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
|
|
# while True:
|
# try:
|
# if not self.queue.empty():
|
# msg = self.msg_queue.get()
|
# channel.basic_publish(exchange='device_state', routing_key='', body=msg)
|
# print(time.asctime(), "send device state message")
|
# elif self.queue.empty():
|
# time.sleep(1)
|
# except Exception as e:
|
# break
|
|
# class device_state_up_msg_receiver_thread(threading.Thread):
|
# def __init__(self, MQparameters):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
# while not self.is_stop:
|
# print("device_state_up_msg_receiver_thread started")
|
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
|
|
# result = channel.queue_declare(queue='', exclusive=True, durable=True)
|
# queue_name = result.method.queue
|
|
# channel.queue_bind(exchange='device_state', queue=queue_name)
|
|
# def callback(ch, method, properties, body):
|
# print(f"Received: {body}")
|
# ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
# channel.basic_consume(queue=queue_name, \
|
# on_message_callback=callback)
|
|
# channel.basic_qos(prefetch_count=1)
|
# channel.start_consuming()
|
|
# class alarm_state_up_msg_sender_thread(threading.Thread):
|
|
# def __init__(self, MQparameters,alarm_state_msg_queue):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.queue = alarm_state_msg_queue
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("alarm_state_up_msg_sender_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
|
|
# while True:
|
# try:
|
# if not self.queue.empty():
|
# msg = self.msg_queue.get()
|
# channel.basic_publish(exchange='alarm_state', routing_key='', body=msg)
|
# print(time.asctime(), "send alarm state message")
|
# elif self.queue.empty():
|
# time.sleep(1)
|
# except Exception as e:
|
# break
|
|
# class alarm_state_up_msg_receiver_thread(threading.Thread):
|
# def __init__(self, MQparameters):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("alarm_state_up_msg_receiver_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
|
|
# result = channel.queue_declare(queue='', exclusive=True, durable=True)
|
# queue_name = result.method.queue
|
|
# channel.queue_bind(exchange='alarm_state', queue=queue_name)
|
|
# def callback(ch, method, properties, body):
|
# print(f"Received: {body}")
|
# ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
# channel.basic_consume(queue=queue_name, \
|
# on_message_callback=callback)
|
|
# channel.basic_qos(prefetch_count=1)
|
# channel.start_consuming()
|
|
# class command_request_msg_sender_thread(threading.Thread):
|
|
# def __init__(self, MQparameters,command_request_msg_queue,routing_key_queue):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.command_queue = command_request_msg_queue
|
# self.key_queue = routing_key_queue
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("command_request_msg_sender_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
|
|
# while True:
|
# try:
|
# if not self.command_queue.empty() and \
|
# not self.key_queue.empty():
|
# msg = self.command_queue.get()
|
# routing_key = self.key_queue.get()
|
# corr_id = str(uuid.uuid4())
|
# channel.basic_publish(exchange='command_msg',
|
# routing_key = routing_key,
|
# body=msg,
|
# properties=pika.BasicProperties(
|
# correlation_id=corr_id
|
# )
|
# )
|
# print(time.asctime(), f"send command request with ID:[{corr_id}]")
|
# elif self.queue.empty():
|
# time.sleep(1)
|
# except Exception as e:
|
# break
|
|
# class command_request_msg_receiver_thread(threading.Thread):
|
# def __init__(self, MQparameters,command_request_msg_queue, \
|
# command_response_msg_queue, routing_key):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.queue = command_request_msg_queue
|
# self.response_queue = command_response_msg_queue
|
# self.routing_key = routing_key
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("command_request_msg_receiver_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
|
|
# result = channel.queue_declare(queue='', exclusive=True, durable=True)
|
# queue_name = result.method.queue
|
|
# channel.queue_bind(exchange='command_msg', queue=queue_name, \
|
# routing_key=self.routing_key)
|
|
# def callback(ch, method, prop, body):
|
# print(time.asctime(), f"reveive command request with ID:[{prop.corr_id}]")
|
# self.queue.put(body)
|
# self.response_queue.put("")
|
# ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
# channel.basic_consume(queue=queue_name, \
|
# on_message_callback=callback)
|
|
# channel.basic_qos(prefetch_count=1)
|
# channel.start_consuming()
|
|
# class command_response_msg_sender_thread(threading.Thread):
|
# def __init__(self, MQparameters,command_request_msg_queue):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.queue = command_request_msg_queue
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# print("command_response_msg_sender_thread started")
|
# channel = self.connection.channel()
|
# channel.exchange_declare(exchange='command_response', exchange_type='fanout')
|
|
# while True:
|
# try:
|
# if not self.queue.empty():
|
# msg = self.msg_queue.get()
|
# channel.basic_publish(exchange='command_response', routing_key='', body=msg)
|
# print(time.asctime(), "send device state message")
|
# elif self.queue.empty():
|
# time.sleep(1)
|
# except Exception as e:
|
# break
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# class command_down_rpc_sender_thread(threading.Thread):
|
# def __init__(self, MQparameters, command_queue, routing_key_queue):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.command_queue = command_queue
|
# self.key_queue = routing_key_queue
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self):
|
|
# while not self.is_stop:
|
# # print("command_down_rpc_sender_thread started")
|
|
# channel = self.connection.channel()
|
|
# # declare a exchange
|
# channel.exchange_declare(exchange="command_rpc", \
|
# exchange_type='topic')
|
|
# # declare a callback queue
|
# result = channel.queue_declare(queue='', exclusive=True)
|
# callback_queue = result.method.queue
|
|
# # bingding the queue
|
# channel.queue_bind(
|
# exchange="command_rpc",
|
# queue=callback_queue,
|
# routing_key=routing_key
|
# )
|
|
# # init the command uuid
|
# corr_id = str(uuid.uuid4())
|
# command_response = None
|
|
# # def a function to print the command respond message
|
# def respond_callback(ch, method, properties, body):
|
# if properties.correlation_id == corr_id:
|
# command_response = body
|
# print(body)
|
|
# # start the cosumer
|
# channel.basic_consume(
|
# queue=callback_queue,
|
# on_message_callback=respond_callback,
|
# auto_ack=True
|
# )
|
|
# # publish with tropic exchange
|
# channel.basic_publish(
|
# exchange="command_rpc",
|
# routing_key=routing_key,
|
# body = request_command,
|
# properties=pika.BasicProperties(
|
# reply_to=callback_queue,
|
# correlation_id=corr_id
|
# ),
|
# )
|
# with command_response is None:
|
# self.connection.process_data_events()
|
|
# class command_down_rpc_receiver_thread(threading.Thread):
|
# def __init__(self, MQparameters):
|
# super().__init__(self)
|
# self.parameters = MQparameters
|
# self.connection = pika.BlockingConnection(self.parameters)
|
|
# def stop(self):
|
# """stop the thread"""
|
# self.is_stop = True
|
# self.connection.close()
|
|
# def run(self,commands_queue):
|
# while not self.is_stop:
|
# print("command_down_rpc_receiver_thread started")
|
# channel = self.connection.channel()
|
|
# # declare a exchange
|
# channel.exchange_declare(exchange="command_rpc", \
|
# exchange_type='topic')
|
|
# # declare a callback queue
|
# result = channel.queue_declare(queue='', exclusive=True)
|
# callback_queue = result.method.queue
|
|
# # bingding the queue
|
# channel.queue_bind(
|
# exchange="command_rpc",
|
# queue=callback_queue,
|
# routing_key='pod*'
|
# )
|
|
# # def a function to reply the request from the sender
|
# def on_request(ch, method, prop, body):
|
# commands_queue.put(body)
|
|
# response = f"{time.asctime()} {self.device_id} receive the command uuid:[{prop.correlation_id}]"
|
# print(response)
|
|
# ch.basic_publish(exchange="command_rpc",
|
# routing_key = prop.reply_to,
|
# properties = pika.BasicProperties(
|
# correlation_id=prop.corr_id
|
# ),
|
# body = response)
|
# ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
# # start the cosumer
|
# channel.basic_qos(prefetch_count=1)
|
# channel.basic_consume(
|
# queue=callback_queue,
|
# on_message_callback=on_request,
|
# auto_ack=True
|
# )
|
# channel.start_consuming()
|
|
# class image_data_up_msg_sender_thread(threading.Thread):
|
# pass
|
|
# class image_data_up_msg_receiver_thread(threading.Thread):
|
# pass
|
|
# class ply_data_up_msg_sender_thread(threading.Thread):
|
# pass
|
|
# class ply_data_up_msg_receiver_thread(threading.Thread):
|
# pass
|
