/ - Diff - Synnefo - Greek Research and Technology Network's projects

     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

     #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

     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
-...
                                   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

     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
-...
                                       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:

     # 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(

     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)

     # 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()

     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'
-...
         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
-...
                       '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:
-...
             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.

     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
-...
             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
-...
                 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__':

Synnefo

Revision 4cfccdd2