From e3179d36a8aabce325f91c4668405ccc232337d1 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 1 Feb 2019 17:19:07 -0500 Subject: [PATCH] Change Confserver to AIOHTTP Change Confserver to AIOHTTP Better threading and performance across --- Pipfile | 1 + bumper.py | 12 +- bumper/confserver.py | 444 +++++++++++++++++++------------------------ bumper/mqttserver.py | 70 ++++--- bumper/xmppserver.py | 2 + 5 files changed, 255 insertions(+), 274 deletions(-) diff --git a/Pipfile b/Pipfile index fff0c8b..db09990 100644 --- a/Pipfile +++ b/Pipfile @@ -6,5 +6,6 @@ name = "pypi" [packages] hbmqtt = "*" paho-mqtt = "*" +aiohttp = "*" [dev-packages] diff --git a/bumper.py b/bumper.py index 5f004d7..6c086c9 100644 --- a/bumper.py +++ b/bumper.py @@ -5,8 +5,15 @@ import bumper import sys, socket import time -logging.basicConfig(level=logging.INFO, - format='%(asctime)s %(levelname)-8s %(message)s') + +args = sys.argv +if len(args) > 0: + if '--debug' in args: + logging.basicConfig(level=logging.DEBUG, + format='%(asctime)s %(levelname)-8s %(message)s') + else: + logging.basicConfig(level=logging.INFO, + format='%(asctime)s %(levelname)-8s %(message)s') #conf_address = (socket.gethostbyname(socket.gethostname()), 443) conf_address = ("0.0.0.0", 443) @@ -32,4 +39,3 @@ conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True, bumpe # start xmpp server (sync) xmpp_server = bumper.XMPPServer(xmpp_address) -conf_server.disconnect() diff --git a/bumper/confserver.py b/bumper/confserver.py index 78e11c7..956af3f 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -1,270 +1,222 @@ #!/usr/bin/env python3 -from http.server import HTTPServer -from http.server import BaseHTTPRequestHandler -from http import HTTPStatus from threading import Thread -import socket, logging, ssl, json, sys -import asyncio -import contextvars +import socket, logging, ssl, json import string import random import bumper import time from datetime import datetime, timedelta -import threading -from socketserver import ThreadingMixIn - -class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): - """Handle requests in a separate thread.""" - -class RequestHandler(BaseHTTPRequestHandler): - bumper_clients = contextvars.ContextVar - helperbot = object - - def do_POST(self): - try: - self.protocol_version = 'HTTP/1.1' - content_length = int(self.headers['Content-Length']) - post_data = self.rfile.read(content_length) - logging.debug("Headers: " + str(self.headers)) - body = {} - if "app_logs" in str(self.path): - body = {} - - else: - request_body = post_data.decode('utf-8') - logging.debug("Request: " + request_body) - json_body = json.loads(request_body) - - if "/product/getProductIotMap" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("getProductIotMap Path Split: %s" %psplit) - body = {"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]} - elif "notify_engine.do" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("notify_engine Path Split: %s" %psplit) - body = {"ret":"ok"} - elif "iot/devmanager.do" in str(self.path): #Handle rest commands - psplit = str(self.path).split("/") - logging.debug("devmanager Path Split: %s" %psplit) - randomid = ''.join(random.sample(string.ascii_letters,6)) - retcmd = self.helperbot.send_command(json_body, randomid) - body = retcmd - elif "/users/user.do" in str(self.path): #Handle user commands - psplit = str(self.path).split("/") - logging.debug("userdo Path Split: %s" %psplit) - todo = json_body['todo'] - if todo == 'FindBest': - service = json_body['service'] - if service == 'EcoMsgNew': - body = {"result":"ok","ip":socket.gethostbyname(socket.gethostname()),"port":5223} - elif service == 'EcoUpdate': - body = {"result":"ok","ip":"47.88.66.164","port":8005} - elif todo == 'loginByItToken': - psplit = str(self.path).split("/") - logging.debug("LoginByItToken Split: %s" %psplit) - body = { - "resource": json_body["resource"], - "result": "ok", - "todo": "result", - "token": json_body["token"], #RandomChar(32) - "userId": json_body["userId"] #RandomChar(16) - } - #TODO: Randomize - #self.send_header("ETag", "ETag1234") - elif todo == 'GetDeviceList': - active_bots = self.bumper_clients.get() - body = { - "devices": active_bots, - "result": "ok", - "todo": "result" - } - - elif "ESTInt" in str(self.path): #Handle user commands: - psplit = str(self.path).split("/") - logging.debug("bigdata Path Split: %s" %psplit) - {"header":{"result_code":"000000"},"body":{}} - else: - psplit = str(self.path).split("/") - logging.debug("Other Post Path Split: %s" %psplit) - body = {} - - body = json.dumps(body) - logging.debug("Response: " + str(body)) - body = body.encode(encoding='UTF-8') - self.send_response(HTTPStatus.OK) - self.send_header('Content-Type', 'application/json; charset=utf-8') - self.send_header('Connection', 'keep-alive') - self.send_header('Content-Length', len(body)) - self.end_headers() - - self.wfile.write(body) - except Exception as e: - logging.error('ConfServer POST: {}'.format(e)) - - def do_GET(self): - try: - self.protocol_version = 'HTTP/1.1' - logging.debug("Path: " + self.path) - - if "/user/login?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("Login Path Split: %s" %psplit) - #Could implement basic auth if you wanted, or just accept anything - body = { - "code": "0000", - "data": { - "accessToken": "tempaccesstoken", #Random chars 32 length - "country": psplit[3], - "email": "null@null.com", - "uid": "fuid_1", #Date(14)_RandomChars(32) - "username": "fusername_1" #Random chars 8 - }, - "msg": "操作成功", - "time": bumper.current_milli_time() - } - - elif "/user/getAuthCode?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("getAuthCode Path Split: %s" %psplit) - body = { - "code": "0000", - "data": { - "authCode": "{}_tempauthcode".format(psplit[3]), #countrycode_randomchars(32) - "ecovacsUid": "fuid_1" #Date(14)_RandomChars(32) - }, - "msg": "操作成功", - "time": bumper.current_milli_time() - } - - elif "/common/checkVersion?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("checkVersion Path Split: %s" %psplit) - body = { - "code": "0000", - "data": { - "c": None, - "img": None, - "r": 0, - "t": None, - "u": None, - "ut": 0, - "v": None - }, - "msg": "操作成功", - "time": bumper.current_milli_time() - } - - - elif "/user/checkAgreement?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("checkAgreement Path Split: %s" %psplit) - body = { - "code": "0000", - "data": [], - "msg": "操作成功", - "time": bumper.current_milli_time() - - } - elif "/campaign/homePageAlert?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("homePageAlert Path Split: %s" %psplit) - nextAlert = bumper.get_milli_time((datetime.now() + timedelta(hours=12)).timestamp()) - logging.debug("Next Alert %s" % nextAlert) - body = { - "code": "0000", - "data": { - "clickSchemeUrl": None, - "clickWebUrl": None, - "hasCampaign": "N", - "imageUrl": None, - "nextAlertTime": nextAlert, - "serverTime": bumper.current_milli_time() - }, - "msg": "操作成功", - "time": bumper.current_milli_time() - } - - - elif "/user/logout?" in str(self.path): - psplit = str(self.path).split("/") - logging.debug("Logout Path Split: %s" %psplit) - body = {"code": "0000","data": None,"msg": "操作成功", "time": bumper.current_milli_time()} - else: - psplit = str(self.path).split("/") - logging.debug("Other Get Path Split: %s" %psplit) - body = {} - - body = json.dumps(body) - logging.debug("Response: " + str(body)) - body = body.encode(encoding='UTF-8') - self.send_response(HTTPStatus.OK) - self.send_header('Content-Type', 'application/json; charset=utf-8') - self.send_header('Connection', 'keep-alive') - self.send_header('Content-Length', len(body)) - self.end_headers() - self.wfile.write(body) - except Exception as e: - logging.error('ConfServer GET: {}'.format(e)) - - -class HTTPServerThread(ThreadedHTTPServer, Thread): - bumper_clients = contextvars.ContextVar - def __init__(self, server_address, bumper_clients, helperbot): - Thread.__init__(self) - self.server_address = server_address - self.bumper_clients = bumper_clients - self.helperbot = helperbot - self.handler = RequestHandler - self.handler.bumper_clients = self.bumper_clients - self.handler.helperbot = self.helperbot - - self.exit_flag = False - - ThreadedHTTPServer.__init__(self, self.server_address, self.handler) - - def handle_error(self, request, client_address): - self.close_request(request) - def run(self): - try: - logging.info('ConfServer: listening on {}:{}'.format(self.server_address[0], self.server_address[1])) - while not self.exit_flag: - self.handle_request() - except Exception as e: - logging.error('ConfServer: {}'.format(e)) - def disconnect(self): - self.exit_flag = True - # make a connection to - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.connect(self.server_address) - s.close() - +import asyncio +import contextvars +from aiohttp import web class ConfServer(): bumper_clients = contextvars.ContextVar - + def __init__(self, address, usessl=False, run_async=True, bumper_clients=contextvars.ContextVar, helperbot=None): - self.run_async = run_async - self.server = HTTPServerThread(address, bumper_clients, helperbot) + self.bumper_clients = bumper_clients + self.helperbot = helperbot + self.usessl = usessl + self.run_async = run_async + self.address = address - try: - if usessl: - self.server.socket = ssl.wrap_socket(self.server.socket, keyfile='./certs/key.pem', certfile='./certs/cert.pem', server_side=True) - if self.run_async: - self.server.start() + + try: + if run_async: + logging.debug("Starting ConfServer Thread: 1") + confserver = Thread(name="ConfServer_Thread",target=self.run_server) + self.server = confserver + confserver.setDaemon(True) + confserver.start() else: try: - self.server.run() + self.run_server() except KeyboardInterrupt: self.disconnect() except Exception as e: logging.error('ConfServer: {}'.format(e)) + + def run_server(self): + try: + loop = asyncio.get_event_loop() + except: + loop = asyncio.new_event_loop() + + loop.run_until_complete(self.start_server()) + loop.run_forever() + + + async def start_server(self): + #formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" + #logging.basicConfig(level=logging.INFO, format=formatter) + app = web.Application() + + app.add_routes([ + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/login', self.handle_login), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout', self.handle_login), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode', self.handle_getAuthCode), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreement', self.handle_checkAgreement), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/getAuthCode', self.handle_checkVersion), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/campaign/homePageAlert', self.handle_homePageAlert), + + web.post('/api/users/user.do', self.handle_usersapi), + web.post('/api/pim/product/getProductIotMap', self.handle_getProductIotMap), + web.post('/api/iot/devmanager.do', self.handle_devmanager) + ]) + + + runner = web.AppRunner(app, access_log=None) + await runner.setup() + + if self.usessl: + ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ssl_ctx.load_cert_chain('./certs/cert.pem', './certs/key.pem') + site = web.TCPSite(runner, host=self.address[0], port=self.address[1],ssl_context=ssl_ctx) + + else: + site = web.TCPSite(runner, host=self.address[0], port=self.address[1]) + + await site.start() + + + async def handle_login(self, request): + #Could implement basic auth if you wanted, or just accept anything + countrycode = request.match_info.get('country', "us") + body = { + "code": "0000", + "data": { + "accessToken": "tempaccesstoken", #Random chars 32 length + "country": countrycode, + "email": "null@null.com", + "uid": "fuid_1", #Date(14)_RandomChars(32) + "username": "fusername_1" #Random chars 8 + }, + "msg": "操作成功", + "time": bumper.current_milli_time() + } + + return web.json_response(body) + + async def handle_logout(self, request): + body = {"code": "0000","data": None,"msg": "操作成功", "time": bumper.current_milli_time()} + + return web.json_response(body) + + + + + async def handle_getAuthCode(self, request): + #Could implement basic auth if you wanted, or just accept anything + countrycode = request.match_info.get('country', "us") + body = { + "code": "0000", + "data": { + "authCode": "{}_tempauthcode".format(countrycode), #countrycode_randomchars(32) + "ecovacsUid": "fuid_1" #Date(14)_RandomChars(32) + }, + "msg": "操作成功", + "time": bumper.current_milli_time() + } + + return web.json_response(body) + + async def handle_checkVersion(self, request): + body = { + "code": "0000", + "data": { + "c": None, + "img": None, + "r": 0, + "t": None, + "u": None, + "ut": 0, + "v": None + }, + "msg": "操作成功", + "time": bumper.current_milli_time() + } + + return web.json_response(body) + + async def handle_checkAgreement(self, request): + body = { + "code": "0000", + "data": [], + "msg": "操作成功", + "time": bumper.current_milli_time() + + } + + return web.json_response(body) + + async def handle_homePageAlert(self, request): + nextAlert = bumper.get_milli_time((datetime.now() + timedelta(hours=12)).timestamp()) + + body = { + "code": "0000", + "data": { + "clickSchemeUrl": None, + "clickWebUrl": None, + "hasCampaign": "N", + "imageUrl": None, + "nextAlertTime": nextAlert, + "serverTime": bumper.current_milli_time() + }, + "msg": "操作成功", + "time": bumper.current_milli_time() + } + + return web.json_response(body) + + async def handle_getProductIotMap(self, request): + #json_body = json.loads(await request.text()) + body = {"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]} + return web.json_response(body) + + async def handle_usersapi(self, request): + #Could implement basic auth if you wanted, or just accept anything + json_body = json.loads(await request.text()) + todo = json_body['todo'] + if todo == 'FindBest': + service = json_body['service'] + if service == 'EcoMsgNew': + body = {"result":"ok","ip":socket.gethostbyname(socket.gethostname()),"port":5223} + elif service == 'EcoUpdate': + body = {"result":"ok","ip":"47.88.66.164","port":8005} + elif todo == 'loginByItToken': + body = { + "resource": json_body["resource"], + "result": "ok", + "todo": "result", + "token": json_body["token"], #RandomChar(32) + "userId": json_body["userId"] #RandomChar(16) + } + elif todo == 'GetDeviceList': + active_bots = self.bumper_clients.get() + body = { + "devices": active_bots, + "result": "ok", + "todo": "result" + } + + return web.json_response(body) + + async def handle_devmanager(self, request): + json_body = json.loads(await request.text()) + randomid = ''.join(random.sample(string.ascii_letters,6)) + retcmd = await self.helperbot.send_command(json_body, randomid) + body = retcmd + + return web.json_response(body) + def disconnect(self): - logging.info('ConfServer: shutting down...') - self.server.disconnect() + logging.info('ConfServer: shutting down...') if(self.run_async): self.server.join() - logging.info('ConfServer: bye') + else: + self.server.disconnect() + logging.info('ConfServer: bye') + \ No newline at end of file diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 3a1f6d1..8fe32c4 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -16,6 +16,7 @@ from paho.mqtt import publish as MQTTPublish from paho.mqtt import subscribe as MQTTSubscribe import bumper import json +from datetime import datetime, timedelta class BumperMQTTPlugin: @@ -27,6 +28,8 @@ class BumperMQTTPlugin: self.context.logger.warning("'bots' section not found in context configuration") logging.debug('Bumper Plugin Initialized') + + async def on_broker_client_connected(self, client_id): logging.debug('Bumper Connection: %s connected' % client_id) connected_bots = self.bots['connected_bots'].get() @@ -56,20 +59,21 @@ class BumperMQTTPlugin: class MQTTHelperBot(ClientMQTT): - def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar): - ClientMQTT.__init__(self) + ClientMQTT.__init__(self) + self.address = address self._client_id = "helper1@bumper/helper1" + self.command_responses = contextvars.ContextVar('command_responses', default=[]) try: if run_async: hloop = asyncio.new_event_loop() logging.debug("Starting MQTT HelperBot Thread: 1") - mserver = Thread(name="MQTTHelperBot_Thread",target=self.run_helperbot, args=(hloop,)) - mserver.setDaemon(True) - mserver.start() + helperbot = Thread(name="MQTTHelperBot_Thread",target=self.run_helperbot, args=(hloop,)) + helperbot.setDaemon(True) + helperbot.start() else: self.run_helperbot() @@ -79,9 +83,10 @@ class MQTTHelperBot(ClientMQTT): pass def run_helperbot(self, loop): - formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" - logging.basicConfig(level=logging.INFO, format=formatter) + #formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" + #logging.basicConfig(level=logging.INFO, format=formatter) asyncio.set_event_loop(loop) + loop.run_until_complete(self.start_helper_bot()) loop.run_forever() @@ -118,15 +123,23 @@ class MQTTHelperBot(ClientMQTT): def get_msg(self, client, userdata, message): logging.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8")))) logging.debug(str(message.payload.decode("utf-8"))) - cresp = self.command_responses.get() - #str(message.payload.decode("utf-8") - cresp.append({"topic": message.topic,"payload":str(message.payload.decode("utf-8"))}) + cresp = self.command_responses.get() + + #Cleanup "expired messages" > 60 seconds from time + for msg in cresp: + expire_time = (datetime.fromtimestamp(msg['time']) + timedelta(seconds=10)).timestamp() + if time.time() > expire_time: + logging.debug("Pruning Message Time: {}, MsgTime: {}, MsgTime+60: {}".format(time.time(), msg['time'], expire_time)) + cresp.remove(msg) + + cresp.append({"time": time.time() ,"topic": message.topic,"payload":str(message.payload.decode("utf-8"))}) self.command_responses.set(cresp) + logging.debug("MQTT Command Response List Count: %s" %len(cresp)) async def wait_for_resp(self, requestid): - t_end = time.time() + 10 + t_end = (datetime.now() + timedelta(seconds=10)).timestamp() while time.time() < t_end: - await asyncio.sleep(0.2) + await asyncio.sleep(0.3) responses = self.command_responses.get() if len(responses) > 0: for msg in responses: @@ -150,13 +163,12 @@ class MQTTHelperBot(ClientMQTT): return { "id": requestid, "errno": "timeout", "ret": "fail" } - def send_command(self, cmdjson, requestid): + async def send_command(self, cmdjson, requestid): ttopic = "iot/p2p/{}/helper1/bumper/helper1/{}/{}/{}/q/{}/{}".format(cmdjson["cmdName"], cmdjson["toId"], cmdjson["toType"], cmdjson["toRes"], requestid, cmdjson["payloadType"]) self.publish(ttopic, str(cmdjson["payload"])) - loop = asyncio.new_event_loop() - resp = loop.run_until_complete(self.wait_for_resp(requestid)) + resp = await self.wait_for_resp(requestid) logging.debug(resp) return resp @@ -169,14 +181,22 @@ class MQTTServer(): async def broker_coro(self): broker = hbmqtt.broker.Broker(config=self.default_config) - for plugin in broker.plugins_manager.plugins: - if plugin.name == 'broker_sys': - broker.plugins_manager.plugins.remove(plugin) - if plugin.name == 'packet_logger_plugin': - broker.plugins_manager.plugins.remove(plugin) + logging.debug(broker.plugins_manager.plugins) await broker.start() + logging.debug("Removing Plugin: broker_sys") + broker.plugins_manager.plugins.remove(broker.plugins_manager.get_plugin('broker_sys')) + + logging.debug("Removing Plugin: topic_taboo") + broker.plugins_manager.plugins.remove(broker.plugins_manager.get_plugin('topic_taboo')) + + logging.debug("Removing Plugin: packet_logger_plugin") + broker.plugins_manager.plugins.remove(broker.plugins_manager.get_plugin('packet_logger_plugin')) + + logging.debug("Started Broker and Removed Plugins") + + async def active_bot_listing(self): while True: await asyncio.sleep(5) @@ -224,9 +244,9 @@ class MQTTServer(): if run_async: sloop = asyncio.new_event_loop() logging.debug("Starting MQTTServer Thread: 1") - mserver = Thread(name="MQTTServer_Thread",target=self.run_server, args=(sloop,)) - mserver.setDaemon(True) - mserver.start() + mqttserver = Thread(name="MQTTServer_Thread",target=self.run_server, args=(sloop,)) + mqttserver.setDaemon(True) + mqttserver.start() else: self.run_server() @@ -237,8 +257,8 @@ class MQTTServer(): def run_server(self, loop): - formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" - logging.basicConfig(level=logging.INFO, format=formatter) + #formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" + #logging.basicConfig(level=logging.INFO, format=formatter) asyncio.set_event_loop(loop) loop.run_until_complete(self.broker_coro()) #loop.run_until_complete(self.active_bot_listing()) diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 6be00d8..b4d7d56 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -27,6 +27,7 @@ class XMPPServer(): client.disconnect() thread_id = uuid.uuid4() client = Client(thread_id, connection, client_address) + client.start() self.clients.append(client) self.socket.close() @@ -61,6 +62,7 @@ class Client(threading.Thread): def __init__(self, thread_id, connection, client_address): threading.Thread.__init__(self) self.id = thread_id + self.name = "XMPP Thread {}".format(self.id) self.type = self.UNKNOWN self.state = self.IDLE self.connection = connection