Merge branch 'develop' of github.com:fritzy/SleekXMPP into develop

This commit is contained in:
Nathan Fritz
2010-06-01 21:45:15 -07:00
13 changed files with 841 additions and 121 deletions

44
sleekxmpp/plugins/jobs.py Normal file
View File

@@ -0,0 +1,44 @@
from . import base
import logging
from xml.etree import cElementTree as ET
class jobs(base.base_plugin):
def plugin_init(self):
self.xep = 'pubsubjob'
self.description = "Job distribution over Pubsub"
def post_init(self):
pass
#TODO add event
def createJobNode(self, host, jid, node, config=None):
pass
def createJob(self, host, node, jobid=None, payload=None):
return self.xmpp.plugin['xep_0060'].setItem(host, node, ((jobid, payload),))
def claimJob(self, host, node, jobid, ifrom=None):
return self._setState(host, node, jobid, ET.Element('{http://andyet.net/protocol/pubsubjob}claimed'))
def unclaimJob(self, jobid):
return self._setState(host, node, jobid, ET.Element('{http://andyet.net/protocol/pubsubjob}unclaimed'))
def finishJob(self, host, node, jobid, payload=None):
finished = ET.Element('{http://andyet.net/protocol/pubsubjob}finished')
if payload is not None:
finished.append(payload)
return self._setState(host, node, jobid, finished)
def _setState(self, host, node, jobid, state, ifrom=None):
iq = self.xmpp.Iq()
iq['to'] = host
if ifrom: iq['from'] = ifrom
iq['type'] = 'set'
iq['psstate']['node'] = node
iq['psstate']['item'] = jobid
iq['psstate']['payload'] = state
result = iq.send()
if result is None or result['type'] != 'result':
return False
return True

View File

@@ -10,6 +10,39 @@ def stanzaPlugin(stanza, plugin):
stanza.plugin_attrib_map[plugin.plugin_attrib] = plugin
stanza.plugin_tag_map["{%s}%s" % (plugin.namespace, plugin.name)] = plugin
class PubsubState(ElementBase):
namespace = 'http://jabber.org/protocol/psstate'
name = 'state'
plugin_attrib = 'psstate'
interfaces = set(('node', 'item', 'payload'))
plugin_attrib_map = {}
plugin_tag_map = {}
def setPayload(self, value):
self.xml.append(value)
def getPayload(self):
childs = self.xml.getchildren()
if len(childs) > 0:
return childs[0]
def delPayload(self):
for child in self.xml.getchildren():
self.xml.remove(child)
stanzaPlugin(Iq, PubsubState)
class PubsubStateEvent(ElementBase):
namespace = 'http://jabber.org/protocol/psstate#event'
name = 'event'
plugin_attrib = 'psstate_event'
intefaces = set(tuple())
plugin_attrib_map = {}
plugin_tag_map = {}
stanzaPlugin(Message, PubsubStateEvent)
stanzaPlugin(PubsubStateEvent, PubsubState)
class Pubsub(ElementBase):
namespace = 'http://jabber.org/protocol/pubsub'
name = 'pubsub'
@@ -321,18 +354,6 @@ class Options(ElementBase):
stanzaPlugin(Pubsub, Options)
stanzaPlugin(Subscribe, Options)
#iq = Iq()
#iq['pubsub']['defaultconfig']
#print(iq)
#from xml.etree import cElementTree as ET
#iq = Iq()
#item = Item()
#item['payload'] = ET.Element("{http://netflint.net/p/crap}stupidshit")
#item['id'] = 'aa11bbcc'
#iq['pubsub']['items'].append(item)
#print(iq)
class OwnerAffiliations(Affiliations):
namespace = 'http://jabber.org/protocol/pubsub#owner'
interfaces = set(('node'))

View File

@@ -188,7 +188,6 @@ class Form(FieldContainer):
#def getXML(self, tostring = False):
def getXML(self, ftype=None):
logging.debug("creating form as %s" % ftype)
if ftype:
self.type = ftype
form = ET.Element('{jabber:x:data}x')

View File

@@ -1,25 +1,184 @@
"""
SleekXMPP: The Sleek XMPP Library
Copyright (C) 2007 Nathanael C. Fritz
This file is part of SleekXMPP.
SleekXMPP: The Sleek XMPP Library
Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
This file is part of SleekXMPP.
SleekXMPP is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
SleekXMPP is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with SleekXMPP; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
See the file license.txt for copying permissio
"""
from . import base
import logging
from xml.etree import cElementTree as ET
from . import base
from .. xmlstream.handler.callback import Callback
from .. xmlstream.matcher.xpath import MatchXPath
from .. xmlstream.stanzabase import ElementBase, ET, JID
from .. stanza.iq import Iq
class DiscoInfo(ElementBase):
namespace = 'http://jabber.org/protocol/disco#info'
name = 'query'
plugin_attrib = 'disco_info'
interfaces = set(('node', 'features', 'identities'))
def getFeatures(self):
features = []
featuresXML = self.xml.findall('{%s}feature' % self.namespace)
for feature in featuresXML:
features.append(feature.attrib['var'])
return features
def setFeatures(self, features):
self.delFeatures()
for name in features:
self.addFeature(name)
def delFeatures(self):
featuresXML = self.xml.findall('{%s}feature' % self.namespace)
for feature in featuresXML:
self.xml.remove(feature)
def addFeature(self, feature):
featureXML = ET.Element('{%s}feature' % self.namespace,
{'var': feature})
self.xml.append(featureXML)
def delFeature(self, feature):
featuresXML = self.xml.findall('{%s}feature' % self.namespace)
for featureXML in featuresXML:
if featureXML.attrib['var'] == feature:
self.xml.remove(featureXML)
def getIdentities(self):
ids = []
idsXML = self.xml.findall('{%s}identity' % self.namespace)
for idXML in idsXML:
idData = (idXML.attrib['category'],
idXML.attrib['type'],
idXML.attrib.get('name', ''))
ids.append(idData)
return ids
def setIdentities(self, ids):
self.delIdentities()
for idData in ids:
self.addIdentity(*idData)
def delIdentities(self):
idsXML = self.xml.findall('{%s}identity' % self.namespace)
for idXML in idsXML:
self.xml.remove(idXML)
def addIdentity(self, category, id_type, name=''):
idXML = ET.Element('{%s}identity' % self.namespace,
{'category': category,
'type': id_type,
'name': name})
self.xml.append(idXML)
def delIdentity(self, category, id_type, name=''):
idsXML = self.xml.findall('{%s}identity' % self.namespace)
for idXML in idsXML:
idData = (idXML.attrib['category'],
idXML.attrib['type'])
delId = (category, id_type)
if idData == delId:
self.xml.remove(idXML)
class DiscoItems(ElementBase):
namespace = 'http://jabber.org/protocol/disco#items'
name = 'query'
plugin_attrib = 'disco_items'
interfaces = set(('node', 'items'))
def getItems(self):
items = []
itemsXML = self.xml.findall('{%s}item' % self.namespace)
for item in itemsXML:
itemData = (item.attrib['jid'],
item.attrib.get('node'),
item.attrib.get('name'))
items.append(itemData)
return items
def setItems(self, items):
self.delItems()
for item in items:
self.addItem(*item)
def delItems(self):
itemsXML = self.xml.findall('{%s}item' % self.namespace)
for item in itemsXML:
self.xml.remove(item)
def addItem(self, jid, node='', name=''):
itemXML = ET.Element('{%s}item' % self.namespace, {'jid': jid})
if name:
itemXML.attrib['name'] = name
if node:
itemXML.attrib['node'] = node
self.xml.append(itemXML)
def delItem(self, jid, node=''):
itemsXML = self.xml.findall('{%s}item' % self.namespace)
for itemXML in itemsXML:
itemData = (itemXML.attrib['jid'],
itemXML.attrib.get('node', ''))
itemDel = (jid, node)
if itemData == itemDel:
self.xml.remove(itemXML)
class DiscoNode(object):
"""
Collection object for grouping info and item information
into nodes.
"""
def __init__(self, name):
self.name = name
self.info = DiscoInfo()
self.items = DiscoItems()
# This is a bit like poor man's inheritance, but
# to simplify adding information to the node we
# map node functions to either the info or items
# stanza objects.
#
# We don't want to make DiscoNode inherit from
# DiscoInfo and DiscoItems because DiscoNode is
# not an actual stanza, and doing so would create
# confusion and potential bugs.
self._map(self.items, 'items', ['get', 'set', 'del'])
self._map(self.items, 'item', ['add', 'del'])
self._map(self.info, 'identities', ['get', 'set', 'del'])
self._map(self.info, 'identity', ['add', 'del'])
self._map(self.info, 'features', ['get', 'set', 'del'])
self._map(self.info, 'feature', ['add', 'del'])
def isEmpty(self):
"""
Test if the node contains any information. Useful for
determining if a node can be deleted.
"""
ids = self.getIdentities()
features = self.getFeatures()
items = self.getItems()
if not ids and not features and not items:
return True
return False
def _map(self, obj, interface, access):
"""
Map functions of the form obj.accessInterface
to self.accessInterface for each given access type.
"""
interface = interface.title()
for access_type in access:
method = access_type + interface
if hasattr(obj, method):
setattr(self, method, getattr(obj, method))
class xep_0030(base.base_plugin):
"""
@@ -29,85 +188,137 @@ class xep_0030(base.base_plugin):
def plugin_init(self):
self.xep = '0030'
self.description = 'Service Discovery'
self.features = {'main': ['http://jabber.org/protocol/disco#info', 'http://jabber.org/protocol/disco#items']}
self.identities = {'main': [{'category': 'client', 'type': 'pc', 'name': 'SleekXMPP'}]}
self.items = {'main': []}
self.xmpp.add_handler("<iq type='get' xmlns='%s'><query xmlns='http://jabber.org/protocol/disco#info' /></iq>" % self.xmpp.default_ns, self.info_handler)
self.xmpp.add_handler("<iq type='get' xmlns='%s'><query xmlns='http://jabber.org/protocol/disco#items' /></iq>" % self.xmpp.default_ns, self.item_handler)
self.xmpp.registerHandler(
Callback('Disco Items',
MatchXPath('{%s}iq/{%s}query' % (self.xmpp.default_ns,
DiscoItems.namespace)),
self.handle_item_query))
self.xmpp.registerHandler(
Callback('Disco Info',
MatchXPath('{%s}iq/{%s}query' % (self.xmpp.default_ns,
DiscoInfo.namespace)),
self.handle_info_query))
self.xmpp.stanzaPlugin(Iq, DiscoInfo)
self.xmpp.stanzaPlugin(Iq, DiscoItems)
self.xmpp.add_event_handler('disco_items_request', self.handle_disco_items)
self.xmpp.add_event_handler('disco_info_request', self.handle_disco_info)
self.nodes = {'main': DiscoNode('main')}
def add_node(self, node):
if node not in self.nodes:
self.nodes[node] = DiscoNode(node)
def del_node(self, node):
if node in self.nodes:
del self.nodes[node]
def handle_item_query(self, iq):
if iq['type'] == 'get':
logging.debug("Items requested by %s" % iq['from'])
self.xmpp.event('disco_items_request', iq)
elif iq['type'] == 'result':
logging.debug("Items result from %s" % iq['from'])
self.xmpp.event('disco_items', iq)
def handle_info_query(self, iq):
if iq['type'] == 'get':
logging.debug("Info requested by %s" % iq['from'])
self.xmpp.event('disco_info_request', iq)
elif iq['type'] == 'result':
logging.debug("Info result from %s" % iq['from'])
self.xmpp.event('disco_info', iq)
def handle_disco_info(self, iq, forwarded=False):
"""
A default handler for disco#info requests. If another
handler is registered, this one will defer and not run.
"""
handlers = self.xmpp.event_handlers['disco_info_request']
if not forwarded and len(handlers) > 1:
return
node_name = iq['disco_info']['node']
if not node_name:
node_name = 'main'
logging.debug("Using default handler for disco#info on node '%s'." % node_name)
if node_name in self.nodes:
node = self.nodes[node_name]
iq.reply().setPayload(node.info.xml).send()
else:
logging.debug("Node %s requested, but does not exist." % node_name)
iq.reply().error().setPayload(iq['disco_info'].xml)
iq['error']['code'] = '404'
iq['error']['type'] = 'cancel'
iq['error']['condition'] = 'item-not-found'
iq.send()
def handle_disco_items(self, iq, forwarded=False):
"""
A default handler for disco#items requests. If another
handler is registered, this one will defer and not run.
If this handler is called by your own custom handler with
forwarded set to True, then it will run as normal.
"""
handlers = self.xmpp.event_handlers['disco_items_request']
if not forwarded and len(handlers) > 1:
return
node_name = iq['disco_items']['node']
if not node_name:
node_name = 'main'
logging.debug("Using default handler for disco#items on node '%s'." % node_name)
if node_name in self.nodes:
node = self.nodes[node_name]
iq.reply().setPayload(node.items.xml).send()
else:
logging.debug("Node %s requested, but does not exist." % node_name)
iq.reply().error().setPayload(iq['disco_items'].xml)
iq['error']['code'] = '404'
iq['error']['type'] = 'cancel'
iq['error']['condition'] = 'item-not-found'
iq.send()
# Older interface methods for backwards compatibility
def getInfo(self, jid, node=''):
iq = self.xmpp.Iq()
iq['type'] = 'get'
iq['to'] = jid
iq['from'] = self.xmpp.fulljid
iq['disco_info']['node'] = node
iq.send()
def getItems(self, jid, node=''):
iq = self.xmpp.Iq()
iq['type'] = 'get'
iq['to'] = jid
iq['from'] = self.xmpp.fulljid
iq['disco_items']['node'] = node
iq.send()
def add_feature(self, feature, node='main'):
if not node in self.features:
self.features[node] = []
self.features[node].append(feature)
self.add_node(node)
self.nodes[node].addFeature(feature)
def add_identity(self, category=None, itype=None, name=None, node='main'):
if not node in self.identities:
self.identities[node] = []
self.identities[node].append({'category': category, 'type': itype, 'name': name})
def add_identity(self, category='', itype='', name='', node='main'):
self.add_node(node)
self.nodes[node].addIdentity(category=category,
id_type=itype,
name=name)
def add_item(self, jid=None, name=None, node='main', subnode=''):
if not node in self.items:
self.items[node] = []
self.items[node].append({'jid': jid, 'name': name, 'node': subnode})
def info_handler(self, xml):
logging.debug("Info request from %s" % xml.get('from', ''))
iq = self.xmpp.makeIqResult(xml.get('id', self.xmpp.getNewId()))
iq.attrib['from'] = xml.get('to')
iq.attrib['to'] = xml.get('from', self.xmpp.server)
query = xml.find('{http://jabber.org/protocol/disco#info}query')
node = query.get('node', 'main')
for identity in self.identities.get(node, []):
idxml = ET.Element('identity')
for attrib in identity:
if identity[attrib]:
idxml.attrib[attrib] = identity[attrib]
query.append(idxml)
for feature in self.features.get(node, []):
featxml = ET.Element('feature')
featxml.attrib['var'] = feature
query.append(featxml)
iq.append(query)
#print ET.tostring(iq)
self.xmpp.send(iq)
def item_handler(self, xml):
logging.debug("Item request from %s" % xml.get('from', ''))
iq = self.xmpp.makeIqResult(xml.get('id', self.xmpp.getNewId()))
iq.attrib['from'] = xml.get('to')
iq.attrib['to'] = xml.get('from', self.xmpp.server)
query = self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#items').find('{http://jabber.org/protocol/disco#items}query')
node = xml.find('{http://jabber.org/protocol/disco#items}query').get('node', 'main')
for item in self.items.get(node, []):
itemxml = ET.Element('item')
itemxml.attrib = item
if itemxml.attrib['jid'] is None:
itemxml.attrib['jid'] = xml.get('to')
query.append(itemxml)
self.xmpp.send(iq)
def getItems(self, jid, node=None):
iq = self.xmpp.makeIqGet()
iq.attrib['from'] = self.xmpp.fulljid
iq.attrib['to'] = jid
self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#items')
if node:
iq.find('{http://jabber.org/protocol/disco#items}query').attrib['node'] = node
return iq.send()
def getInfo(self, jid, node=None):
iq = self.xmpp.makeIqGet()
iq.attrib['from'] = self.xmpp.fulljid
iq.attrib['to'] = jid
self.xmpp.makeIqQuery(iq, 'http://jabber.org/protocol/disco#info')
if node:
iq.find('{http://jabber.org/protocol/disco#info}query').attrib['node'] = node
return iq.send()
def parseInfo(self, xml):
result = {'identity': {}, 'feature': []}
for identity in xml.findall('{http://jabber.org/protocol/disco#info}query/{{http://jabber.org/protocol/disco#info}identity'):
result['identity'][identity['name']] = identity.attrib
for feature in xml.findall('{http://jabber.org/protocol/disco#info}query/{{http://jabber.org/protocol/disco#info}feature'):
result['feature'].append(feature.get('var', '__unknown__'))
return result
def add_item(self, jid=None, name='', node='main', subnode=''):
self.add_node(node)
self.add_node(subnode)
if jid is None:
jid = self.xmpp.fulljid
self.nodes[node].addItem(jid=jid, name=name, node=subnode)

View File

@@ -14,12 +14,14 @@ class xep_0060(base.base_plugin):
self.xep = '0060'
self.description = 'Publish-Subscribe'
def create_node(self, jid, node, config=None, collection=False):
def create_node(self, jid, node, config=None, collection=False, ntype=None):
pubsub = ET.Element('{http://jabber.org/protocol/pubsub}pubsub')
create = ET.Element('create')
create.set('node', node)
pubsub.append(create)
configure = ET.Element('configure')
if collection:
ntype = 'collection'
#if config is None:
# submitform = self.xmpp.plugin['xep_0004'].makeForm('submit')
#else:
@@ -29,11 +31,11 @@ class xep_0060(base.base_plugin):
submitform.field['FORM_TYPE'].setValue('http://jabber.org/protocol/pubsub#node_config')
else:
submitform.addField('FORM_TYPE', 'hidden', value='http://jabber.org/protocol/pubsub#node_config')
if collection:
if ntype:
if 'pubsub#node_type' in submitform.field:
submitform.field['pubsub#node_type'].setValue('collection')
submitform.field['pubsub#node_type'].setValue(ntype)
else:
submitform.addField('pubsub#node_type', value='collection')
submitform.addField('pubsub#node_type', value=ntype)
else:
if 'pubsub#node_type' in submitform.field:
submitform.field['pubsub#node_type'].setValue('leaf')

View File

@@ -11,7 +11,7 @@ class Error(ElementBase):
namespace = 'jabber:client'
name = 'error'
plugin_attrib = 'error'
conditions = set(('bad-request', 'conflict', 'feature-not-implemented', 'forbidden', 'gone', 'item-not-found', 'jid-malformed', 'not-acceptable', 'not-allowed', 'not-authorized', 'payment-required', 'recipient-unavailable', 'redirect', 'registration-required', 'remote-server-not-found', 'remote-server-timeout', 'service-unavailable', 'subscription-required', 'undefined-condition', 'unexpected-request'))
conditions = set(('bad-request', 'conflict', 'feature-not-implemented', 'forbidden', 'gone', 'internal-server-error', 'item-not-found', 'jid-malformed', 'not-acceptable', 'not-allowed', 'not-authorized', 'payment-required', 'recipient-unavailable', 'redirect', 'registration-required', 'remote-server-not-found', 'remote-server-timeout', 'resource-constraint', 'service-unavailable', 'subscription-required', 'undefined-condition', 'unexpected-request'))
interfaces = set(('code', 'condition', 'text', 'type'))
types = set(('cancel', 'continue', 'modify', 'auth', 'wait'))
sub_interfaces = set(('text',))

View File

@@ -0,0 +1,87 @@
try:
import queue
except ImportError:
import Queue as queue
import time
import threading
import logging
class Task(object):
"""Task object for the Scheduler class"""
def __init__(self, name, seconds, callback, args=None, kwargs=None, repeat=False, qpointer=None):
self.name = name
self.seconds = seconds
self.callback = callback
self.args = args or tuple()
self.kwargs = kwargs or {}
self.repeat = repeat
self.next = time.time() + self.seconds
self.qpointer = qpointer
def run(self):
if self.qpointer is not None:
self.qpointer.put(('schedule', self.callback, self.args))
else:
self.callback(*self.args, **self.kwargs)
self.reset()
return self.repeat
def reset(self):
self.next = time.time() + self.seconds
class Scheduler(object):
"""Threaded scheduler that allows for updates mid-execution unlike http://docs.python.org/library/sched.html#module-sched"""
def __init__(self, parentqueue=None):
self.addq = queue.Queue()
self.schedule = []
self.thread = None
self.run = False
self.parentqueue = parentqueue
def process(self, threaded=True):
if threaded:
self.thread = threading.Thread(name='shedulerprocess', target=self._process)
self.thread.start()
else:
self._process()
def _process(self):
self.run = True
while self.run:
try:
wait = 1
updated = False
if self.schedule:
wait = self.schedule[0].next - time.time()
try:
if wait <= 0.0:
newtask = self.addq.get(False)
else:
newtask = self.addq.get(True, wait)
except queue.Empty:
cleanup = []
for task in self.schedule:
if time.time() >= task.next:
updated = True
if not task.run():
cleanup.append(task)
else:
break
for task in cleanup:
x = self.schedule.pop(self.schedule.index(task))
else:
updated = True
self.schedule.append(newtask)
finally:
if updated: self.schedule = sorted(self.schedule, key=lambda task: task.next)
except KeyboardInterrupt:
self.run = False
logging.debug("Qutting Scheduler thread")
if self.parentqueue is not None:
self.parentqueue.put(('quit', None, None))
def add(self, name, seconds, callback, args=None, kwargs=None, repeat=False, qpointer=None):
self.addq.put(Task(name, seconds, callback, args, kwargs, repeat, qpointer))
def quit(self):
self.run = False

View File

@@ -78,6 +78,9 @@ class ElementBase(tostring.ToString):
def __iter__(self):
self.idx = 0
return self
def __bool__(self):
return True
def __next__(self):
self.idx += 1
@@ -319,6 +322,8 @@ class StanzaBase(ElementBase):
def __init__(self, stream=None, xml=None, stype=None, sto=None, sfrom=None, sid=None):
self.stream = stream
if stream is not None:
self.namespace = stream.default_ns
ElementBase.__init__(self, xml)
if stype is not None:
self['type'] = stype
@@ -326,8 +331,6 @@ class StanzaBase(ElementBase):
self['to'] = sto
if sfrom is not None:
self['from'] = sfrom
if stream is not None:
self.namespace = stream.default_ns
self.tag = "{%s}%s" % (self.namespace, self.name)
def setType(self, value):

View File

@@ -22,6 +22,7 @@ import time
import traceback
import types
import xml.sax.saxutils
from . import scheduler
HANDLER_THREADS = 1
@@ -75,6 +76,7 @@ class XMLStream(object):
self.eventqueue = queue.Queue()
self.sendqueue = queue.Queue()
self.scheduler = scheduler.Scheduler(self.eventqueue)
self.namespace_map = {}
@@ -145,7 +147,9 @@ class XMLStream(object):
raise RestartStream()
def process(self, threaded=True):
self.scheduler.process(threaded=True)
for t in range(0, HANDLER_THREADS):
logging.debug("Starting HANDLER THREAD")
self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner)
self.__thread['eventhandle%s' % t].start()
self.__thread['sendthread'] = threading.Thread(name='sendthread', target=self._sendThread)
@@ -156,8 +160,8 @@ class XMLStream(object):
else:
self._process()
def schedule(self, seconds, handler, args=None):
threading.Timer(seconds, handler, args).start()
def schedule(self, name, seconds, callback, args=None, kwargs=None, repeat=False):
self.scheduler.add(name, seconds, callback, args, kwargs, repeat, qpointer=self.eventqueue)
def _process(self):
"Start processing the socket."
@@ -177,6 +181,7 @@ class XMLStream(object):
self.state.set('reconnect', False)
self.disconnect()
self.run = False
self.scheduler.run = False
self.eventqueue.put(('quit', None, None))
return
except CloseStream:
@@ -223,6 +228,7 @@ class XMLStream(object):
edepth += -1
if edepth == 0 and event == b'end':
self.disconnect(reconnect=self.state['reconnect'])
logging.debug("Ending readXML loop")
return False
elif edepth == 1:
#self.xmlin.put(xmlobj)
@@ -231,11 +237,13 @@ class XMLStream(object):
except RestartStream:
return True
except CloseStream:
logging.debug("Ending readXML loop")
return False
if root:
root.clear()
if event == b'start':
edepth += 1
logging.debug("Ending readXML loop")
def _sendThread(self):
while self.run:
@@ -265,6 +273,7 @@ class XMLStream(object):
logging.debug("Disconnecting...")
self.state.set('disconnecting', True)
self.run = False
self.scheduler.run = False
if self.state['connected']:
self.sendRaw(self.stream_footer)
time.sleep(1)
@@ -323,6 +332,9 @@ class XMLStream(object):
event = self.eventqueue.get(True, timeout=5)
except queue.Empty:
event = None
except KeyboardInterrupt:
self.run = False
self.scheduler.run = False
if event is not None:
etype = event[0]
handler = event[1]
@@ -334,9 +346,10 @@ class XMLStream(object):
except Exception as e:
traceback.print_exc()
args[0].exception(e)
elif etype == 'sched':
elif etype == 'schedule':
try:
handler.run(*args)
logging.debug(args)
handler(*args[0])
except:
logging.error(traceback.format_exc())
elif etype == 'quit':