diff options
Diffstat (limited to 'qpid/cpp/management/python/bin/qpid-config')
| -rwxr-xr-x | qpid/cpp/management/python/bin/qpid-config | 878 |
1 files changed, 878 insertions, 0 deletions
diff --git a/qpid/cpp/management/python/bin/qpid-config b/qpid/cpp/management/python/bin/qpid-config new file mode 100755 index 0000000000..3d4bb6036a --- /dev/null +++ b/qpid/cpp/management/python/bin/qpid-config @@ -0,0 +1,878 @@ +#!/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, ConnectionError +from qpidtoollibs import BrokerAgent +from qpidtoollibs import Display, Header + +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 + qpid-config [OPTIONS] add <type> <name> [--argument <property-name>=<property-value>] + qpid-config [OPTIONS] del <type> <name> + qpid-config [OPTIONS] list <type> [--show-property <property-name>] + qpid-config [OPTIONS] log [<logstring>] + qpid-config [OPTIONS] shutdown""" + +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 + ring - Replace oldest unacquired message with new + +Replication levels: + + none - no replication + configuration - replicate queue and exchange existence and bindings, but not messages. + all - replicate configuration and messages + +Log <logstring> value: + + Comma separated <module>:<level> pairs, e.g. 'info+,debug+:Broker,trace+:Queue' +""" + +REPLICATE_LEVELS= ["none", "configuration", "all"] +DEFAULT_PROPERTIES = {"exchange":["name", "type", "durable"], "queue":["name", "durable", "autoDelete"]} + +def get_value(r): + if len(r) == 2: + try: + value = int(r[1]) + except: + value = r[1] + else: value = None + return value + +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._if_empty = True + self._if_unused = True + self._fileCount = None + self._fileSize = None + self._efp_partition_num = None + self._efp_pool_file_size = None + 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 + self._list_properties = [] + + def getOptions(self): + options = {} + for a in self._extra_arguments: + r = a.split("=", 1) + options[r[0]] = get_value(r) + return options + + +config = Config() +conn_options = {} + +FILECOUNT = "qpid.file_count" +FILESIZE = "qpid.file_size" +EFP_PARTITION_NUM = "qpid.efp_partition_num" +EFP_POOL_FILE_SIZE = "qpid.efp_pool_file_size" +MAX_QUEUE_SIZE = "qpid.max_size" +MAX_QUEUE_COUNT = "qpid.max_count" +POLICY_TYPE = "qpid.policy_type" +LVQ_KEY = "qpid.last_value_queue_key" +MSG_SEQUENCE = "qpid.msg_sequence" +IVE = "qpid.ive" +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,EFP_PARTITION_NUM,EFP_POOL_FILE_SIZE, + MAX_QUEUE_SIZE,MAX_QUEUE_COUNT,POLICY_TYPE, + LVQ_KEY,MSG_SEQUENCE,IVE, + FLOW_STOP_COUNT,FLOW_RESUME_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", metavar="<address>", help="Address of qpidd broker with syntax: [username/password@] hostname | ip-address [:<port>]") + group1.add_option("-a", "--broker-addr", action="store", type="string", metavar="<address>") + group1.add_option("--sasl-mechanism", action="store", type="string", metavar="<mech>", help="SASL mechanism for authentication (e.g. EXTERNAL, ANONYMOUS, PLAIN, CRAM-MD5, DIGEST-MD5, GSSAPI). SASL automatically picks the most secure available mechanism - use this option to override.") + group1.add_option("--sasl-service-name", action="store", type="string", help="SASL service name to use") + 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("--file-count", action="store", type="int", metavar="<n>", help="[legacystore] Number of files in queue's persistence journal") + group3.add_option("--file-size", action="store", type="int", metavar="<n>", help="[legactystore] File size in pages (64KiB/page)") + group3.add_option("--efp-partition-num", action="store", type="int", metavar="<n>", help="[linearstore] EFP partition number") + group3.add_option("--efp-pool-file-size", action="store", type="int", metavar="<n>", help="[linearstore] EFP file size (KiB)") + 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", "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("--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) + + group_7 = OptionGroup(parser, "Formatting options for 'list' action") + group_7.add_option("--show-property", dest="list_properties", action="append", default=[], + metavar="<property-name>", help="Specify a property of an object to be included in output") + parser.add_option_group(group_7) + + 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.broker_addr: + config._host = opts.broker_addr + if config._host is None: config._host="localhost:5672" + 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.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.efp_partition_num is not None: + config._efp_partition_num = opts.efp_partition_num + if opts.efp_pool_file_size is not None: + config._efp_pool_file_size = opts.efp_pool_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.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.list_properties: + config._list_properties = opts.list_properties + + if opts.sasl_mechanism: + conn_options['sasl_mechanisms'] = opts.sasl_mechanism + if opts.sasl_service_name: + conn_options['sasl_service'] = opts.sasl_service_name + if opts.ssl_certificate: + conn_options['ssl_certfile'] = opts.ssl_certificate + if opts.ssl_key: + if not opts.ssl_certificate: + parser.error("missing '--ssl-certificate' (required by '--ssl-key')") + conn_options['ssl_keyfile'] = 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, ignore=True): + if self.conn: + try: + self.conn.close() + except Exception, e: + if ignore: + # suppress close errors to avoid + # tracebacks when a previous + # exception will be printed to stdout + pass + else: + # raise last exception so complete + # trackback is preserved + raise + + 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]: print "--sequence", + if IVE in args and args[IVE]: 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 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 EFP_PARTITION_NUM in args: print "--efp-partition-num=%s" % args[EFP_PARTITION_NUM], + if EFP_POOL_FILE_SIZE in args: print "--efp-pool-file-size=%s" % args[EFP_POOL_FILE_SIZE], + 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 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) + declArgs[r[0]] = get_value(r) + + 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) + declArgs[r[0]] = get_value(r) + + if config._durable: + # allow the default fileCount and fileSize specified + # in qpid config file to take prededence + if config._fileCount: + declArgs[FILECOUNT] = config._fileCount + if config._fileSize: + declArgs[FILESIZE] = config._fileSize + if config._efp_partition_num: + declArgs[EFP_PARTITION_NUM] = config._efp_partition_num + if config._efp_pool_file_size: + declArgs[EFP_POOL_FILE_SIZE] = config._efp_pool_file_size + + 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 == "ring": + declArgs[POLICY_TYPE] = "ring" + + if config._lvq_key: + declArgs[LVQ_KEY] = config._lvq_key + + 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, if_empty=config._if_empty, if_unused=config._if_unused) + + + + 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 not res: + pass + elif 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 _clean_ref(o): + if isinstance(o, dict) and "_object_name" in o: + fqn = o["_object_name"] + parts = fqn.split(":",2) + return parts[len(parts)-1] + else: + return o + +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:]) + elif len(args) > 2: + bm.broker.create(modifier, args[2], config.getOptions()) + else: + Usage() + elif cmd == "del": + if modifier == "exchange": + bm.DelExchange(args[2:]) + elif modifier == "queue": + bm.DelQueue(args[2:]) + elif len(args) > 2: + bm.broker.delete(modifier, args[2], {}) + else: + Usage() + elif cmd == "bind": + bm.Bind(args[1:]) + elif cmd == "unbind": + bm.Unbind(args[1:]) + elif cmd == "reload-acl": + bm.ReloadAcl() + elif cmd == "list" and len(args) > 1: + # fetch objects + objects = bm.broker.list(modifier) + + # collect available attributes + attributes = [] + for o in objects: + for k in o.keys(): + if k == "name" and k not in attributes: + attributes.insert(0, k) + elif k not in attributes: + attributes.append(k) + + # determine which attributes to display + desired = [] + if len(config._list_properties): + for p in config._list_properties: + if p not in attributes: print "Warning: No such property '%s' for type '%s'" % (p, modifier) + else: desired.append(p) + elif modifier in DEFAULT_PROPERTIES: + desired = DEFAULT_PROPERTIES[modifier] + else: + desired = attributes[:6] + + # display + display = Display(prefix=" ") + headers = [Header(a) for a in desired] + rows = [tuple([_clean_ref(o.get(a, "n/a")) for a in desired]) for o in objects] + display.formattedTable("Objects of type '%s'" % modifier, headers, rows) + elif cmd == "log" and len (args) == 1: + print "Log level:", bm.broker.getLogLevel()["level"] + elif cmd == "log" and len (args) == 2: + bm.broker.setLogLevel(args[1]) + elif cmd == "shutdown": + try: + bm.broker._method("shutdown", {}) + except ConnectionError: + pass # Normal, the broker has been shut down! + bm.conn = None # Don't try to close again + 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(ignore=False) + 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()) + |