summaryrefslogtreecommitdiff
path: root/python/pgq/coopconsumer.py
diff options
context:
space:
mode:
authorMarko Kreen2009-09-09 10:52:58 +0000
committerMarko Kreen2009-09-09 10:52:58 +0000
commit8647a60005315bf0c15c966213322ea1baf7bff6 (patch)
treeabc9bb744cc1efb5e783321cfb09e75b97635cb0 /python/pgq/coopconsumer.py
parente436a281174fce7b16e1a2dd837e5003a7faa097 (diff)
pgq.CoopConsumer for Python
Simply wrapper around Consumer which redirects few calls to pgq_coop schema.
Diffstat (limited to 'python/pgq/coopconsumer.py')
-rw-r--r--python/pgq/coopconsumer.py63
1 files changed, 63 insertions, 0 deletions
diff --git a/python/pgq/coopconsumer.py b/python/pgq/coopconsumer.py
new file mode 100644
index 00000000..3e6cdec8
--- /dev/null
+++ b/python/pgq/coopconsumer.py
@@ -0,0 +1,63 @@
+
+"""PgQ cooperative consumer for Python.
+"""
+
+from pgq.consumer import Consumer
+
+__all__ = ['CoopConsumer']
+
+class CoopConsumer(Consumer):
+ """Cooperative Consumer base class.
+
+ There will be one dbscript process per subconsumer.
+ """
+
+ def __init__(self, service_name, db_name, args):
+ """Initialize new subconsumer.
+
+ @param service_name: service_name for DBScript
+ @param db_name: name of database for get_database()
+ @param args: cmdline args for DBScript
+ """
+
+ Consumer.__init__(self, service_name, db_name, args)
+
+ self.subconsumer_name = self.cf.get("subconsumer_name")
+
+ def register_consumer(self):
+ """Registration for subconsumer."""
+
+ self.log.info("Registering consumer on source queue")
+ db = self.get_database(self.db_name)
+ cx = db.cursor()
+ cx.execute("select pgq_coop.register_subconsumer(%s, %s, %s)",
+ [self.queue_name, self.consumer_name, self.subconsumer_name])
+ res = cx.fetchone()[0]
+ db.commit()
+
+ return res
+
+ def unregister_consumer(self):
+ """Unregistration for subconsumer."""
+
+ self.log.info("Unregistering consumer from source queue")
+ db = self.get_database(self.db_name)
+ cx = db.cursor()
+ cx.execute("select pgq_coop.unregister_consumer(%s, %s, %s)",
+ [self.queue_name, self.consumer_name, self.subconsumer_name])
+ db.commit()
+
+
+ def _load_next_batch(self, curs):
+ """Allocate next batch. (internal)"""
+
+ q = "select pgq_coop.next_batch(%s, %s, %s)"
+ curs.execute(q, [self.queue_name, self.consumer_name, self.subconsumer_name])
+ return curs.fetchone()[0]
+
+ def _finish_batch(self, curs, batch_id, list):
+ """Finish batch. (internal)"""
+
+ self._flush_retry(curs, batch_id, list)
+ curs.execute("select pgq_coop.finish_batch(%s)", [batch_id])
+