summaryrefslogtreecommitdiff
path: root/sleekxmpp/xmlstream/xmlstream.py
diff options
context:
space:
mode:
Diffstat (limited to 'sleekxmpp/xmlstream/xmlstream.py')
-rw-r--r--sleekxmpp/xmlstream/xmlstream.py39
1 files changed, 25 insertions, 14 deletions
diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py
index 401865c6..948b19db 100644
--- a/sleekxmpp/xmlstream/xmlstream.py
+++ b/sleekxmpp/xmlstream/xmlstream.py
@@ -398,6 +398,11 @@ class XMLStream(object):
stream processing and not during normal event
processing.
"""
+ # To prevent circular dependencies, we must load the matcher
+ # and handler classes here.
+ from sleekxmpp.xmlstream.matcher import MatchXMLMask
+ from sleekxmpp.xmlstream.handler import XMLCallback
+
if name is None:
name = 'add_handler_%s' % self.getNewId()
self.registerHandler(XMLCallback(name, MatchXMLMask(mask), pointer,
@@ -478,19 +483,13 @@ class XMLStream(object):
Defaults to an empty dictionary.
"""
for handler in self.__event_handlers.get(name, []):
- func, threaded, disposable = handler
-
- handler_data = copy.copy(data)
- if threaded:
- x = threading.Thread(name="Event_%s" % str(func),
- target=func,
- args=(handler_data,))
- x.start()
- else:
- func(handler_data)
- if disposable:
+ self.event_queue.put(('event', handler, copy.copy(data)))
+ if handler[2]:
+ # If the handler is disposable, we will go ahead and
+ # remove it now instead of waiting for it to be
+ # processed in the queue.
with self.__event_handlers_lock:
- handler_index = self.event_handlers[name].index(handler)
+ handler_index = self.__event_handlers[name].index(handler)
self.__event_handlers[name].pop(handler_index)
def schedule(self, name, seconds, callback, args=None,
@@ -522,7 +521,7 @@ class XMLStream(object):
"""
return xml
- def send(self, data, mask, timeout=RESPONSE_TIMEOUT):
+ def send(self, data, mask=None, timeout=RESPONSE_TIMEOUT):
"""
A wrapper for send_raw for sending stanza objects.
@@ -772,7 +771,7 @@ class XMLStream(object):
try:
handler.run(args[0])
except Exception as e:
- error_msg = 'Error processing event handler: %s'
+ error_msg = 'Error processing stream handler: %s'
logging.exception(error_msg % handler.name)
args[0].exception(e)
elif etype == 'schedule':
@@ -781,6 +780,18 @@ class XMLStream(object):
handler(*args[0])
except:
logging.exception('Error processing scheduled task')
+ elif etype == 'event':
+ func, threaded, disposable = handler
+ try:
+ if threaded:
+ x = threading.Thread(name="Event_%s" % str(func),
+ target=func,
+ args=args)
+ x.start()
+ else:
+ func(*args)
+ except:
+ logging.exception('Error processing event handler: %s')
elif etype == 'quit':
logging.debug("Quitting event runner thread")
return False