| #!/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 os |
| from optparse import OptionParser, OptionGroup, IndentedHelpFormatter |
| import sys |
| import locale |
| from qmf.console import Session |
| |
| 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]""" |
| |
| description = """ |
| Examples: |
| |
| $ qpid-config add queue q |
| $ qpid-config add exchange direct d -a localhost:5672 |
| $ qpid-config exchanges -a 10.1.1.7:10000 |
| $ qpid-config queues -a 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 |
| |
| Queue Ordering Policies |
| |
| fifo (default) - First in, first out |
| lvq - Last Value Queue ordering, allows queue browsing |
| lvq-no-browse - Last Value Queue ordering, browsing clients may lose data""" |
| |
| |
| class Config: |
| def __init__(self): |
| self._recursive = False |
| self._host = "localhost" |
| self._connTimeout = 10 |
| self._ignoreDefault = False |
| self._altern_ex = None |
| self._passive = False |
| self._durable = False |
| 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._order = None |
| self._msgSequence = False |
| self._ive = False |
| self._eventGeneration = None |
| self._file = None |
| self._sasl_mechanism = None |
| self._flowStopCount = None |
| self._flowResumeCount = None |
| self._flowStopSize = None |
| self._flowResumeSize = None |
| self._msgGroupHeader = None |
| self._sharedMsgGroup = False |
| self._extra_arguments = [] |
| self._returnCode = 0 |
| |
| config = Config() |
| |
| 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 = "qpid.last_value_queue" |
| LVQNB = "qpid.last_value_queue_no_browse" |
| 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" |
| #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,LVQNB,MSG_SEQUENCE,IVE,QUEUE_EVENT_GENERATION,FLOW_STOP_COUNT,FLOW_STOP_SIZE,FLOW_RESUME_SIZE, |
| MSG_GROUP_HDR_KEY,SHARED_MSG_GROUP] |
| |
| 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 |
| 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("-b", "--bindings", action="store_true", help="Show bindings in queue or exchange list") |
| group1.add_option("-a", "--broker-addr", 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.") |
| 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("--passive", "--dry-run", action="store_true", help="Do not actually add the exchange or queue, ensure that all parameters and permissions are correct and would allow it to be created.") |
| group2.add_option("--durable", action="store_true", help="The new queue or exchange is durable.") |
| 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("--order", action="store", choices=["fifo", "lvq", "lvq-no-browse"], metavar="<ordering>", help="Queue ordering policy") |
| 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") |
| # 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.bindings: |
| config._recursive = True |
| if opts.broker_addr: |
| config._host = opts.broker_addr |
| if opts.timeout: |
| 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.passive: |
| config._passive = True |
| if opts.durable: |
| config._durable = True |
| if opts.cluster_durable: |
| config._clusterDurable = True |
| if opts.file: |
| config._file = opts.file |
| if opts.file_count: |
| config._fileCount = opts.file_count |
| if opts.file_size: |
| config._fileSize = opts.file_size |
| if opts.max_queue_size != None: |
| config._maxQueueSize = opts.max_queue_size |
| if opts.max_queue_count: |
| config._maxQueueCount = opts.max_queue_count |
| if opts.limit_policy: |
| config._limitPolicy = opts.limit_policy |
| if opts.order: |
| config._order = opts.order |
| if opts.sequence: |
| config._msgSequence = True |
| 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: |
| config._flowStopSize = opts.flow_stop_size |
| if opts.flow_resume_size: |
| config._flowResumeSize = opts.flow_resume_size |
| if opts.flow_stop_count: |
| config._flowStopCount = opts.flow_stop_count |
| if opts.flow_resume_count: |
| 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 |
| 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.qmf = None |
| self.broker = None |
| self.mechanism = None |
| |
| def SetBroker(self, brokerUrl, mechanism): |
| self.url = brokerUrl |
| self.qmf = Session() |
| self.broker = self.qmf.addBroker(brokerUrl, config._connTimeout, mechanism) |
| agents = self.qmf.getAgents() |
| for a in agents: |
| if a.getAgentBank() == '0': |
| self.brokerAgent = a |
| |
| def Disconnect(self): |
| if self.broker: |
| self.qmf.delBroker(self.broker) |
| |
| def Overview(self): |
| exchanges = self.qmf.getObjects(_class="exchange", _agent=self.brokerAgent) |
| queues = self.qmf.getObjects(_class="queue", _agent=self.brokerAgent) |
| 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.qmf.getObjects(_class="exchange", _agent=self.brokerAgent) |
| 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 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_.name, |
| print |
| |
| def ExchangeListRecurse(self, filter): |
| exchanges = self.qmf.getObjects(_class="exchange", _agent=self.brokerAgent) |
| bindings = self.qmf.getObjects(_class="binding", _agent=self.brokerAgent) |
| queues = self.qmf.getObjects(_class="queue", _agent=self.brokerAgent) |
| 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.getObjectId(): |
| qname = "<unknown>" |
| queue = self.findById(queues, bind.queueRef) |
| if queue != None: |
| qname = queue.name |
| print " bind [%s] => %s" % (bind.bindingKey, qname) |
| |
| |
| def QueueList(self, filter): |
| queues = self.qmf.getObjects(_class="queue", _agent=self.brokerAgent) |
| 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 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 in args and args[LVQ] == 1: print "--order lvq", |
| if LVQNB in args and args[LVQNB] == 1: print "--order lvq-no-browse", |
| if QUEUE_EVENT_GENERATION in args: print "--generate-queue-events=%s" % args[QUEUE_EVENT_GENERATION], |
| if q.altExchange: |
| print "--alternate-exchange=%s" % q._altExchange_.name, |
| 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.qmf.getObjects(_class="exchange", _agent=self.brokerAgent) |
| bindings = self.qmf.getObjects(_class="binding", _agent=self.brokerAgent) |
| queues = self.qmf.getObjects(_class="queue", _agent=self.brokerAgent) |
| for queue in queues: |
| if self.match(queue.name, filter): |
| print "Queue '%s'" % queue.name |
| for bind in bindings: |
| if bind.queueRef == queue.getObjectId(): |
| ename = "<unknown>" |
| ex = self.findById(exchanges, bind.exchangeRef) |
| if ex != None: |
| ename = ex.name |
| if ename == "": |
| if config._ignoreDefault: continue |
| ename = "''" |
| print " bind [%s] => %s" % (bind.bindingKey, ename) |
| |
| def AddExchange(self, args): |
| if len(args) < 2: |
| Usage() |
| etype = args[0] |
| ename = args[1] |
| declArgs = {} |
| if config._msgSequence: |
| declArgs[MSG_SEQUENCE] = 1 |
| if config._ive: |
| declArgs[IVE] = 1 |
| if config._altern_ex != None: |
| self.broker.getAmqpSession().exchange_declare(exchange=ename, type=etype, alternate_exchange=config._altern_ex, passive=config._passive, durable=config._durable, arguments=declArgs) |
| else: |
| self.broker.getAmqpSession().exchange_declare(exchange=ename, type=etype, passive=config._passive, durable=config._durable, arguments=declArgs) |
| |
| def DelExchange(self, args): |
| if len(args) < 1: |
| Usage() |
| ename = args[0] |
| self.broker.getAmqpSession().exchange_delete(exchange=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 != None: |
| declArgs[MAX_QUEUE_SIZE] = config._maxQueueSize |
| if config._maxQueueCount: |
| 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._order: |
| if config._order == "fifo": |
| pass |
| elif config._order == "lvq": |
| declArgs[LVQ] = 1 |
| elif config._order == "lvq-no-browse": |
| declArgs[LVQNB] = 1 |
| if config._eventGeneration: |
| declArgs[QUEUE_EVENT_GENERATION] = config._eventGeneration |
| |
| if config._flowStopSize: |
| declArgs[FLOW_STOP_SIZE] = config._flowStopSize |
| if config._flowResumeSize: |
| declArgs[FLOW_RESUME_SIZE] = config._flowResumeSize |
| if config._flowStopCount: |
| declArgs[FLOW_STOP_COUNT] = config._flowStopCount |
| if config._flowResumeCount: |
| 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 != None: |
| self.broker.getAmqpSession().queue_declare(queue=qname, alternate_exchange=config._altern_ex, passive=config._passive, durable=config._durable, arguments=declArgs) |
| else: |
| self.broker.getAmqpSession().queue_declare(queue=qname, passive=config._passive, durable=config._durable, arguments=declArgs) |
| |
| |
| def DelQueue(self, args): |
| if len(args) < 1: |
| Usage() |
| qname = args[0] |
| self.broker.getAmqpSession().queue_delete(queue=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.getAmqpSession().exchange_query(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 = None |
| 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.getAmqpSession().exchange_bind(queue=qname, |
| exchange=ename, |
| binding_key=key, |
| arguments=_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.getAmqpSession().exchange_unbind(queue=qname, exchange=ename, binding_key=key) |
| |
| def findById(self, items, id): |
| for item in items: |
| if item.getObjectId() == 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, config._sasl_mechanism) |
| 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:]) |
| 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()) |
| |
