From 0bda5fd3f28da14478db87c98fb89f647994deaa Mon Sep 17 00:00:00 2001 From: Nathan Fritz Date: Wed, 26 May 2010 18:32:28 -0700 Subject: adding scheduler --- sleekxmpp/xmlstream/scheduler.py | 76 ++++++++++++++++++++++++++++++++++++++++ sleekxmpp/xmlstream/xmlstream.py | 9 +++-- 2 files changed, 82 insertions(+), 3 deletions(-) create mode 100644 sleekxmpp/xmlstream/scheduler.py (limited to 'sleekxmpp/xmlstream') diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py new file mode 100644 index 00000000..5cb8aff0 --- /dev/null +++ b/sleekxmpp/xmlstream/scheduler.py @@ -0,0 +1,76 @@ +try: + import queue +except ImportError: + import Queue as queue +import time +import threading + +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): + self.addq = queue.Queue() + self.schedule = [] + self.thread = None + self.run = True + + 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): + while self.run: + wait = 5 + updated = False + if self.schedule: + wait = self.schedule[0].next - time.time() + try: + 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) + print [x.name for x in self.schedule] + + 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 diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 025884b7..96b47411 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -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.namespace_map = {} @@ -145,6 +147,7 @@ class XMLStream(object): raise RestartStream() def process(self, threaded=True): + self.scheduler.process(threaded=True) for t in range(0, HANDLER_THREADS): self.__thread['eventhandle%s' % t] = threading.Thread(name='eventhandle%s' % t, target=self._eventRunner) self.__thread['eventhandle%s' % t].start() @@ -156,8 +159,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." @@ -334,7 +337,7 @@ class XMLStream(object): except Exception as e: traceback.print_exc() args[0].exception(e) - elif etype == 'sched': + elif etype == 'schedule': try: handler.run(*args) except: -- cgit v1.2.3 From 3a28f9e5d247b9124e0d14c26c2a6e79aaee86ff Mon Sep 17 00:00:00 2001 From: Nathan Fritz Date: Thu, 27 May 2010 04:58:57 -0700 Subject: added pubsub state stanzas and scheduled events --- sleekxmpp/xmlstream/scheduler.py | 6 ++++-- sleekxmpp/xmlstream/stanzabase.py | 4 ++-- sleekxmpp/xmlstream/xmlstream.py | 8 +++++++- 3 files changed, 13 insertions(+), 5 deletions(-) (limited to 'sleekxmpp/xmlstream') diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py index 5cb8aff0..7aa59f3d 100644 --- a/sleekxmpp/xmlstream/scheduler.py +++ b/sleekxmpp/xmlstream/scheduler.py @@ -4,6 +4,7 @@ except ImportError: import Queue as queue import time import threading +import logging class Task(object): """Task object for the Scheduler class""" @@ -34,7 +35,7 @@ class Scheduler(object): self.addq = queue.Queue() self.schedule = [] self.thread = None - self.run = True + self.run = False def process(self, threaded=True): if threaded: @@ -44,6 +45,7 @@ class Scheduler(object): self._process() def _process(self): + self.run = True while self.run: wait = 5 updated = False @@ -67,7 +69,7 @@ class Scheduler(object): self.schedule.append(newtask) finally: if updated: self.schedule = sorted(self.schedule, key=lambda task: task.next) - print [x.name for x in self.schedule] + logging.debug("Qutting Scheduler thread") 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)) diff --git a/sleekxmpp/xmlstream/stanzabase.py b/sleekxmpp/xmlstream/stanzabase.py index 3f3f5e08..c40922be 100644 --- a/sleekxmpp/xmlstream/stanzabase.py +++ b/sleekxmpp/xmlstream/stanzabase.py @@ -319,6 +319,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 +328,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): diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 96b47411..7fbc5f9e 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -180,6 +180,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: @@ -226,6 +227,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) @@ -234,11 +236,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: @@ -268,6 +272,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) @@ -339,7 +344,8 @@ class XMLStream(object): args[0].exception(e) elif etype == 'schedule': try: - handler.run(*args) + logging.debug(args) + handler(*args[0]) except: logging.error(traceback.format_exc()) elif etype == 'quit': -- cgit v1.2.3 From 2f1ba368e2adbac27434f70753a6497c0a257301 Mon Sep 17 00:00:00 2001 From: Nathan Fritz Date: Fri, 28 May 2010 19:19:28 -0700 Subject: control-c fixes --- sleekxmpp/xmlstream/scheduler.py | 51 +++++++++++++++++++++++----------------- sleekxmpp/xmlstream/xmlstream.py | 6 ++++- 2 files changed, 35 insertions(+), 22 deletions(-) (limited to 'sleekxmpp/xmlstream') diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py index 7aa59f3d..18d9229e 100644 --- a/sleekxmpp/xmlstream/scheduler.py +++ b/sleekxmpp/xmlstream/scheduler.py @@ -31,11 +31,12 @@ class Task(object): class Scheduler(object): """Threaded scheduler that allows for updates mid-execution unlike http://docs.python.org/library/sched.html#module-sched""" - def __init__(self): + 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: @@ -47,29 +48,37 @@ class Scheduler(object): def _process(self): self.run = True while self.run: - wait = 5 - updated = False - if self.schedule: - wait = self.schedule[0].next - time.time() try: - 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) + wait = 5 + updated = False + if self.schedule: + wait = self.schedule[0].next - time.time() + try: + if wait <= 0.0: + newtask = self.addq.get(False) 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) + 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)) diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 7fbc5f9e..6b92abca 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -76,7 +76,7 @@ class XMLStream(object): self.eventqueue = queue.Queue() self.sendqueue = queue.Queue() - self.scheduler = scheduler.Scheduler() + self.scheduler = scheduler.Scheduler(self.eventqueue) self.namespace_map = {} @@ -149,6 +149,7 @@ class XMLStream(object): 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) @@ -331,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] -- cgit v1.2.3 From 82a3918aa475dbd4abc8941b8c5601f61ab2e508 Mon Sep 17 00:00:00 2001 From: Nathan Fritz Date: Mon, 31 May 2010 03:36:25 -0700 Subject: Scheduler waits too longer, and pubsubstate registration was backwards --- sleekxmpp/xmlstream/scheduler.py | 2 +- sleekxmpp/xmlstream/stanzabase.py | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) (limited to 'sleekxmpp/xmlstream') diff --git a/sleekxmpp/xmlstream/scheduler.py b/sleekxmpp/xmlstream/scheduler.py index 18d9229e..945d9fad 100644 --- a/sleekxmpp/xmlstream/scheduler.py +++ b/sleekxmpp/xmlstream/scheduler.py @@ -49,7 +49,7 @@ class Scheduler(object): self.run = True while self.run: try: - wait = 5 + wait = 1 updated = False if self.schedule: wait = self.schedule[0].next - time.time() diff --git a/sleekxmpp/xmlstream/stanzabase.py b/sleekxmpp/xmlstream/stanzabase.py index c40922be..64020c8f 100644 --- a/sleekxmpp/xmlstream/stanzabase.py +++ b/sleekxmpp/xmlstream/stanzabase.py @@ -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 -- cgit v1.2.3