|
import os, sys
|
import queue
|
import time
|
import pika
|
import threading
|
from .DustCleanerThread import DustCleanerThread
|
|
# 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 ControllThread(threading.Thread):
|
"""
|
controll thread
|
accept controll message from rabbitmq
|
"""
|
def __init__(self, host: str, user: str, password: str, callback_queue: str, config):
|
threading.Thread.__init__(self)
|
# setup dust cleaner command queue and thread
|
self.config = config
|
self._dust_cleaner_cmd_queue = queue.Queue()
|
self._dust_cleaner_thread = DustCleanerThread(self._dust_cleaner_cmd_queue, self.config)
|
self._dust_cleaner_thread.setDaemon(True)
|
self._dust_cleaner_thread.start()
|
|
self._init_callback(host, user, password, callback_queue)
|
|
def _def_routes(self):
|
"""define controll message routes"""
|
self.routes = {
|
pb.ControlSysCommand.DUST_CLEANER_COMMAND: self._on_dust_cleaner_command,
|
}
|
|
def _on_dust_cleaner_command(self, msg: pb.ControllMessage):
|
# add message to dust cleaner command queue
|
self._dust_cleaner_cmd_queue.put(msg)
|
return
|
|
def _default(self, req: pb.ControllMessage):
|
print('default controll handler')
|
return
|
|
def _init_callback(self, host: str, user: str, password: str, queue: str):
|
"""init rabbitmq channel"""
|
credentials = pika.PlainCredentials(user, password)
|
parameters = pika.ConnectionParameters(
|
credentials=credentials, host=host)
|
connection = pika.BlockingConnection(parameters)
|
self._channel = connection.channel()
|
self._channel.queue_declare(queue=queue, durable=False)
|
self._channel.basic_qos(prefetch_count=1)
|
self._channel.basic_consume(
|
on_message_callback=self._on_callback_request, queue=queue)
|
|
def _on_callback_request(self, ch, method, properties, body):
|
print('callback received')
|
msg = pb.ControllMessage()
|
msg.ParseFromString(body)
|
dispatcher = self.routes.get(msg.sys_command, self._default)
|
dispatcher(msg)
|
|
ch.basic_ack(delivery_tag=method.delivery_tag)
|
return
|
|
def stop(self):
|
"""stop controll thread"""
|
self._channel.stop_consuming()
|
self._dust_cleaner_thread.stop()
|
|
def run(self):
|
print("ControllThread started")
|
self._channel.start_consuming()
|
|
|
if __name__ == "__main__":
|
print("Begin CallbackThread Test")
|
cb_thread = ControllThread("localhost", "chai", "password123", "pod1")
|
cb_thread.start()
|
|
time.sleep(3)
|
|
cb_thread.stop()
|
