# -*- coding: utf-8 -*- from json import loads, dumps from traceback import format_exc import paho.mqtt.client as mqtt import time from loguru import logger from bean.Feedback import upload_result, rtmp_result from enums.ExceptionEnum import ExceptionType, RtmpExceptionType from enums.StatusEnum import UploadTaskStatusType from util.RWUtils import getConfigs class MqttClient: __slots__ = ("client", "config", "msg_queue", "fb_queue", "res_topic") def __init__(self, base_dir, msg_queue, fb_queue): self.client = None self.config = getConfigs(base_dir, "config/mqtt.yml") self.res_topic = { "upload": self.config["topic"]["upload"]["res_topic"] % self.config["mqtt"]["airport_code"], "rtmp": self.config["topic"]["easy_rtmp_live"]["res_topic"] % self.config["mqtt"]["airport_code"] } self.msg_queue = msg_queue self.fb_queue = fb_queue def create_client(self): if self.client is None: try: client_id = self.config["mqtt"]["client_id"] airport_code = self.config["mqtt"]["airport_code"] self.client = mqtt.Client(client_id=client_id, clean_session=True) # self.client.enable_logger(logger=logging) self.client.disable_logger() self.client.on_connect = self.on_connect self.client.on_subscribe = self.on_subscribe self.client.on_message = self.on_message # self.client.on_disconnect = self.on_disconnect # self.client.on_log = self.on_log upload_topic = self.config["topic"]["upload"]["sub_topic"] % airport_code rtmp_topic = self.config["topic"]["easy_rtmp_live"]["sub_topic"] % airport_code self.client.user_data_set({ "sub_upload_topic": upload_topic, "sub_rtmp_topic": rtmp_topic, "msg_queue": self.msg_queue, "fb_queue": self.fb_queue, "topic": { upload_topic: "upload", rtmp_topic: "rtmp" } }) self.client.username_pw_set(self.config["mqtt"]["username"], self.config["mqtt"]["password"]) # self.client.will_set("willTopic", dumps({"message", "与服务器断开连接!!"}), 0, False) self.client.connect(host=self.config["mqtt"]["host"], port=self.config["mqtt"]["port"], keepalive=self.config["mqtt"]["keepalive"]) except Exception: if self.client: self.client.disconnect() self.client.loop_stop() self.client = None logger.error("创建mqtt客户端异常: {}", format_exc()) @staticmethod def on_connect(client, userdata, flags, rc): logger.info("mqtt客户端连接结果编号: {}， flags: {}", rc, flags) if 0 == rc: client.subscribe(userdata.get("sub_upload_topic"), 2) client.subscribe(userdata.get("sub_rtmp_topic"), 2) logger.info("mqtt客户端连接成功！sub_upload_topic: {}", userdata.get("sub_upload_topic")) logger.info("mqtt客户端连接成功！sub_rtmp_topic: {}", userdata.get("sub_rtmp_topic")) elif 1 == rc: logger.error("mqtt连接失败, 不正确的协议版本！rc:{}", rc) elif 2 == rc: logger.error("mqtt连接失败, 无效的客户端标识符！rc:{}", rc) elif 3 == rc: logger.error("mqtt连接失败, 服务器不可用！rc:{}", rc) elif 4 == rc: logger.error("mqtt连接失败, 错误的用户名或密码！rc:{}", rc) elif 5 == rc: logger.error("mqtt连接失败, 未授权！rc:{}", rc) else: logger.error("mqtt连接失败, 未知异常！rc:{}", rc) # 当代理响应订阅请求时被调用 @staticmethod def on_subscribe(client, userdata, mid, granted_qos): logger.info("mqtt开始订阅: {},{}！", mid, granted_qos) # @staticmethod # def on_disconnect(client, userdata, rc): # if rc != 0: # logger.info("mqtt端口连接！开始重新连接！！！！") # # 重新连接 # client.reconnect() @staticmethod def on_publish(client, obj, mid): logger.info("mqtt开始发布: {},{}！", mid, obj) # 当收到关于客户订阅的主题的消息时调用。 message是一个描述所有消息参数的MQTTMessage。 @staticmethod def on_message(client, userdata, msg): fb_queue, topic = userdata["fb_queue"], userdata["topic"] try: if msg is None or msg.payload is None or len(msg.payload) == 0: logger.error("mqtt监听到空消息！！！！！！！！！") return message = loads(msg.payload.decode()) logger.info("mqtt消息监听, topic:{}, message: {}！", msg.topic, message) if topic.get(msg.topic) is not None: msg_queue = userdata["msg_queue"] msg_queue.put((topic.get(msg.topic), message), timeout=2) except Exception: logger.error("mqtt消息监听异常, topic:{}, message: {}, 异常: {}！", msg.topic, msg.payload, format_exc()) if userdata.get("sub_upload_topic") == msg.topic: message = loads(msg.payload.decode()) upload_result(fb_queue, message.get("request_id"), errorCode=ExceptionType.ILLEGAL_PARAMETER_FORMAT.value[0], errorMsg=ExceptionType.ILLEGAL_PARAMETER_FORMAT.value[1], status=UploadTaskStatusType.FAILED.value[0], imageList=[], videoList=[]) if userdata.get("sub_rtmp_topic") == msg.topic: rtmp_result(fb_queue, RtmpExceptionType.ILLEGAL_PARAMETER_FORMAT.value[0], RtmpExceptionType.ILLEGAL_PARAMETER_FORMAT.value[1]) @staticmethod def on_log(client, obj, level, string): logger.info("当前日志信息, level: {}, msg: {}！", level, string) def start(self): try: self.create_client() self.client.loop_start() except Exception as e: logger.error("mqtt启动异常: {}", format_exc()) raise e def stop(self): if self.client: self.client.disconnect() self.client.loop_stop() def publish(self, result): res_topic = self.res_topic retry_count = 0 while True: try: logger.info("mqtt发布信, 消息体: {}", result) self.client.publish(res_topic.get(result[0]), payload=dumps(result[1]), qos=2) break except Exception as e: retry_count += 1 time.sleep(1) logger.error("mqtt发布信息失败, 消息体: {}, 异常: {}", dumps(result), format_exc()) if retry_count > 3: raise e del result # if __name__ == "__main__": # mq = MqttClient(r"D:\tuoheng\codenew\tuoheng_push_stream") # mq.start() # i = 0 # while True: # time.sleep(1) # mq.publish({"tip": "1111111111111111111111"}) # print('state: ', mq.client._state, 'loop进程：', mq. client._thread) # print(mq.client.is_connected())