diff options
author | Marko Kreen | 2009-09-09 10:52:58 +0000 |
---|---|---|
committer | Marko Kreen | 2009-09-09 10:52:58 +0000 |
commit | 8647a60005315bf0c15c966213322ea1baf7bff6 (patch) | |
tree | abc9bb744cc1efb5e783321cfb09e75b97635cb0 /python/pgq/coopconsumer.py | |
parent | e436a281174fce7b16e1a2dd837e5003a7faa097 (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.py | 63 |
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]) + |