import threading
|
import pika
|
import time
|
|
|
|
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, 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():
|
msg = self.msg_queue.get()
|
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:
|
break
|
|
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")
|
|
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"
|
)
|
|
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)
|
|
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"
|
)
|
|
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)
|
|
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"
|
)
|
|
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)
|
|
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"
|
)
|
|
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)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 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
|
