path: root/tools/src/py/qpid-config

#!/usr/bin/env python

#
# 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.
#
import pdb

import os
from optparse import OptionParser, OptionGroup, IndentedHelpFormatter
import sys
import locale

home = os.environ.get("QPID_TOOLS_HOME", os.path.normpath("/usr/share/qpid-tools"))
sys.path.append(os.path.join(home, "python"))

from qpid.messaging import Connection
from qpidtoollibs import BrokerAgent

usage = """
Usage:  qpid-config [OPTIONS]
        qpid-config [OPTIONS] exchanges [filter-string]
        qpid-config [OPTIONS] queues    [filter-string]
        qpid-config [OPTIONS] add exchange <type> <name> [AddExchangeOptions]
        qpid-config [OPTIONS] del exchange <name>
        qpid-config [OPTIONS] add queue <name> [AddQueueOptions]
        qpid-config [OPTIONS] del queue <name> [DelQueueOptions]
        qpid-config [OPTIONS] bind   <exchange-name> <queue-name> [binding-key]
                  <for type xml>     [-f -|filename]
                  <for type header>  [all|any] k1=v1 [, k2=v2...]
        qpid-config [OPTIONS] unbind <exchange-name> <queue-name> [binding-key]
        qpid-config [OPTIONS] reload-acl"""

description = """
Examples:

$ qpid-config add queue q
$ qpid-config add exchange direct d -a localhost:5672
$ qpid-config exchanges -b 10.1.1.7:10000
$ qpid-config queues -b guest/guest@broker-host:10000

Add Exchange <type> values:

    direct     Direct exchange for point-to-point communication
    fanout     Fanout exchange for broadcast communication
    topic      Topic exchange that routes messages using binding keys with wildcards
    headers    Headers exchange that matches header fields against the binding keys
    xml        XML Exchange - allows content filtering using an XQuery


Queue Limit Actions:

    none (default) - Use broker's default policy
    reject         - Reject enqueued messages
    flow-to-disk   - Page messages to disk
    ring           - Replace oldest unacquired message with new
    ring-strict    - Replace oldest message, reject if oldest is acquired

Replication levels:

    none           - no replication
    configuration  - replicate queue and exchange existence and bindings, but not messages.
    all            - replicate configuration and messages
"""

REPLICATE_LEVELS= ["none", "configuration", "all"]

class Config:
    def __init__(self):
        self._recursive         = False
        self._host              = "localhost"
        self._connTimeout       = 10
        self._ignoreDefault     = False
        self._altern_ex         = None
        self._durable           = False
        self._replicate       = None
        self._clusterDurable    = False
        self._if_empty          = True
        self._if_unused         = True
        self._fileCount         = 8
        self._fileSize          = 24
        self._maxQueueSize      = None
        self._maxQueueCount     = None
        self._limitPolicy       = None
        self._msgSequence       = False
        self._lvq_key           = None
        self._ive               = False
        self._eventGeneration   = None
        self._file              = None
        self._flowStopCount     = None
        self._flowResumeCount   = None
        self._flowStopSize      = None
        self._flowResumeSize    = None
        self._msgGroupHeader    = None
        self._sharedMsgGroup    = False
        self._extra_arguments   = []
        self._start_replica    = None
        self._returnCode        = 0

config = Config()
conn_options = {}

FILECOUNT = "qpid.file_count"
FILESIZE  = "qpid.file_size"
MAX_QUEUE_SIZE  = "qpid.max_size"
MAX_QUEUE_COUNT  = "qpid.max_count"
POLICY_TYPE  = "qpid.policy_type"
CLUSTER_DURABLE = "qpid.persist_last_node"
LVQ_KEY = "qpid.last_value_queue_key"
MSG_SEQUENCE = "qpid.msg_sequence"
IVE = "qpid.ive"
QUEUE_EVENT_GENERATION = "qpid.queue_event_generation"
FLOW_STOP_COUNT   = "qpid.flow_stop_count"
FLOW_RESUME_COUNT = "qpid.flow_resume_count"
FLOW_STOP_SIZE    = "qpid.flow_stop_size"
FLOW_RESUME_SIZE  = "qpid.flow_resume_size"
MSG_GROUP_HDR_KEY = "qpid.group_header_key"
SHARED_MSG_GROUP  = "qpid.shared_msg_group"
REPLICATE = "qpid.replicate"
#There are various arguments to declare that have specific program
#options in this utility. However there is now a generic mechanism for
#passing arguments as well. The SPECIAL_ARGS list contains the
#arguments for which there are specific program options defined
#i.e. the arguments for which there is special processing on add and
#list
SPECIAL_ARGS=[
    FILECOUNT,FILESIZE,MAX_QUEUE_SIZE,MAX_QUEUE_COUNT,POLICY_TYPE,CLUSTER_DURABLE,
    LVQ_KEY,MSG_SEQUENCE,IVE,QUEUE_EVENT_GENERATION,
    FLOW_STOP_COUNT,FLOW_STOP_SIZE,FLOW_RESUME_SIZE,
    MSG_GROUP_HDR_KEY,SHARED_MSG_GROUP,REPLICATE]

class JHelpFormatter(IndentedHelpFormatter):
    """Format usage and description without stripping newlines from usage strings
    """

    def format_usage(self, usage):
        return usage


    def format_description(self, description):
        if description:
            return description + "\n"
        else:
            return ""

def Usage():
    print usage
    sys.exit(-1)

def OptionsAndArguments(argv):
    """ Set global variables for options, return arguments """

    global config


    parser = OptionParser(usage=usage,
                          description=description,
                          formatter=JHelpFormatter())

    group1 = OptionGroup(parser, "General Options")
    group1.add_option("-t", "--timeout", action="store", type="int", default=10, metavar="<secs>", help="Maximum time to wait for broker connection (in seconds)")
    group1.add_option("-r", "--recursive", action="store_true", help="Show bindings in queue or exchange list")
    group1.add_option("-b", "--broker", action="store", type="string", default="localhost:5672", metavar="<address>", help="Address of qpidd broker with syntax: [username/password@] hostname | ip-address [:<port>]")
    group1.add_option("--sasl-mechanism", action="store", type="string", metavar="<mech>", help="SASL mechanism for authentication (e.g. EXTERNAL, ANONYMOUS, PLAIN, CRAM-MD, DIGEST-MD5, GSSAPI). SASL automatically picks the most secure available mechanism - use this option to override.")
    group1.add_option("--ssl-certificate", action="store", type="string", metavar="<cert>", help="Client SSL certificate (PEM Format)")
    group1.add_option("--ssl-key", action="store", type="string", metavar="<key>", help="Client SSL private key (PEM Format)")
    group1.add_option("--ha-admin", action="store_true", help="Allow connection to a HA backup broker.")
    parser.add_option_group(group1)

    group_ls = OptionGroup(parser, "Options for Listing Exchanges and Queues")
    group_ls.add_option("--ignore-default", action="store_true", help="Ignore the default exchange in exchange or queue list")
    parser.add_option_group(group_ls)

    group2 = OptionGroup(parser, "Options for Adding Exchanges and Queues")
    group2.add_option("--alternate-exchange", action="store", type="string", metavar="<aexname>", help="Name of the alternate-exchange for the new queue or exchange. Exchanges route messages to the alternate exchange if they are unable to route them elsewhere. Queues route messages to the alternate exchange if they are rejected by a subscriber or orphaned by queue deletion.")
    group2.add_option("--durable", action="store_true", help="The new queue or exchange is durable.")
    group2.add_option("--replicate", action="store", metavar="<level>", help="Enable automatic replication in a HA cluster. <level> is 'none', 'configuration' or 'all').")
    parser.add_option_group(group2)

    group3 = OptionGroup(parser, "Options for Adding Queues")
    group3.add_option("--cluster-durable", action="store_true", help="The new queue becomes durable if there is only one functioning cluster node")
    group3.add_option("--file-count", action="store", type="int", default=8, metavar="<n>", help="Number of files in queue's persistence journal")
    group3.add_option("--file-size", action="store", type="int", default=24, metavar="<n>", help="File size in pages (64KiB/page)")
    group3.add_option("--max-queue-size", action="store", type="int", metavar="<n>", help="Maximum in-memory queue size as bytes")
    group3.add_option("--max-queue-count", action="store", type="int", metavar="<n>", help="Maximum in-memory queue size as a number of messages")
    group3.add_option("--limit-policy", action="store", choices=["none", "reject", "flow-to-disk", "ring", "ring-strict"], metavar="<policy>", help="Action to take when queue limit is reached")
    group3.add_option("--lvq-key", action="store", metavar="<key>", help="Last Value Queue key")
    group3.add_option("--generate-queue-events", action="store", type="int", metavar="<n>", help="If set to 1, every enqueue will generate an event that can be processed by registered listeners (e.g. for replication). If set to 2, events will be generated for enqueues and dequeues.")
    group3.add_option("--flow-stop-size", action="store", type="int", metavar="<n>",
                      help="Turn on sender flow control when the number of queued bytes exceeds this value.")
    group3.add_option("--flow-resume-size", action="store", type="int", metavar="<n>",
                      help="Turn off sender flow control when the number of queued bytes drops below this value.")
    group3.add_option("--flow-stop-count", action="store", type="int", metavar="<n>",
                      help="Turn on sender flow control when the number of queued messages exceeds this value.")
    group3.add_option("--flow-resume-count", action="store", type="int", metavar="<n>",
                      help="Turn off sender flow control when the number of queued messages drops below this value.")
    group3.add_option("--group-header", action="store", type="string", metavar="<header-name>",
                      help="Enable message groups. Specify name of header that holds group identifier.")
    group3.add_option("--shared-groups", action="store_true",
                      help="Allow message group consumption across multiple consumers.")
    group3.add_option("--argument", dest="extra_arguments", action="append", default=[],
                      metavar="<NAME=VALUE>", help="Specify a key-value pair to add to queue arguments")
    group3.add_option("--start-replica", metavar="<broker-url>", help="Start replication from the same-named queue at <broker-url>")
    # no option for declaring an exclusive queue - which can only be used by the session that creates it.
    parser.add_option_group(group3)

    group4 = OptionGroup(parser, "Options for Adding Exchanges")
    group4.add_option("--sequence", action="store_true", help="Exchange will insert a 'qpid.msg_sequence' field in the message header")
    group4.add_option("--ive", action="store_true", help="Exchange will behave as an 'initial-value-exchange', keeping a reference  to the last message forwarded and enqueuing that message to newly bound queues.")
    parser.add_option_group(group4)

    group5 = OptionGroup(parser, "Options for Deleting Queues")
    group5.add_option("--force", action="store_true", help="Force delete of queue even if it's currently used or it's not empty")
    group5.add_option("--force-if-not-empty", action="store_true", help="Force delete of queue even if it's not empty")
    group5.add_option("--force-if-used", action="store_true", help="Force delete of queue even if it's currently used")
    parser.add_option_group(group5)

    group6 = OptionGroup(parser, "Options for Declaring Bindings")
    group6.add_option("-f", "--file", action="store", type="string", metavar="<file.xq>", help="For XML Exchange bindings - specifies the name of a file containing an XQuery.")
    parser.add_option_group(group6)

    opts, encArgs = parser.parse_args(args=argv)

    try:
        encoding = locale.getpreferredencoding()
        args = [a.decode(encoding) for a in encArgs]
    except:
        args = encArgs

    if opts.recursive:
        config._recursive = True
    if opts.broker:
        config._host = opts.broker
    if opts.timeout is not None:
        config._connTimeout = opts.timeout
        if config._connTimeout == 0:
            config._connTimeout = None
    if opts.ignore_default:
        config._ignoreDefault = True
    if opts.alternate_exchange:
        config._altern_ex = opts.alternate_exchange
    if opts.durable:
        config._durable = True
    if opts.replicate:
        if not opts.replicate in REPLICATE_LEVELS:
            raise Exception("Invalid replication level '%s', should be one of: %s" % (opts.replicate, ", ".join(REPLICATE_LEVELS)))
        config._replicate = opts.replicate
    if opts.ha_admin: config._ha_admin = True
    if opts.cluster_durable:
        config._clusterDurable = True
    if opts.file:
        config._file = opts.file
    if opts.file_count is not None:
        config._fileCount = opts.file_count
    if opts.file_size is not None:
        config._fileSize = opts.file_size
    if opts.max_queue_size is not None:
        config._maxQueueSize = opts.max_queue_size
    if opts.max_queue_count is not None:
        config._maxQueueCount = opts.max_queue_count
    if opts.limit_policy:
           config._limitPolicy = opts.limit_policy
    if opts.sequence:
        config._msgSequence = True
    if opts.lvq_key:
        config._lvq_key = opts.lvq_key
    if opts.ive:
        config._ive = True
    if opts.generate_queue_events:
        config._eventGeneration = opts.generate_queue_events
    if opts.force:
        config._if_empty = False
        config._if_unused = False
    if opts.force_if_not_empty:
        config._if_empty = False
    if opts.force_if_used:
        config._if_unused = False
    if opts.sasl_mechanism:
        config._sasl_mechanism = opts.sasl_mechanism
    if opts.flow_stop_size is not None:
        config._flowStopSize = opts.flow_stop_size
    if opts.flow_resume_size is not None:
        config._flowResumeSize = opts.flow_resume_size
    if opts.flow_stop_count is not None:
        config._flowStopCount = opts.flow_stop_count
    if opts.flow_resume_count is not None:
        config._flowResumeCount = opts.flow_resume_count
    if opts.group_header:
        config._msgGroupHeader = opts.group_header
    if opts.shared_groups:
        config._sharedMsgGroup = True
    if opts.extra_arguments:
        config._extra_arguments = opts.extra_arguments
    if opts.start_replica:
        config._start_replica = opts.start_replica

    if opts.sasl_mechanism:
        conn_options['sasl_mechanisms'] = opts.sasl_mechanism
    if opts.ssl_certificate:
        conn_options['ssl_certfile'] = opts.ssl_certificate
    if opts.ssl_key:
        conn_options['ssl_key'] = opts.ssl_key
    if opts.ha_admin:
        conn_options['client_properties'] = {'qpid.ha-admin' : 1}

    return args


#
# helpers for the arg parsing in bind().  return multiple values; "ok"
# followed by the resultant args

#
# accept -f followed by either
# a filename or "-", for stdin.  pull the bits into a string, to be
# passed to the xml binding.
#
def snarf_xquery_args():
    if not config._file:
        print "Invalid args to bind xml:  need an input file or stdin"
        return [False]
    if config._file == "-":
        res = sys.stdin.read()
    else:
        f = open(config._file)   # let this signal if it can't find it
        res = f.read()
        f.close()
    return [True, res]

#
# look for "any"/"all" and grok the rest of argv into a map
#
def snarf_header_args(args):

    if len(args) < 2:
        print "Invalid args to bind headers:  need 'any'/'all' plus conditions"
        return [False]
    op = args[0]
    if op == "all" or op == "any":
        kv = {}
        for thing in args[1:]:
            k_and_v = thing.split("=")
            kv[k_and_v[0]] = k_and_v[1]
        return [True, op, kv]
    else:
        print "Invalid condition arg to bind headers, need 'any' or 'all', not '" + op + "'"
        return [False]

class BrokerManager:
    def __init__(self):
        self.brokerName = None
        self.conn       = None
        self.broker     = None

    def SetBroker(self, brokerUrl):
        self.url = brokerUrl
        self.conn = Connection.establish(self.url, **conn_options)
        self.broker = BrokerAgent(self.conn)

    def Disconnect(self):
        if self.conn:
            self.conn.close()

    def Overview(self):
        exchanges = self.broker.getAllExchanges()
        queues    = self.broker.getAllQueues()
        print "Total Exchanges: %d" % len(exchanges)
        etype = {}
        for ex in exchanges:
            if ex.type not in etype:
                etype[ex.type] = 1
            else:
                etype[ex.type] = etype[ex.type] + 1
        for typ in etype:
            print "%15s: %d" % (typ, etype[typ])

        print
        print "   Total Queues: %d" % len(queues)
        durable = 0
        for queue in queues:
            if queue.durable:
                durable = durable + 1
        print "        durable: %d" % durable
        print "    non-durable: %d" % (len(queues) - durable)

    def ExchangeList(self, filter):
        exchanges = self.broker.getAllExchanges()
        caption1 = "Type      "
        caption2 = "Exchange Name"
        maxNameLen = len(caption2)
        found = False
        for ex in exchanges:
            if self.match(ex.name, filter):
                if len(ex.name) > maxNameLen:  maxNameLen = len(ex.name)
                found = True
        if not found:
            global config
            config._returnCode = 1
            return

        print "%s%-*s  Attributes" % (caption1, maxNameLen, caption2)
        line = ""
        for i in range(((maxNameLen + len(caption1)) / 5) + 5):
            line += "====="
        print line

        for ex in exchanges:
            if config._ignoreDefault and not ex.name: continue
            if self.match(ex.name, filter):
                print "%-10s%-*s " % (ex.type, maxNameLen, ex.name),
                args = ex.arguments
                if not args: args = {}
                if ex.durable:    print "--durable",
                if REPLICATE in args: print "--replicate=%s" % args[REPLICATE],
                if MSG_SEQUENCE in args and args[MSG_SEQUENCE] == 1: print "--sequence",
                if IVE in args and args[IVE] == 1: print "--ive",
                if ex.altExchange:
                    print "--alternate-exchange=%s" % ex.altExchange,
                print

    def ExchangeListRecurse(self, filter):
        exchanges = self.broker.getAllExchanges()
        bindings  = self.broker.getAllBindings()
        queues    = self.broker.getAllQueues()
        for ex in exchanges:
            if config._ignoreDefault and not ex.name: continue
            if self.match(ex.name, filter):
                print "Exchange '%s' (%s)" % (ex.name, ex.type)
                for bind in bindings:
                    if bind.exchangeRef == ex.name:
                        qname = "<unknown>"
                        queue = self.findById(queues, bind.queueRef)
                        if queue != None:
                            qname = queue.name
                        if bind.arguments:
                            print "    bind [%s] => %s %s" % (bind.bindingKey, qname, bind.arguments)
                        else:
                            print "    bind [%s] => %s" % (bind.bindingKey, qname)


    def QueueList(self, filter):
        queues = self.broker.getAllQueues()
        caption = "Queue Name"
        maxNameLen = len(caption)
        found = False
        for q in queues:
            if self.match(q.name, filter):
                if len(q.name) > maxNameLen:  maxNameLen = len(q.name)
                found = True
        if not found:
            global config
            config._returnCode = 1
            return

        print "%-*s  Attributes" % (maxNameLen, caption)
        line = ""
        for i in range((maxNameLen / 5) + 5):
            line += "====="
        print line

        for q in queues:
            if self.match(q.name, filter):
                print "%-*s " % (maxNameLen, q.name),
                args = q.arguments
                if not args: args = {}
                if q.durable:    print "--durable",
                if REPLICATE in args: print "--replicate=%s" % args[REPLICATE],
                if CLUSTER_DURABLE in args and args[CLUSTER_DURABLE] == 1: print "--cluster-durable",
                if q.autoDelete: print "auto-del",
                if q.exclusive:  print "excl",
                if FILESIZE in args: print "--file-size=%s" % args[FILESIZE],
                if FILECOUNT in args: print "--file-count=%s" % args[FILECOUNT],
                if MAX_QUEUE_SIZE in args: print "--max-queue-size=%s" % args[MAX_QUEUE_SIZE],
                if MAX_QUEUE_COUNT in args: print "--max-queue-count=%s" % args[MAX_QUEUE_COUNT],
                if POLICY_TYPE in args: print "--limit-policy=%s" % args[POLICY_TYPE].replace("_", "-"),
                if LVQ_KEY in args: print "--lvq-key=%s" % args[LVQ_KEY],
                if QUEUE_EVENT_GENERATION in args: print "--generate-queue-events=%s" % args[QUEUE_EVENT_GENERATION],
                if q.altExchange:
                    print "--alternate-exchange=%s" % q.altExchange,
                if FLOW_STOP_SIZE in args: print "--flow-stop-size=%s" % args[FLOW_STOP_SIZE],
                if FLOW_RESUME_SIZE in args: print "--flow-resume-size=%s" % args[FLOW_RESUME_SIZE],
                if FLOW_STOP_COUNT in args: print "--flow-stop-count=%s" % args[FLOW_STOP_COUNT],
                if FLOW_RESUME_COUNT in args: print "--flow-resume-count=%s" % args[FLOW_RESUME_COUNT],
                if MSG_GROUP_HDR_KEY in args: print "--group-header=%s" % args[MSG_GROUP_HDR_KEY],
                if SHARED_MSG_GROUP in args and args[SHARED_MSG_GROUP] == 1: print "--shared-groups",
                print " ".join(["--argument %s=%s" % (k, v) for k,v in args.iteritems() if not k in SPECIAL_ARGS])

    def QueueListRecurse(self, filter):
        exchanges = self.broker.getAllExchanges()
        bindings  = self.broker.getAllBindings()
        queues    = self.broker.getAllQueues()
        for queue in queues:
            if self.match(queue.name, filter):
                print "Queue '%s'" % queue.name
                for bind in bindings:
                    if bind.queueRef == queue.name:
                        ename = "<unknown>"
                        ex    = self.findById(exchanges, bind.exchangeRef)
                        if ex != None:
                            ename = ex.name
                            if ename == "":
                                if config._ignoreDefault: continue
                                ename = "''"
                        if bind.arguments:
                            print "    bind [%s] => %s %s" % (bind.bindingKey, ename, bind.arguments)
                        else:
                            print "    bind [%s] => %s" % (bind.bindingKey, ename)

    def AddExchange(self, args):
        if len(args) < 2:
            Usage()
        etype = args[0]
        ename = args[1]
        declArgs = {}
        for a in config._extra_arguments:
            r = a.split("=", 1)
            if len(r) == 2: value = r[1]
            else: value = None
            declArgs[r[0]] = value

        if config._msgSequence:
            declArgs[MSG_SEQUENCE] = 1
        if config._ive:
            declArgs[IVE] = 1
        if config._altern_ex:
            declArgs['alternate-exchange'] = config._altern_ex
        if config._durable:
            declArgs['durable'] = 1
        if config._replicate:
            declArgs[REPLICATE] = config._replicate
        self.broker.addExchange(etype, ename, declArgs)


    def DelExchange(self, args):
        if len(args) < 1:
            Usage()
        ename = args[0]
        self.broker.delExchange(ename)


    def AddQueue(self, args):
        if len(args) < 1:
            Usage()
        qname    = args[0]
        declArgs = {}
        for a in config._extra_arguments:
            r = a.split("=", 1)
            if len(r) == 2: value = r[1]
            else: value = None
            declArgs[r[0]] = value

        if config._durable:
            declArgs[FILECOUNT] = config._fileCount
            declArgs[FILESIZE]  = config._fileSize

        if config._maxQueueSize is not None:
            declArgs[MAX_QUEUE_SIZE]  = config._maxQueueSize
        if config._maxQueueCount is not None:
            declArgs[MAX_QUEUE_COUNT]  = config._maxQueueCount
        if config._limitPolicy:
            if config._limitPolicy == "none":
                pass
            elif config._limitPolicy == "reject":
                declArgs[POLICY_TYPE] = "reject"
            elif config._limitPolicy == "flow-to-disk":
                declArgs[POLICY_TYPE] = "flow_to_disk"
            elif config._limitPolicy == "ring":
                declArgs[POLICY_TYPE] = "ring"
            elif config._limitPolicy == "ring-strict":
                declArgs[POLICY_TYPE] = "ring_strict"

        if config._clusterDurable:
            declArgs[CLUSTER_DURABLE] = 1
        if config._lvq_key:
            declArgs[LVQ_KEY] = config._lvq_key
        if config._eventGeneration:
            declArgs[QUEUE_EVENT_GENERATION]  = config._eventGeneration

        if config._flowStopSize is not None:
            declArgs[FLOW_STOP_SIZE]  = config._flowStopSize
        if config._flowResumeSize is not None:
            declArgs[FLOW_RESUME_SIZE]  = config._flowResumeSize
        if config._flowStopCount is not None:
            declArgs[FLOW_STOP_COUNT]  = config._flowStopCount
        if config._flowResumeCount is not None:
            declArgs[FLOW_RESUME_COUNT]  = config._flowResumeCount

        if config._msgGroupHeader:
            declArgs[MSG_GROUP_HDR_KEY] = config._msgGroupHeader
        if config._sharedMsgGroup:
            declArgs[SHARED_MSG_GROUP] = 1

        if config._altern_ex:
            declArgs['alternate-exchange'] = config._altern_ex
        if config._durable:
            declArgs['durable'] = 1
        if config._replicate:
            declArgs[REPLICATE] = config._replicate
        self.broker.addQueue(qname, declArgs)
        if config._start_replica:      # Start replication
            self.broker._method("replicate", {"broker":config._start_replica, "queue":qname}, "org.apache.qpid.ha:habroker:ha-broker")

    def DelQueue(self, args):
        if len(args) < 1:
            Usage()
        qname = args[0]
        self.broker.delQueue(qname)


    def Bind(self, args):
        if len(args) < 2:
            Usage()
        ename = args[0]
        qname = args[1]
        key   = ""
        if len(args) > 2:
            key = args[2]

        # query the exchange to determine its type.
        res = self.broker.getExchange(ename)

        # type of the xchg determines the processing of the rest of
        # argv.  if it's an xml xchg, we want to find a file
        # containing an x-query, and pass that.  if it's a headers
        # exchange, we need to pass either "any" or all, followed by a
        # map containing key/value pairs.  if neither of those, extra
        # args are ignored.
        ok = True
        _args = {}
        if res.type == "xml":
            # this checks/imports the -f arg
            [ok, xquery] = snarf_xquery_args()
            _args = { "xquery" : xquery }
        else:
            if res.type == "headers":
                [ok, op, kv] = snarf_header_args(args[3:])
                _args = kv
                _args["x-match"] = op

        if not ok:
            sys.exit(1)

        self.broker.bind(ename, qname, key, _args)

    def Unbind(self, args):
        if len(args) < 2:
            Usage()
        ename = args[0]
        qname = args[1]
        key   = ""
        if len(args) > 2:
            key = args[2]
        self.broker.unbind(ename, qname, key)

    def ReloadAcl(self):
        try:
            self.broker.reloadAclFile()
        except Exception, e:
            if str(e).find('No object found') != -1:
                print "Failed: ACL Module Not Loaded in Broker"
            else:
                raise

    def findById(self, items, id):
        for item in items:
            if item.name == id:
                return item
        return None

    def match(self, name, filter):
        if filter == "":
            return True
        if name.find(filter) == -1:
            return False
        return True

def YN(bool):
    if bool:
        return 'Y'
    return 'N'


def main(argv=None):
    args = OptionsAndArguments(argv)
    bm   = BrokerManager()

    try:
        bm.SetBroker(config._host)
        if len(args) == 0:
            bm.Overview()
        else:
            cmd = args[0]
            modifier = ""
            if len(args) > 1:
                modifier = args[1]
            if cmd == "exchanges":
                if config._recursive:
                    bm.ExchangeListRecurse(modifier)
                else:
                    bm.ExchangeList(modifier)
            elif cmd == "queues":
                if config._recursive:
                    bm.QueueListRecurse(modifier)
                else:
                    bm.QueueList(modifier)
            elif cmd == "add":
                if modifier == "exchange":
                    bm.AddExchange(args[2:])
                elif modifier == "queue":
                    bm.AddQueue(args[2:])
                else:
                    Usage()
            elif cmd == "del":
                if modifier == "exchange":
                    bm.DelExchange(args[2:])
                elif modifier == "queue":
                    bm.DelQueue(args[2:])
                else:
                    Usage()
            elif cmd == "bind":
                bm.Bind(args[1:])
            elif cmd == "unbind":
                bm.Unbind(args[1:])
            elif cmd == "reload-acl":
                bm.ReloadAcl()
            else:
                Usage()
    except KeyboardInterrupt:
        print
    except IOError, e:
        print e
        bm.Disconnect()
        return 1
    except SystemExit, e:
        bm.Disconnect()
        return 1
    except Exception,e:
        if e.__class__.__name__ != "Timeout":
            # ignore Timeout exception, handle in the loop below
            print "Failed: %s: %s" % (e.__class__.__name__, e)
            bm.Disconnect()
            return 1

    while True:
        # some commands take longer than the default amqp timeout to complete,
        # so attempt to disconnect until successful, ignoring Timeouts
        try:
            bm.Disconnect()
            break
        except Exception, e:
            if e.__class__.__name__ != "Timeout":
                print "Failed: %s: %s" % (e.__class__.__name__, e)
                return 1
    return config._returnCode


if __name__ == "__main__":
        sys.exit(main())