updated scheduler

Author: Matt George

pyres/__init__.py

@@ -183,7 +183,7 @@ def enqueue_at(self, timestamp, klass, *args):
 
     def delayed_push(self, timestamp, item):
         key = int(time.mktime(timestamp.timetuple()))
-        self.redis.push('resque:delayed:%s' % key, ResQ.encode(item))
+        self.redis.rpush('resque:delayed:%s' % key, ResQ.encode(item))
         self.redis.zadd('resque:delayed_queue_schedule', key, key)
 
     def delayed_queue_peek(self, start, count):
@@ -207,7 +207,7 @@ def next_delayed_timestamp(self):
     def next_item_for_timestamp(self, timestamp):
         #key = int(time.mktime(timestamp.timetuple()))
         key = "resque:delayed:%s" % timestamp
-        ret = self.redis.pop(key)
+        ret = self.redis.lpop(key)
         item = None
         if ret:
             item = ResQ.decode(ret)

pyres/scheduler.py

@@ -15,21 +15,27 @@ def __init__(self, server="localhost:6379", password=None):
             raise Exception("Bad server argument")
 
     def register_signal_handlers(self):
+        print 'registering signals'
         signal.signal(signal.SIGTERM, self.schedule_shutdown)
         signal.signal(signal.SIGINT, self.schedule_shutdown)
         signal.signal(signal.SIGQUIT, self.schedule_shutdown)
 
-    def schedule_shutdown(self):
+    def schedule_shutdown(self, signal, frame):
+        print 'shutting down started'
         self._shutdown = True
 
-    def run(self):
+    def __call__(self):
+        print 'starting up'
         self.register_signal_handlers()
         #self.load_schedule()
+        print 'looking for delayed items'
         while True:
             if self._shutdown is True:
                 break
             self.handle_delayed_items()
+            print 'sleeping'
             time.sleep(5)
+        print 'shutting down complete'
 
     def next_timestamp(self):
         while True:
@@ -50,12 +56,18 @@ def next_item(self, timestamp):
 
     def handle_delayed_items(self):
         for timestamp in self.next_timestamp():
+            print 'handling timestamp: %s' % timestamp
             for item in self.next_item(timestamp):
-                print 'queueing item'
+                print 'queueing item %s' % item
                 klass = item['class']
                 queue = item['queue']
                 args = item['args']
-                self.resq.enqueue_from_string(klass, queue, args)
+                self.resq.enqueue_from_string(klass, queue, *args)
 
 
+    @classmethod
+    def run(cls, server, password=None):
+        sched = cls(server=server, password=password)
+        sched()
 
+    

scripts/pyres_scheduler

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+from pyres.scheduler import Scheduler
+def main():
+    from optparse import OptionParser
+    usage = "usage: %prog [options] arg1"
+    parser = OptionParser(usage=usage)
+    #parser.add_option("-q", dest="queue_list")
+    parser.add_option("--host", dest="host", default="localhost")
+    parser.add_option("--port",dest="port",type="int", default=6379)
+    (options,args) = parser.parse_args()
+    server = '%s:%s' % (options.host,options.port)
+    Scheduler.run(server)
+
+if __name__ == "__main__":
+    main()