Merge commit 'acc9414ab51a9692e5660dbaad175564d6e67582' into refactor_merge

Conflicts:
	pyres/worker.py
	src/resweb/server.py

Author: Matt George

pyres/__init__.py

@@ -26,26 +26,26 @@ def __init__(self, server="localhost:6379"):
 
     def push(self, queue, item):
         self.watch_queue(queue)
-        self.redis.push("queue:%s" % queue, ResQ.encode(item))
+        self.redis.push("resque:queue:%s" % queue, ResQ.encode(item))
 
     def pop(self, queue):
-        ret = self.redis.pop("queue:%s" % queue)
+        ret = self.redis.pop("resque:queue:%s" % queue)
         if ret:
             return ResQ.decode(ret)
         return ret
 
     def size(self, queue):
-        return int(self.redis.llen("queue:%s" % queue))
+        return int(self.redis.llen("resque:queue:%s" % queue))
 
     def watch_queue(self, queue):
         if queue in self._watched_queues:
             return
         else:
-            if self.redis.sadd('queues',str(queue)):
+            if self.redis.sadd('resque:queues',str(queue)):
                 self._watched_queues.add(queue)
 
     def peek(self, queue, start=0, count=1):
-        return self.list_range('queue:%s' % queue, start, count)
+        return self.list_range('resque:queue:%s' % queue, start, count)
 
     def list_range(self, key, start, count):
         items = self.redis.lrange(key,start,start+count-1)
@@ -81,7 +81,7 @@ def enqueue_from_string(self, klass_as_string, queue, *args):
         self.push(queue, {'class':klass_as_string,'args':args})
 
     def queues(self):
-        return self.redis.smembers("queues")
+        return self.redis.smembers("resque:queues")
 
     def info(self):
         pending = 0
@@ -118,8 +118,8 @@ def working(self):
     def remove_queue(self, queue):
         if queue in self._watched_queues:
             self._watched_queues.remove(queue)
-        self.redis.srem('queues',queue)
-        del self.redis['queue:%s' % queue]
+        self.redis.srem('resque:queues',queue)
+        del self.redis['resque:queue:%s' % queue]
 
     @classmethod
     def encode(cls, item):
@@ -146,7 +146,7 @@ def _enqueue(cls, klass, *args):
 class Stat(object):
     def __init__(self, name, resq):
         self.name = name
-        self.key = "stat:%s" % self.name
+        self.key = "resque:stat:%s" % self.name
         self.resq = resq
 
     def get(self):

pyres/extensions.py

@@ -3,6 +3,8 @@
 except:
     import sys
     sys.exit("multiprocessing was not available")
+import os, datetime, time, signal
+from pyres import ResQ
 
 from pyres.exceptions import NoQueueError
 from pyres.worker import Worker

pyres/worker.py

@@ -6,6 +6,7 @@
 import os, sys
 import time
 import simplejson
+
 class Worker(object):
     def __init__(self, queues=[], server="localhost:6379"):
         self.queues = queues
@@ -26,7 +27,7 @@ def validate_queues(self):
             raise NoQueueError("Please give each worker at least one queue.")
 
     def register_worker(self):
-        self.resq.redis.sadd('workers',str(self))
+        self.resq.redis.sadd('resque:workers',str(self))
         #self.resq._redis.add("worker:#{self}:started", Time.now.to_s)
         self.started = datetime.datetime.now()
         #Stat.clear("processed:#{self}")
@@ -47,7 +48,7 @@ def _get_started(self):
     started = property(_get_started, _set_started)
 
     def unregister_worker(self):
-        self.resq.redis.srem('workers',str(self))
+        self.resq.redis.srem('resque:workers',str(self))
         self.started = None
 
     def startup(self):
@@ -141,14 +142,14 @@ def working_on(self, job):
             'payload': job._payload
         }
         data = simplejson.dumps(data)
-        self.resq.redis["worker:%s" % str(self)] = data
+        self.resq.redis["resque:worker:%s" % str(self)] = data
         print "worker:%s" % str(self)
         print self.resq.redis["worker:%s" % str(self)]
 
     def done_working(self):
         print 'done working'
         self.processed()
-        self.resq.redis.delete("worker:%s" % str(self))
+        self.resq.redis.delete("resque:worker:%s" % str(self))
 
     def processed(self):
         total_processed = Stat("processed", self.resq)
@@ -167,7 +168,7 @@ def failed(self):
     def get_failed(self):
         return Stat("failed:%s" % self, self.resq).get()
     def job(self):
-        data = self.resq.redis.get("worker:%s" % self)
+        data = self.resq.redis.get("resque:worker:%s" % self)
         if data:
             return ResQ.decode(data)
         return {}
@@ -190,7 +191,7 @@ def all(cls, host="localhost:6379"):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
-        return [Worker.find(w,resq) for w in resq.redis.smembers('workers')]
+        return [Worker.find(w,resq) for w in resq.redis.smembers('resque:workers')]
 
     @classmethod
     def working(cls, host):
@@ -200,7 +201,8 @@ def working(cls, host):
             resq = host
         total = []
         for key in Worker.all(host):
-            total.append('worker:%s' % key)
+            if Worker.exists(key, resq):
+                total.append('resque:worker:%s' % key)
         names = []
         for key in total:
             value = resq.redis.get(key)
@@ -221,7 +223,7 @@ def find(cls, worker_id, resq):
 
     @classmethod
     def exists(cls, worker_id, resq):
-        return resq.redis.sismember('workers', worker_id)
+        return resq.redis.sismember('resque:workers', worker_id)
 
 
 if __name__ == "__main__":
@@ -236,4 +238,4 @@ def exists(cls, worker_id, resq):
     queues = options.queue_list.split(',')
     import sys
     sys.path.insert(0,'/Users/mgeorge/dev/pyres/src')
-    Worker.run(queues, options.server)
+    Worker.run(queues, options.server)