Remove paho and use hbmqtt for helper

Remove paho and use hbmqtt for helper
Fix sleep bug

Pipfile

@@ -5,7 +5,6 @@ name = "pypi"
 
 [packages]
 hbmqtt = "*"
-paho-mqtt = "*"
 aiohttp = "*"
 
 [dev-packages]

bumper/__init__.py

@@ -9,7 +9,6 @@ import contextvars
 import time
 
 bumper_clients_var = contextvars.ContextVar('bumper_clients', default=[])
-current_milli_time = lambda: int(round(time.time() * 1000))
 
 def get_milli_time(timetoconvert):
     return int(round(timetoconvert * 1000))

bumper/confserver.py

@@ -93,13 +93,13 @@ class ConfServer():
                 "username": "fusername_1" #Random chars 8
                 },
                 "msg": "操作成功",
-                "time": bumper.current_milli_time()
+                "time": bumper.get_milli_time(time.time())
                 }    
                                             
         return web.json_response(body)
 
     async def handle_logout(self, request):                      
-        body = {"code": "0000","data": None,"msg": "操作成功", "time": bumper.current_milli_time()}
+        body = {"code": "0000","data": None,"msg": "操作成功", "time": bumper.get_milli_time(time.time())}
                                             
         return web.json_response(body)        
 
@@ -112,7 +112,7 @@ class ConfServer():
                 "ecovacsUid": "fuid_1" #Date(14)_RandomChars(32)
                 },
                 "msg": "操作成功",
-                "time": bumper.current_milli_time()
+                "time": bumper.get_milli_time(time.time())
                 }         
                                                     
         return web.json_response(body)     
@@ -130,7 +130,7 @@ class ConfServer():
                     "v": None
                 },
                 "msg": "操作成功",
-                "time": bumper.current_milli_time()
+                "time": bumper.get_milli_time(time.time())
             }        
                                                     
         return web.json_response(body)            
@@ -140,7 +140,7 @@ class ConfServer():
                 "code": "0000",
                 "data": [],
                 "msg": "操作成功",
-                "time": bumper.current_milli_time()
+                "time": bumper.get_milli_time(time.time())
             
             }     
                                                     
@@ -157,10 +157,10 @@ class ConfServer():
                     "hasCampaign": "N",
                     "imageUrl": None,
                     "nextAlertTime": nextAlert,
-                    "serverTime": bumper.current_milli_time()
+                    "serverTime": bumper.get_milli_time(time.time())
                 },
                 "msg": "操作成功",
-                "time": bumper.current_milli_time()
+                "time": bumper.get_milli_time(time.time())
             }   
                                                 
         return web.json_response(body)          

bumper/mqttserver.py

@@ -5,72 +5,24 @@ import asyncio
 import os
 import hbmqtt
 from hbmqtt.broker import Broker
+from hbmqtt.client import MQTTClient
+from hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2
 import pkg_resources
 import contextvars
 import time
 from threading import Thread
 import ssl
-from paho.mqtt.client import Client as ClientMQTT
-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:    
+class MQTTHelperBot():
-    def __init__(self, context):
+    Client = MQTTClient()
-        self.context = context        
-        try:
-            self.bots = self.context.config['bots']
-        except KeyError:
-             self.context.logger.warning("'bots' section not found in context configuration")
-            
-
-    async def on_broker_client_connected(self, client_id):
-        logging.debug('Bumper Connection: %s connected' % client_id)
-        connected_bots = self.bots['connected_bots'].get()                
-        didsplit = str(client_id).split("@")
-        #If this isn't a fake user (fuid) then add as a bot
-        if not (str(didsplit[0]).startswith("fuid") or str(didsplit[0]).startswith("helper")):
-            tmpbotdetail = str(didsplit[1]).split("/")        
-            newbot = bumper.VacBotDevice()
-            newbot.did = didsplit[0]        
-            newbot.vac_bot_device_class = tmpbotdetail[0]
-            newbot.resource = tmpbotdetail[1]
-            botactive = False
-            for bot in connected_bots:
-                if bot['did'] == newbot.did:
-                    botactive = True
-            
-            if botactive == False:
-                connected_bots.append(newbot.asdict())
-            
-            self.bots['connected_bots'].set(connected_bots)
-
-        logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())
-
-    async def on_broker_client_disconnected(self, client_id):
-        logging.debug('Bumper Connection: %s disconnected' % client_id)
-        connected_bots = self.bots['connected_bots'].get()                
-        didsplit = str(client_id).split("@")
-        #If the did is in the list, remove it
-        for bot in connected_bots:
-            if didsplit[0] == bot['did']:
-                logging.debug("Removing bot from list: {}".format(bot))
-                connected_bots.remove(bot)
-                self.bots['connected_bots'].set(connected_bots)
-
-        logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())
-
-class MQTTHelperBot(ClientMQTT):
-
     def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar):
-        ClientMQTT.__init__(self)        
         
         self.address = address
-        self._client_id = "helper1@bumper/helper1"
+        self.client_id = "helper1@bumper/helper1"       
-        
         self.command_responses = contextvars.ContextVar('command_responses', default=[])
         
         try:            
@@ -89,60 +41,51 @@ class MQTTHelperBot(ClientMQTT):
             pass    
 
     def run_helperbot(self, loop):     
-        asyncio.set_event_loop(loop)
         
+        asyncio.set_event_loop(loop)                
+        self.Client = MQTTClient(client_id=self.client_id, config={'check_hostname':False})      
         loop.run_until_complete(self.start_helper_bot())    
+        loop.run_until_complete(self.get_msg()) 
         loop.run_forever() 
     
     async def start_helper_bot(self):        
-        #self._on_log = self.on_log #This provides more logging than needed, even for debug
-        self._on_message = self.get_msg
-        self._on_connect = self.on_connect        
 
-        #TODO: This is pretty insecure and accepts any cert, maybe actually check?
+        try:
-        ssl_ctx = ssl.create_default_context()
+            await self.Client.connect('mqtts://{}:{}/'.format(self.address[0], self.address[1]), cafile='./certs/CA/cacert.pem')
-        ssl_ctx.check_hostname = False
+            await self.Client.subscribe([
-        ssl_ctx.verify_mode = ssl.CERT_NONE
+                ('iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+',QOS_0),
-        self.tls_set_context(ssl_ctx)
+                ('iot/p2p/+',QOS_0)                
-        self.tls_insecure_set(True)
+            ])    
-
+        except hbmqtt.client.ClientException as ce:
-        self.connect(self.address[0], self.address[1])
+            logging.exception("Client exception: %s" % ce)
-        self.loop_start()        
 
 
-    def on_connect(self, client, userdata, flags, rc):        
+    async def get_msg(self):
-        if rc != 0:
+        try:
-            logging.error("HelperBot - error connecting with MQTT Return {}".format(rc))
+            while True:
-            raise RuntimeError("HelperBot - error connecting with MQTT Return {}".format(rc))
+                message = await self.Client.deliver_message()          
             
-        else:
+                #logging.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8"))))
-            logging.debug("HelperBot - Connected with result code "+str(rc))
+                cresp = self.command_responses.get()        
-            logging.debug("HelperBot - Subscribing to all")        
                 
+                #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)                                  
 
-            self.subscribe('iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+', qos=0)
+                cresp.append({"time": time.time() ,"topic": message.topic,"payload":str(message.data.decode("utf-8"))})
-            self.subscribe('iot/p2p/+', qos=0)
+                self.command_responses.set(cresp)                          
+                logging.debug("MQTT Command Response List Count: %s" %len(cresp))
+        except hbmqtt.client.ClientException as ce:
+            logging.error("Client exception: %s" % ce)
    
 
-    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"))))
-        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 = (datetime.now() + timedelta(seconds=10)).timestamp()
         while time.time() < t_end:
-            await asyncio.sleep(0.3)
+            await asyncio.sleep(0.1)
             responses = self.command_responses.get()   
             if len(responses) > 0:    
                 for msg in responses:
@@ -166,9 +109,13 @@ class MQTTHelperBot(ClientMQTT):
         return { "id": requestid, "errno": "timeout", "ret": "fail" }
 
     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"])
+            cmdjson["toId"], cmdjson["toType"], cmdjson["toRes"], requestid, cmdjson["payloadType"])
-        self.publish(ttopic, str(cmdjson["payload"]))
+        try:                
+            await self.Client.publish(ttopic, str(cmdjson["payload"]).encode(),QOS_0)
+        except:
+            logging.exception("Exception at send_command")
         
         resp = await self.wait_for_resp(requestid)        
                 
@@ -181,21 +128,8 @@ class MQTTServer():
 
     async def broker_coro(self):         
         broker = hbmqtt.broker.Broker(config=self.default_config)
-              
         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)
@@ -205,11 +139,9 @@ class MQTTServer():
         
         #The below adds a plugin to the hbmqtt.broker.plugins without having to futz with setup.py
         distribution = pkg_resources.Distribution("hbmqtt.broker.plugins")
-        bumper_plugin = pkg_resources.EntryPoint.parse('bumper = bumper.mqttserver:BumperMQTTPlugin', dist=distribution)        
+        bumper_plugin = pkg_resources.EntryPoint.parse('bumper = bumper.mqttserver:BumperMQTTServer_Plugin', dist=distribution)        
         distribution._ep_map = {"hbmqtt.broker.plugins": {"bumper": bumper_plugin}}
         pkg_resources.working_set.add(distribution)
-        #for entry_point in pkg_resources.iter_entry_points("hbmqtt.broker.plugins"):
-        #   print(entry_point)
 
         self.bumper_clients = bumper_clients
         try:            
@@ -262,3 +194,47 @@ class MQTTServer():
         #loop.run_until_complete(self.active_bot_listing())                     
         loop.run_forever()  
 
+class BumperMQTTServer_Plugin:    
+    def __init__(self, context):
+        self.context = context        
+        try:
+            self.bots = self.context.config['bots']
+        except KeyError:
+             self.context.logger.warning("'bots' section not found in context configuration")
+            
+
+    async def on_broker_client_connected(self, client_id):
+        logging.debug('Bumper Connection: %s connected' % client_id)
+        connected_bots = self.bots['connected_bots'].get()                
+        didsplit = str(client_id).split("@")
+        #If this isn't a fake user (fuid) then add as a bot
+        if not (str(didsplit[0]).startswith("fuid") or str(didsplit[0]).startswith("helper")):
+            tmpbotdetail = str(didsplit[1]).split("/")        
+            newbot = bumper.VacBotDevice()
+            newbot.did = didsplit[0]        
+            newbot.vac_bot_device_class = tmpbotdetail[0]
+            newbot.resource = tmpbotdetail[1]
+            botactive = False
+            for bot in connected_bots:
+                if bot['did'] == newbot.did:
+                    botactive = True
+            
+            if botactive == False:
+                connected_bots.append(newbot.asdict())
+            
+            self.bots['connected_bots'].set(connected_bots)
+
+        logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())
+
+    async def on_broker_client_disconnected(self, client_id):
+        logging.debug('Bumper Connection: %s disconnected' % client_id)
+        connected_bots = self.bots['connected_bots'].get()                
+        didsplit = str(client_id).split("@")
+        #If the did is in the list, remove it
+        for bot in connected_bots:
+            if didsplit[0] == bot['did']:
+                logging.debug("Removing bot from list: {}".format(bot))
+                connected_bots.remove(bot)
+                self.bots['connected_bots'].set(connected_bots)
+
+        logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())

bumper/xmppserver.py

@@ -127,6 +127,7 @@ class Client(threading.Thread):
             logging.info('client connected: {}'.format(self.address))
             self._set_state('CONNECT')
             while True:
+                time.sleep(0.2)
                 if not self.connection._closed:               
                     data = self.connection.recv(4096)
                 if data: