mia/bumper
1
0
Fork 0
Code Issues 47 Pull requests 11 Projects Releases Packages Wiki Activity Actions

Split bots and clients, Fix XMPP threads not exiting

Browse source 
Split bots and clients, Fix XMPP threads not exiting
This commit is contained in:
Brian Martin 2019-02-03 16:16:29 -05:00
parent e0c9382156
commit f7d91aa650
5 changed files with 177 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

74
bumper/mqttserver.py
View file

@ -19,7 +19,7 @@ from datetime import datetime, timedelta
class MQTTHelperBot():
Client = MQTTClient()
def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar):
def __init__(self, address, run_async=False, bumper_bots=contextvars.ContextVar, bumper_clients=contextvars.ContextVar):
self.address = address
self.client_id = "helper1@bumper/helper1"
@ -51,7 +51,7 @@ class MQTTHelperBot():
async def start_helper_bot(self):
try:
await self.Client.connect('mqtts://{}:{}/'.format(self.address[0], self.address[1]), cafile='./certs/CA/cacert.pem')
await self.Client.connect('mqtts://{}:{}/'.format(self.address[0], self.address[1]), cafile=bumper.ca_cert)
await self.Client.subscribe([
('iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+',QOS_0),
('iot/p2p/+',QOS_0)
@ -125,6 +125,7 @@ class MQTTHelperBot():
class MQTTServer():
default_config = {}
bumper_clients = []
bumper_bots = []
async def broker_coro(self):
broker = hbmqtt.broker.Broker(config=self.default_config)
@ -133,16 +134,16 @@ class MQTTServer():
async def active_bot_listing(self):
while True:
await asyncio.sleep(5)
logging.debug('Connected bots: %s' % self.bumper_clients.get())
logging.debug('Connected bots: %s' % self.bumper_bots.get())
def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar):
def __init__(self, address, run_async=False, bumper_bots=contextvars.ContextVar, bumper_clients=contextvars.ContextVar):
#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:BumperMQTTServer_Plugin', dist=distribution)
distribution._ep_map = {"hbmqtt.broker.plugins": {"bumper": bumper_plugin}}
pkg_resources.working_set.add(distribution)
self.bumper_bots = bumper_bots
self.bumper_clients = bumper_clients
try:
# Initialize bot server
@ -154,8 +155,8 @@ class MQTTServer():
'tls1': {
'bind': "{}:{}".format(address[0], address[1]),
'ssl': 'on',
'certfile': './certs/cert.pem',
'keyfile': './certs/key.pem',
'certfile': bumper.server_cert,
'keyfile': bumper.server_key,
},
},
'sys_interval': 10,
@ -169,8 +170,10 @@ class MQTTServer():
'topic-check': {
'enabled': False
},
'bots':{
'connected_bots': bumper_clients
'clients':{
'connected_bots': bumper_bots,
'connected_clients': bumper_clients
}
}
if run_async:
@ -197,15 +200,17 @@ class MQTTServer():
class BumperMQTTServer_Plugin:
def __init__(self, context):
self.context = context
try:
self.bots = self.context.config['bots']
try:
self.clients = self.context.config['clients']
except KeyError:
self.context.logger.warning("'bots' section not found in context configuration")
self.context.logger.warning("'clients' 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()
connected_bots = self.clients['connected_bots'].get()
connected_clients = self.clients['connected_clients'].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")):
@ -221,20 +226,51 @@ class BumperMQTTServer_Plugin:
if botactive == False:
connected_bots.append(newbot.asdict())
logging.info("Adding bot to list: {}".format(newbot.asdict()))
self.bots['connected_bots'].set(connected_bots)
self.clients['connected_bots'].set(connected_bots)
else:
tmpuserdetail = str(didsplit[1]).split("/")
newuser = bumper.VacBotUser()
newuser.userid = didsplit[0]
newuser.realm = tmpuserdetail[0]
newuser.resource = tmpuserdetail[1]
clientactive = False
for client in connected_clients:
if client['userid'] == newuser.userid:
clientactive = True
if clientactive == False:
connected_clients.append(newuser.asdict())
logging.info("Adding client to list: {}".format(newuser.asdict()))
self.clients['connected_clients'].set(connected_clients)
logging.debug('Connected Bots: %s' %self.clients['connected_bots'].get())
logging.debug('Connected Clients: %s' %self.clients['connected_clients'].get())
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()
connected_bots = self.clients['connected_bots'].get()
connected_clients = self.clients['connected_clients'].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))
logging.info("Removing bot from list: {}".format(bot['did']))
connected_bots.remove(bot)
self.bots['connected_bots'].set(connected_bots)
self.clients['connected_bots'].set(connected_bots)
logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())
logging.debug('Connected Bots: %s' %self.clients['connected_bots'].get())
for client in connected_clients:
if didsplit[0] == client['userid']:
logging.info("Removing client from list: {}".format(client['userid']))
connected_clients.remove(client)
self.clients['connected_clients'].set(connected_clients)
logging.debug('Connected Clients: %s' %self.clients['connected_clients'].get())