management/python/lib/qmf/client.py

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License
#

"""
AMQP 1.0 QMF client for the Qpid C++ broker.

This client is based on the Qpid-Proton library which only supports AMQP 1.0, it
is intended for forward-looking projects that want to move to the newer client
libraries. One important feature is that it does not start background threads,
which makes it more suitable for environments that may fork.
"""

from proton import Message
from proton.utils import BlockingConnection, IncomingMessageHandler, ConnectionException
import threading, struct, time

class ReconnectDelays(object):
    """
    An iterable that returns a (possibly unlimited) sequence of delays to for
    successive reconnect attempts.
    """

    def __init__(self, shortest, longest, repeat=True):
        """
        First delay is 0, then `shortest`. Successive delays are doubled up to a
        maximum of `longest`. If `repeat` is a number > 0 then the `longest` value
        is generated `repeat` more times.  If `repeat` is True, the longest
        value is returned without limit.

        For example: ReconnectDelays(.125, 1, True) will generate the following sequence:
            0, .125, .25, .5, 1, 1, 1, 1, 1, 1 ... (forever)
        """
        if shortest <= 0 or shortest > longest or repeat < 0:
            raise ValueError("invalid arguments for reconnect_delays()")
        self.shortest, self.longest, self.repeat = shortest, longest, repeat

    def _generate(self):
        yield 0
        delay = self.shortest
        while delay < self.longest:
            yield delay
            delay *= 2
        yield self.longest
        if self.repeat is True:
            while True:
                yield self.longest
        elif self.repeat:
            for i in xrange(self.repeat):
                yield self.longest

    def __iter__(self):
        return self._generate()


class SyncRequestResponse(IncomingMessageHandler):
    """
    Implementation of the synchronous request-responce (aka RPC) pattern.
    @ivar address: Address for all requests, may be None.
    @ivar connection: Connection for requests and responses.
    """

    def __init__(self, connection, address=None):
        """
        Send requests and receive responses. A single instance can send many requests
        to the same or different addresses.

        @param connection: A L{BlockingConnection}
        @param address: Address for all requests.
            If not specified, each request must have the address property set.
            Sucessive messages may have different addresses.
        """
        super(SyncRequestResponse, self).__init__()
        self.address = address
        self.response = None
        self._cid = 0
        self.lock = threading.Lock()
        self.reconnect(connection)

    def reconnect(self, connection):
        self.connection = connection
        self.sender = self.connection.create_sender(self.address)
        # dynamic=true generates a unique address dynamically for this receiver.
        # credit=1 because we want to receive 1 response message initially.
        self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)

    def _next(self):
        """Get the next correlation ID"""
        self.lock.acquire()
        try:
            self._cid += 1;
            return struct.pack("I", self._cid).decode('ascii')
        finally:
            self.lock.release()

    def call(self, request):
        """
        Send a request message, wait for and return the response message.

        @param request: A L{proton.Message}. If L{self.address} is not set the 
            L{self.address} must be set and will be used.
        """
        return self.wait(self.send(request))

    def send(self, request):
        """
        Send a request and return the correlation_id immediately. Use wait() to get the response.
        @param request: A L{proton.Message}. If L{self.address} is not set the 
            L{self.address} must be set and will be used.
        """
        if not self.address and not request.address:
            raise ValueError("Request message has no address: %s" % request)
        request.reply_to = self.reply_to
        request.correlation_id = self._next()
        self.sender.send(request)
        return request.correlation_id

    def wait(self, correlation_id):
        """Wait for and return a single response to a request previously sent with send()"""
        def wakeup():
            return self.response and (self.response.correlation_id == correlation_id)
        self.connection.wait(wakeup, msg="Waiting for response")
        response = self.response
        self.response = None    # Ready for next response.
        self.receiver.flow(1)   # Set up credit for the next response.
        return response

    @property
    def reply_to(self):
        """Return the dynamic address of our receiver."""
        return self.receiver.remote_source.address

    def on_message(self, event):
        """Called when we receive a message for our receiver."""
        self.response = event.message
        self.connection.container.yield_() # Wake up the wait() loop to handle the message.


class BrokerAgent(object):
    """Proxy for a manageable Qpid broker"""

    @staticmethod
    def connection(url=None, timeout=10, ssl_domain=None, sasl=None):
        """Return a BlockingConnection suitable for use with a BrokerAgent."""
        return BlockingConnection(url,
                                  timeout=timeout,
                                  ssl_domain=ssl_domain,
                                  allowed_mechs=str(sasl.mechs) if sasl else None,
                                  user=str(sasl.user) if sasl else None,
                                  password=str(sasl.password) if sasl else None)

    @staticmethod
    def connect(url=None, timeout=10, ssl_domain=None, sasl=None, reconnect_delays=None):
        """
        Return a BrokerAgent connected with the given parameters.
        @param reconnect_delays: iterable of delays for successive automatic re-connect attempts,
        see class ReconnectDelays. If None there is no automatic re-connect
        """
        f = lambda: BrokerAgent.connection(url, timeout, ssl_domain, sasl)
        ba = BrokerAgent(f())
        ba.make_connection = f
        ba.reconnect_delays = reconnect_delays or []
        return ba

    def __init__(self, connection):
        """
        Create a management node proxy using the given connection.
        @param locales: Default list of locales for management operations.
        @param connection: a L{BlockingConnection} to the management agent.
        """
        path = connection.url.path or "qmf.default.direct"
        self._client = SyncRequestResponse(connection, path)
        self.reconnect_delays = None

    def reconnect(self, connection):
        self._client.reconnect(connection)

    def close(self):
        """Shut down the node"""
        if self._client:
            self._client.connection.close()
            self._client = None

    def __repr__(self):
        return "%s(%s)"%(self.__class__.__name__, self._client.connection.url)

    def _reconnect(self):
        for d in self.reconnect_delays:
            time.sleep(d)
            try:
                self.reconnect(self.make_connection())
                return True
            except ConnectionException:
                pass
        return False

    def _retry(self, f, *args, **kwargs):
        while True:
            try:
                return f(*args, **kwargs)
            except ConnectionException:
                if not self._reconnect():
                    raise

    def _request(self, opcode, content):
        props = {u'method'             : u'request',
                 u'qmf.opcode'         : opcode,
                 u'x-amqp-0-10.app-id' : u'qmf2'}
        return self._client.call(Message(body=content, properties=props, subject="broker"))

    def _method(self, method, arguments=None, addr="org.apache.qpid.broker:broker:amqp-broker"):
        """
        Make a L{proton.Message} containing a QMF method request.
        """
        content = {'_object_id'   : {'_object_name' : addr},
                   '_method_name' : method,
                   '_arguments'   : arguments or {}}
        response = self._retry(self._request, u'_method_request', content)
        if response.properties[u'qmf.opcode'] == u'_exception':
            raise Exception("management error: %r" % response.body['_values'])
        if response.properties[u'qmf.opcode'] != u'_method_response':
            raise Exception("bad response: %r" % response.properties)
        return response.body['_arguments']

    def _gather(self, response):
        items = response.body
        while u'partial' in response.properties:
            response = self._client.wait()
            items += self._client.wait(response.correlation_id).body
        return items

    def _classQuery(self, class_name):
        query = {'_what' : 'OBJECT', '_schema_id' : {'_class_name' : class_name}}
        def f():
            response = self._request(u'_query_request', query)
            if response.properties[u'qmf.opcode'] != u'_query_response':
                raise Exception("bad response")
            return self._gather(response)
        return self._retry(f)

    def _nameQuery(self, object_id):
        query = {'_what'      : 'OBJECT', '_object_id' : {'_object_name' : object_id}}
        def f():
            response = self._request(u'_query_request', query)
            if response.properties[u'qmf.opcode'] != u'_query_response':
                raise Exception("bad response")
            items = self._gather(response)
            if len(items) == 1:
                return items[0]
            return None
        return self._retry(f)

    def _getAll(self, cls):
        return [cls(self, x) for x in self._classQuery(cls.__name__.lower())]

    def _getSingle(self, cls):
        l = self._getAll(cls)
        return l and l[0]

    def _get(self, cls, oid):
        x = self._nameQuery(oid)
        return x and cls(self, x)

    def getBroker(self): return self._getSingle(Broker)
    def getCluster(self): return self._getSingle(Cluster)
    def getHaBroker(self): return self._getSingle(HaBroker)
    def getAllConnections(self): return self._getAll(Connection)
    def getConnection(self, oid): return self._get(Connection, "org.apache.qpid.broker:connection:%s" % oid)
    def getAllSessions(self): return self._getAll(Session)
    def getSession(self, oid): return self._get(Session, "org.apache.qpid.broker:session:%s" % oid)
    def getAllSubscriptions(self): return self._getAll(Subscription)
    def getSubscription(self, oid): return self._get(Subscription, "org.apache.qpid.broker:subscription:%s" % oid)
    def getAllExchanges(self): return self._getAll(Exchange)
    def getExchange(self, name): return self._get(Exchange, "org.apache.qpid.broker:exchange:%s" % name)
    def getAllQueues(self): return self._getAll(Queue)
    def getQueue(self, name): return self._get(Queue, "org.apache.qpid.broker:queue:%s" % name)
    def getAllBindings(self): return self._getAll(Binding)
    def getAllLinks(self): return self._getAll(Link)
    def getAcl(self): return self._getSingle(Acl)
    def getMemory(self): return self._getSingle(Memory)

    def echo(self, sequence = 1, body = "Body"):
        """Request a response to test the path to the management broker"""
        return self._method('echo', {'sequence' : sequence, 'body' : body})

    def queueMoveMessages(self, srcQueue, destQueue, qty):
        """Move messages from one queue to another"""
        self._method("queueMoveMessages", {'srcQueue':srcQueue,'destQueue':destQueue,'qty':qty})

    def queueRedirect(self, sourceQueue, targetQueue):
        """Enable/disable delivery redirect for indicated queues"""
        self._method("queueRedirect", {'sourceQueue':sourceQueue,'targetQueue':targetQueue})

    def setLogLevel(self, level):
        """Set the log level"""
        self._method("setLogLevel", {'level':level})

    def getLogLevel(self):
        """Get the log level"""
        return self._method('getLogLevel')

    def setTimestampConfig(self, receive):
        """Set the message timestamping configuration"""
        self._method("setTimestampConfig", {'receive':receive})

    def getTimestampConfig(self):
        """Get the message timestamping configuration"""
        return self._method('getTimestampConfig')

    def setLogHiresTimestamp(self, logHires):
        """Set the high resolution timestamp in logs"""
        self._method("setLogHiresTimestamp", {'logHires':logHires})

    def getLogHiresTimestamp(self):
        """Get the high resolution timestamp in logs"""
        return self._method('getLogHiresTimestamp')

    def addExchange(self, exchange_type, name, options={}, **kwargs):
        properties = {}
        properties['exchange-type'] = exchange_type
        for k,v in options.items():
            properties[k] = v
        for k,v in kwargs.items():
            properties[k] = v
        args = {'type':       'exchange',
                'name':        name,
                'properties':  properties,
                'strict':      True}
        self._method('create', args)

    def delExchange(self, name):
        args = {'type': 'exchange', 'name': name}
        self._method('delete', args)

    def addQueue(self, name, options={}, **kwargs):
        properties = options
        for k,v in kwargs.items():
            properties[k] = v
        args = {'type':       'queue',
                'name':        name,
                'properties':  properties,
                'strict':      True}
        self._method('create', args)

    def delQueue(self, name, if_empty=True, if_unused=True):
        options = {'if_empty':  if_empty,
                   'if_unused': if_unused}
        args = {'type':        'queue', 
                'name':         name,
                'options':      options}
        self._method('delete', args)

    def bind(self, exchange, queue, key="", options={}, **kwargs):
        properties = options
        for k,v in kwargs.items():
            properties[k] = v
        args = {'type':       'binding',
                'name':       "%s/%s/%s" % (exchange, queue, key),
                'properties':  properties,
                'strict':      True}
        self._method('create', args)

    def unbind(self, exchange, queue, key, **kwargs):
        args = {'type':       'binding',
                'name':       "%s/%s/%s" % (exchange, queue, key),
                'strict':      True}
        self._method('delete', args)

    def reloadAclFile(self):
        self._method('reloadACLFile', {}, "org.apache.qpid.acl:acl:org.apache.qpid.broker:broker:amqp-broker")

    def acl_lookup(self, userName, action, aclObj, aclObjName, propMap):
        args = {'userId':      userName,
                'action':      action,
                'object':      aclObj,
                'objectName':  aclObjName,
                'propertyMap': propMap}
        return self._method('Lookup', args, "org.apache.qpid.acl:acl:org.apache.qpid.broker:broker:amqp-broker")

    def acl_lookupPublish(self, userName, exchange, key):
        args = {'userId':       userName,
                'exchangeName': exchange,
                'routingKey':   key}
        return self._method('LookupPublish', args, "org.apache.qpid.acl:acl:org.apache.qpid.broker:broker:amqp-broker")

    def Redirect(self, sourceQueue, targetQueue):
        args = {'sourceQueue': sourceQueue,
                'targetQueue': targetQueue}
        return self._method('queueRedirect', args, "org.apache.qpid.broker:broker:amqp-broker")

    def create(self, _type, name, properties={}, strict=False):
        """Create an object of the specified type"""
        args = {'type': _type,
                'name': name,
                'properties': properties,
                'strict': strict}
        return self._method('create', args)

    def delete(self, _type, name, options):
        """Delete an object of the specified type"""
        args = {'type': _type,
                'name': name,
                'options': options}
        return self._method('delete', args)

    def list(self, _type):
        """List objects of the specified type"""
        return [i["_values"] for i in self._classQuery(_type.lower())]

    def query(self, _type, oid):
        """Query the current state of an object"""
        return self._get(self, _type, oid)


class BrokerObject(object):
    def __init__(self, broker, content):
        self.broker = broker
        self.content = content

    @property
    def values(self):
        return self.content['_values']

    def __getattr__(self, key):
        return self.values.get(key)

    def getObjectId(self):
        return self.content['_object_id']['_object_name']

    def getAttributes(self):
        return self.values

    def getCreateTime(self):
        return self.content['_create_ts']

    def getDeleteTime(self):
        return self.content['_delete_ts']

    def getUpdateTime(self):
        return self.content['_update_ts']

    def update(self):
        """
        Reload the property values from the agent.
        """
        refreshed = self.broker._get(self.__class__, self.getObjectId())
        if refreshed:
            self.content = refreshed.content
            self.values = self.content['_values']
        else:
            raise Exception("No longer exists on the broker")

    def __repr__(self):
        return "%s(%s)"%(self.__class__.__name__, self.content)

    def __str__(self):
        return self.getObjectId()


class Broker(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Cluster(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class HaBroker(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Memory(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Connection(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

    def close(self):
        self.broker._method("close", {}, "org.apache.qpid.broker:connection:%s" % self.address)

class Session(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Subscription(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Exchange(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Binding(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Queue(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

    def purge(self, request):
        """Discard all or some messages on a queue"""
        self.broker._method("purge", {'request':request}, "org.apache.qpid.broker:queue:%s" % self.name)

    def reroute(self, request, useAltExchange, exchange, filter={}):
        """Remove all or some messages on this queue and route them to an exchange"""
        self.broker._method("reroute", {'request':request,'useAltExchange':useAltExchange,'exchange':exchange,'filter':filter},
                          "org.apache.qpid.broker:queue:%s" % self.name)

class Link(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)

class Acl(BrokerObject):
    def __init__(self, broker, values):
        BrokerObject.__init__(self, broker, values)