commit f4fbb0fa72b3ab8891584c0d208955c703f31072
Author: Sofia Papagiannaki <papagian@gmail.com>
Date:   Tue Sep 11 15:09:32 2012 +0300

    incorporate AMQPClient instead of synnefo.lib.queue methods for sending queue messages

diff --git a/snf-pithos-app/README b/snf-pithos-app/README
index 7a674b1..2fdf996 100644
--- a/snf-pithos-app/README
+++ b/snf-pithos-app/README
@@ -38,7 +38,8 @@ PITHOS_BACKEND_BLOCK_MODULE      pithos.backends.lib.hashfiler
 PITHOS_BACKEND_BLOCK_PATH        /tmp/pithos-data/                                 Map and block storage path
 PITHOS_BACKEND_BLOCK_UMASK       0o022                                             Map and block storage umask
 PITHOS_BACKEND_QUEUE_MODULE      None                                              Use ``pithos.backends.lib.rabbitmq`` to enable
-PITHOS_BACKEND_QUEUE_CONNECTION  None                                              Format like ``rabbitmq://guest:guest@localhost:5672/pithos``
+PITHOS_BACKEND_QUEUE_HOSTS       None                                              Format like [``amqp://guest:guest@localhost:5672``
+PITHOS_BACKEND_QUEUE_EXCHANGE    pithos
 PITHOS_BACKEND_QUOTA             50 GB (50 * 1024 ** 3)                            Default user quota
 PITHOS_BACKEND_VERSIONING        auto                                              Default versioning policy for containers
 PITHOS_UPDATE_MD5                True                                              Update object checksums when using hashmaps
diff --git a/snf-pithos-app/conf/20-snf-pithos-app-settings.conf b/snf-pithos-app/conf/20-snf-pithos-app-settings.conf
index b1cfd0e..aa3c999 100644
--- a/snf-pithos-app/conf/20-snf-pithos-app-settings.conf
+++ b/snf-pithos-app/conf/20-snf-pithos-app-settings.conf
@@ -20,8 +20,10 @@ PITHOS_BACKEND_BLOCK_PATH = '/usr/share/synnefo/pithos/data'
 #PITHOS_BACKEND_QUEUE_MODULE = None
 # Example: 'pithos.backends.lib.rabbitmq'
 #
-#PITHOS_BACKEND_QUEUE_CONNECTION = None
-# Example: 'rabbitmq://guest:guest@localhost:5672/pithos'
+#PITHOS_BACKEND_QUEUE_HOSTS = None
+# Example: ['ammq://guest:guest@localhost:5672/']
+#
+#PITHOS_BACKEND_QUEUE_EXHANGE = 'pithos'
 
 # Default setting for new accounts.
 #PITHOS_BACKEND_QUOTA = 50 * 1024 * 1024 * 1024
diff --git a/snf-pithos-app/pithos/api/dispatch.py b/snf-pithos-app/pithos/api/dispatch.py
index 3f763e8..521b467 100644
--- a/snf-pithos-app/pithos/api/dispatch.py
+++ b/snf-pithos-app/pithos/api/dispatch.py
@@ -1,7 +1,8 @@
 from pithos.api.settings import (BACKEND_DB_MODULE, BACKEND_DB_CONNECTION,
                                  BACKEND_BLOCK_MODULE, BACKEND_BLOCK_PATH,
                                  BACKEND_BLOCK_UMASK,
-                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_CONNECTION,
+                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_HOSTS,
+                                 BACKEND_QUEUE_EXCHANGE,
                                  BACKEND_QUOTA, BACKEND_VERSIONING)
 from pithos.backends import connect_backend
 from pithos.api.util import hashmap_md5
@@ -25,7 +26,8 @@ def update_md5(m):
                               block_path=BACKEND_BLOCK_PATH,
                               block_umask=BACKEND_BLOCK_UMASK,
                               queue_module=BACKEND_QUEUE_MODULE,
-                              queue_connection=BACKEND_QUEUE_CONNECTION)
+                              queue_hosts=BACKEND_QUEUE_HOSTS,
+                              queue_exchange=BACKEND_QUEUE_EXHANGE)
     backend.default_policy['quota'] = BACKEND_QUOTA
     backend.default_policy['versioning'] = BACKEND_VERSIONING
 
diff --git a/snf-pithos-app/pithos/api/management/commands/storagequota.py b/snf-pithos-app/pithos/api/management/commands/storagequota.py
index 34d5d15..7e6f12d 100644
--- a/snf-pithos-app/pithos/api/management/commands/storagequota.py
+++ b/snf-pithos-app/pithos/api/management/commands/storagequota.py
@@ -38,7 +38,8 @@ from django.core.management.base import BaseCommand, CommandError
 from pithos.api.settings import (BACKEND_DB_MODULE, BACKEND_DB_CONNECTION,
                                  BACKEND_BLOCK_MODULE, BACKEND_BLOCK_PATH,
                                  BACKEND_BLOCK_UMASK,
-                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_CONNECTION,
+                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_HOSTS,
+                                 BACKEND_QUEUE_EXCHANGE,
                                  BACKEND_QUOTA, BACKEND_VERSIONING)
 from pithos.backends import connect_backend
 
@@ -72,7 +73,8 @@ class Command(BaseCommand):
                                   block_path=BACKEND_BLOCK_PATH,
                                   block_umask=BACKEND_BLOCK_UMASK,
                                   queue_module=BACKEND_QUEUE_MODULE,
-                                  queue_connection=BACKEND_QUEUE_CONNECTION)
+                                  queue_connection=BACKEND_QUEUE_HOSTS,
+                                  queue_exchange=BACKEND_QUEUE_EXCHANGE)
         backend.default_policy['quota'] = BACKEND_QUOTA
         backend.default_policy['versioning'] = BACKEND_VERSIONING
         if quota is not None:
diff --git a/snf-pithos-app/pithos/api/settings.py b/snf-pithos-app/pithos/api/settings.py
index 4b6ff12..7b78707 100644
--- a/snf-pithos-app/pithos/api/settings.py
+++ b/snf-pithos-app/pithos/api/settings.py
@@ -37,7 +37,8 @@ BACKEND_BLOCK_UMASK = getattr(settings, 'PITHOS_BACKEND_BLOCK_UMASK', 0o022)
 # Queue for billing.
 BACKEND_QUEUE_MODULE = getattr(settings, 'PITHOS_BACKEND_QUEUE_MODULE',
                                None)  # Example: 'pithos.backends.lib.rabbitmq'
-BACKEND_QUEUE_CONNECTION = getattr(settings, 'PITHOS_BACKEND_QUEUE_CONNECTION', None)  # Example: 'rabbitmq://guest:guest@localhost:5672/pithos'
+BACKEND_QUEUE_HOSTS = getattr(settings, 'PITHOS_BACKEND_QUEUE_HOSTS', None) # Example: "['amqp://guest:guest@localhost:5672']"
+BACKEND_QUEUE_EXCHANGE = getattr(settings, 'PITHOS_BACKEND_QUEUE_EXCHANGE', 'pithos')
 
 # Default setting for new accounts.
 BACKEND_QUOTA = getattr(
diff --git a/snf-pithos-app/pithos/api/util.py b/snf-pithos-app/pithos/api/util.py
index 98435a3..a2df91f 100644
--- a/snf-pithos-app/pithos/api/util.py
+++ b/snf-pithos-app/pithos/api/util.py
@@ -59,7 +59,8 @@ from pithos.api.short_url import encode_url
 from pithos.api.settings import (BACKEND_DB_MODULE, BACKEND_DB_CONNECTION,
                                  BACKEND_BLOCK_MODULE, BACKEND_BLOCK_PATH,
                                  BACKEND_BLOCK_UMASK,
-                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_CONNECTION,
+                                 BACKEND_QUEUE_MODULE, BACKEND_QUEUE_HOSTS,
+                                 BACKEND_QUEUE_EXCHANGE,
                                  BACKEND_QUOTA, BACKEND_VERSIONING,
                                  AUTHENTICATION_URL, AUTHENTICATION_USERS,
                                  SERVICE_TOKEN, COOKIE_NAME)
@@ -856,7 +857,8 @@ def _get_backend():
                               block_path=BACKEND_BLOCK_PATH,
                               block_umask=BACKEND_BLOCK_UMASK,
                               queue_module=BACKEND_QUEUE_MODULE,
-                              queue_connection=BACKEND_QUEUE_CONNECTION)
+                              queue_hosts=BACKEND_QUEUE_HOSTS,
+                              queue_exchange=BACKEND_QUEUE_EXCHANGE)
     backend.default_policy['quota'] = BACKEND_QUOTA
     backend.default_policy['versioning'] = BACKEND_VERSIONING
     return backend
diff --git a/snf-pithos-backend/pithos/backends/lib/rabbitmq/queue.py b/snf-pithos-backend/pithos/backends/lib/rabbitmq/queue.py
index ff97b89..c4d3838 100644
--- a/snf-pithos-backend/pithos/backends/lib/rabbitmq/queue.py
+++ b/snf-pithos-backend/pithos/backends/lib/rabbitmq/queue.py
@@ -31,23 +31,32 @@
 # interpreted as representing official policies, either expressed
 # or implied, of GRNET S.A.
 
-from synnefo.lib.queue import exchange_connect, exchange_send, exchange_close, Receipt
+import json
 
+from synnefo.lib.amqp import AMQPClient
+from synnefo.lib.queue import Receipt
 
 class Queue(object):
     """Queue.
-       Required constructor parameters: exchange, client_id.
+       Required constructor parameters: hosts, exchange, client_id.
     """
 
     def __init__(self, **params):
-        exchange = params['exchange']
-        self.conn = exchange_connect(exchange)
+        hosts = params['hosts']
+        self.exchange = params['exchange']
         self.client_id = params['client_id']
 
-    def send(self, message_key, user, instance, resource, value, details):
-        body = Receipt(
-            self.client_id, user, instance, resource, value, details).format()
-        exchange_send(self.conn, message_key, body)
+        self.client = AMQPClient(hosts=hosts)
+        self.client.connect()
+
+        self.client.exchange_declare(exchange=self.exchange,
+                                     type='topic')
 
+    def send(self, message_key, user, instance, resource, value, details):
+        body = Receipt(self.client_id, user, instance, resource, value, details).format()
+        self.client.basic_publish(exchange=self.exchange,
+                                  routing_key=message_key,
+                                  body=json.dumps(body))
+    
     def close(self):
-        exchange_close(self.conn)
+        self.client.close()
diff --git a/snf-pithos-backend/pithos/backends/modular.py b/snf-pithos-backend/pithos/backends/modular.py
index 7f02acd..d8a2455 100644
--- a/snf-pithos-backend/pithos/backends/modular.py
+++ b/snf-pithos-backend/pithos/backends/modular.py
@@ -79,7 +79,8 @@ DEFAULT_BLOCK_MODULE = 'pithos.backends.lib.hashfiler'
 DEFAULT_BLOCK_PATH = 'data/'
 DEFAULT_BLOCK_UMASK = 0o022
 #DEFAULT_QUEUE_MODULE = 'pithos.backends.lib.rabbitmq'
-#DEFAULT_QUEUE_CONNECTION = 'rabbitmq://guest:guest@localhost:5672/pithos'
+#DEFAULT_QUEUE_HOSTS = '[amqp://guest:guest@localhost:5672]'
+#DEFAULT_QUEUE_EXCHANGE = 'pithos'
 
 QUEUE_MESSAGE_KEY_PREFIX = 'pithos.%s'
 QUEUE_CLIENT_ID = 'pithos'
@@ -127,15 +128,17 @@ class ModularBackend(BaseBackend):
 
     def __init__(self, db_module=None, db_connection=None,
                  block_module=None, block_path=None, block_umask=None,
-                 queue_module=None, queue_connection=None):
+                 queue_module=None, queue_hosts=None,
+                 queue_exchange=None):
         db_module = db_module or DEFAULT_DB_MODULE
         db_connection = db_connection or DEFAULT_DB_CONNECTION
         block_module = block_module or DEFAULT_BLOCK_MODULE
         block_path = block_path or DEFAULT_BLOCK_PATH
         block_umask = block_umask or DEFAULT_BLOCK_UMASK
         #queue_module = queue_module or DEFAULT_QUEUE_MODULE
-        #queue_connection = queue_connection or DEFAULT_QUEUE_CONNECTION
-
+        #queue_hosts = queue_hosts or DEFAULT_QUEUE_HOSTS
+        #queue_exchange = queue_exchange or DEFAULT_QUEUE_EXCHANGE
+		
         self.hash_algorithm = 'sha256'
         self.block_size = 4 * 1024 * 1024  # 4MB
 
@@ -165,9 +168,10 @@ class ModularBackend(BaseBackend):
                   'umask': block_umask}
         self.store = self.block_module.Store(**params)
 
-        if queue_module and queue_connection:
+        if queue_module and queue_hosts:
             self.queue_module = load_module(queue_module)
-            params = {'exchange': queue_connection,
+            params = {'hosts': queue_hosts,
+            		  'exchange': queue_exchange,
                       'client_id': QUEUE_CLIENT_ID}
             self.queue = self.queue_module.Queue(**params)
         else:
@@ -1221,15 +1225,15 @@ class ModularBackend(BaseBackend):
         details.update({'user': user})
         logger.debug("_report_object_change: %s %s %s %s", user,
                      account, path, details)
-        self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % (
-            'object',), account, QUEUE_INSTANCE_ID, 'object', path, details))
+        self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % ('object',),
+        					  account, QUEUE_INSTANCE_ID, 'object', path, details))
 
     def _report_sharing_change(self, user, account, path, details={}):
         logger.debug("_report_permissions_change: %s %s %s %s",
                      user, account, path, details)
         details.update({'user': user})
-        self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % ('sharing',), account,
-                             QUEUE_INSTANCE_ID, 'sharing', path, details))
+        self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % ('sharing',),
+        					  account, QUEUE_INSTANCE_ID, 'sharing', path, details))
 
     # Policy functions.
 
diff --git a/snf-pithos-tools/pithos/tools/dispatcher.py b/snf-pithos-tools/pithos/tools/dispatcher.py
index 5a8309f..acaa723 100755
--- a/snf-pithos-tools/pithos/tools/dispatcher.py
+++ b/snf-pithos-tools/pithos/tools/dispatcher.py
@@ -35,9 +35,9 @@
 
 import sys
 import logging
+import json
 
-from synnefo.lib.queue import (exchange_connect, exchange_close,
-                               exchange_send, exchange_route, queue_callback, queue_start)
+from synnefo.lib.amqp import AMQPClient
 
 from optparse import OptionParser
 
@@ -94,12 +94,21 @@ def main():
         level=logging.DEBUG if DEBUG else logging.INFO)
     logger = logging.getLogger('dispatcher')
 
-    exchange = 'rabbitmq://%s:%s@%s:%s/%s' % (
-        opts.user, opts.password, opts.host, opts.port, opts.exchange)
-    connection = exchange_connect(exchange)
+    host =  'amqp://%s:%s@%s:%s' % (opts.user, opts.password, opts.host, opts.port)
+    queue = opts.queue
+    key = opts.key
+    exchange = opts.exchange
+    
+    client = AMQPClient(hosts=[host])
+    client.connect()
+
     if opts.test:
-        exchange_send(connection, opts.key, {"test": "0123456789"})
-        exchange_close(connection)
+        client.exchange_declare(exchange=exchange,
+                                type='topic')
+        client.basic_publish(exchange=exchange,
+                             routing_key=key,
+                             body= json.dumps({"test": "0123456789"}))
+        client.close()
         sys.exit()
 
     callback = None
@@ -110,18 +119,26 @@ def main():
             cb_module = sys.modules[cb[0]]
             callback = getattr(cb_module, cb[1])
 
-    def handle_message(msg):
-        print msg
+    def handle_message(client, msg):
         logger.debug('%s', msg)
         if callback:
             callback(msg)
+        client.basic_ack(msg)
+
+    client.queue_declare(queue=queue)
+    client.queue_bind(queue=queue,
+                      exchange=exchange,
+                      routing_key=key)
+
+    client.basic_consume(queue=queue, callback=handle_message)
 
-    exchange_route(connection, opts.key, opts.queue)
-    queue_callback(connection, opts.queue, handle_message)
     try:
-        queue_start(connection)
+        while True:
+            client.basic_wait()
     except KeyboardInterrupt:
         pass
+    finally:
+        client.close()
 
 
 if __name__ == '__main__':
