import os import threading import pika import time import os # 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 # 根据新的数据结构改写run class senderThread(threading.Thread): def __init__(self, MQparameters,input_msg_queue,sender_id,exchangeName,exchangeType): super(senderThread,self).__init__() self.parameters = MQparameters self.queue = input_msg_queue self.exchangeName = exchangeName self.exchangeType = exchangeType self.sender_id = sender_id self.is_stop = False 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: channel = self.connection.channel() channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType) while True: try: if not self.queue.empty(): data= self.queue.get() if isinstance(data, pb.CapturedData): msg = pb.DataMessage() else: body = msg.body routing_key = msg.routing_key channel.basic_publish(exchange=self.exchangeName, routing_key=routing_key, body=body) response_tag = f"{time.asctime()},{self.sender_id} send {self.exchangeName} message to {routing_key}" print(response_tag) elif self.queue.empty(): time.sleep(1) except Exception as e: print(e) break # 根据新的数据结构改写run class receiverThread(threading.Thread): def __init__(self, MQparameters,received_msg_queue,output_response_queue,exchangeName,exchangeType,receiver_id): super(receiverThread,self).__init__() self.parameters = MQparameters self.queue = received_msg_queue self.response_queue = output_response_queue self.exchangeName = exchangeName self.exchangeType = exchangeType self.receiver_id = receiver_id self.is_stop = False 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: channel = self.connection.channel() channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType) result = channel.queue_declare(queue='', exclusive=True, durable=True) queue_name = result.method.queue channel.queue_bind(exchange=self.exchangeName, queue=queue_name, routing_key=self.receiver_id) channel.basic_consume(queue=queue_name, \ on_message_callback=self.callback) channel.basic_qos(prefetch_count=1) channel.start_consuming() def callback(ch, method, properties, body): raise NotImplementedError( "Need to implement the callback function") # 根据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 # 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