diff options
Diffstat (limited to 'python/commands/qpid-route')
| -rwxr-xr-x | python/commands/qpid-route | 593 |
1 files changed, 373 insertions, 220 deletions
diff --git a/python/commands/qpid-route b/python/commands/qpid-route index 3cd9109a6a..9965047000 100755 --- a/python/commands/qpid-route +++ b/python/commands/qpid-route @@ -22,280 +22,379 @@ import getopt import sys import socket -import qpid import os -from qpid.management import managementClient -from qpid.managementdata import Broker -from qpid.peer import Closed -from qpid.connection import Connection, ConnectionFailed -from qpid.util import connect - -def Usage (): - print "Usage: qpid-route [OPTIONS] link add <dest-broker> <src-broker>" - print " qpid-route [OPTIONS] link del <dest-broker> <src-broker>" - print " qpid-route [OPTIONS] link list [<dest-broker>]" +import locale +from qmf.console import Session, BrokerURL + +def Usage(): + print "Usage: qpid-route [OPTIONS] dynamic add <dest-broker> <src-broker> <exchange> [tag] [exclude-list]" + print " qpid-route [OPTIONS] dynamic del <dest-broker> <src-broker> <exchange>" print print " qpid-route [OPTIONS] route add <dest-broker> <src-broker> <exchange> <routing-key> [tag] [exclude-list]" print " qpid-route [OPTIONS] route del <dest-broker> <src-broker> <exchange> <routing-key>" + print " qpid-route [OPTIONS] queue add <dest-broker> <src-broker> <exchange> <queue>" + print " qpid-route [OPTIONS] queue del <dest-broker> <src-broker> <exchange> <queue>" print " qpid-route [OPTIONS] route list [<dest-broker>]" print " qpid-route [OPTIONS] route flush [<dest-broker>]" + print " qpid-route [OPTIONS] route map [<broker>]" + print + print " qpid-route [OPTIONS] link add <dest-broker> <src-broker>" + print " qpid-route [OPTIONS] link del <dest-broker> <src-broker>" + print " qpid-route [OPTIONS] link list [<dest-broker>]" print print "Options:" + print " --timeout seconds (10) Maximum time to wait for broker connection" print " -v [ --verbose ] Verbose output" print " -q [ --quiet ] Quiet output, don't print duplicate warnings" print " -d [ --durable ] Added configuration shall be durable" print " -e [ --del-empty-link ] Delete link after deleting last route on the link" + print " -s [ --src-local ] Make connection to source broker (push route)" + print " --ack N Acknowledge transfers over the bridge in batches of N" + print " -t <transport> [ --transport <transport>]" + print " Specify transport to use for links, defaults to tcp" print print " dest-broker and src-broker are in the form: [username/password@] hostname | ip-address [:<port>]" print " ex: localhost, 10.1.1.7:10000, broker-host:10000, guest/guest@localhost" print - sys.exit (1) + sys.exit(1) -_verbose = False -_quiet = False -_durable = False -_dellink = False +_verbose = False +_quiet = False +_durable = False +_dellink = False +_srclocal = False +_transport = "tcp" +_ack = 0 +_connTimeout = 10 class RouteManager: - def __init__ (self, destBroker): - self.dest = Broker (destBroker) - self.src = None + def __init__(self, localBroker): + self.local = BrokerURL(localBroker) + self.remote = None + self.qmf = Session() + self.broker = self.qmf.addBroker(localBroker, _connTimeout) - def ConnectToBroker (self): - broker = self.dest - if _verbose: - print "Connecting to broker: %s:%d" % (broker.host, broker.port) - try: - self.sessionId = "%s.%d" % (os.uname()[1], os.getpid()) - self.conn = Connection (connect (broker.host, broker.port), \ - username=broker.username, password=broker.password) - self.conn.start () - self.session = self.conn.session(self.sessionId) - self.mclient = managementClient (self.conn.spec) - self.mch = self.mclient.addChannel (self.session) - self.mclient.syncWaitForStable (self.mch) - except socket.error, e: - print "Socket Error %s - %s" % (e[0], e[1]) - sys.exit (1) - except Closed, e: - print "Connect Failed %d - %s" % (e[0], e[1]) - sys.exit (1) - except ConnectionFailed, e: - print "Connect Failed %d - %s" % (e[0], e[1]) - sys.exit(1) - - def Disconnect (self): - self.mclient.removeChannel (self.mch) - self.session.close(timeout=10) - self.conn.close(timeout=10) - - def getLink (self): - links = self.mclient.syncGetObjects (self.mch, "link") + def disconnect(self): + self.qmf.delBroker(self.broker) + + def getLink(self): + links = self.qmf.getObjects(_class="link") for link in links: - if "%s:%d" % (link.host, link.port) == self.src.name (): + if self.remote.match(link.host, link.port): return link return None - def AddLink (self, srcBroker): - self.src = Broker (srcBroker) - mc = self.mclient - - if self.dest.name() == self.src.name(): - print "Linking broker to itself is not permitted" - sys.exit(1) + def addLink(self, remoteBroker): + self.remote = BrokerURL(remoteBroker) + if self.local.match(self.remote.host, self.remote.port): + raise Exception("Linking broker to itself is not permitted") - brokers = mc.syncGetObjects (self.mch, "broker") + brokers = self.qmf.getObjects(_class="broker") broker = brokers[0] link = self.getLink() - if link != None: - print "Link already exists" - sys.exit(1) - - connectArgs = {} - connectArgs["host"] = self.src.host - connectArgs["port"] = self.src.port - connectArgs["useSsl"] = False - connectArgs["durable"] = _durable - if self.src.username == "anonymous": - connectArgs["authMechanism"] = "ANONYMOUS" - else: - connectArgs["authMechanism"] = "PLAIN" - connectArgs["username"] = self.src.username - connectArgs["password"] = self.src.password - res = mc.syncCallMethod (self.mch, broker.id, broker.classKey, "connect", connectArgs) - if _verbose: - print "Connect method returned:", res.status, res.statusText - link = self.getLink () - - def DelLink (self, srcBroker): - self.src = Broker (srcBroker) - mc = self.mclient + if link == None: + if not self.remote.authName or self.remote.authName == "anonymous": + mech = "ANONYMOUS" + else: + mech = "PLAIN" + res = broker.connect(self.remote.host, self.remote.port, _durable, + mech, self.remote.authName or "", self.remote.authPass or "", + _transport) + if _verbose: + print "Connect method returned:", res.status, res.text - brokers = mc.syncGetObjects (self.mch, "broker") + def delLink(self, remoteBroker): + self.remote = BrokerURL(remoteBroker) + brokers = self.qmf.getObjects(_class="broker") broker = brokers[0] link = self.getLink() if link == None: - print "Link not found" - sys.exit(1) + raise Exception("Link not found") - res = mc.syncCallMethod (self.mch, link.id, link.classKey, "close") + res = link.close() if _verbose: - print "Close method returned:", res.status, res.statusText + print "Close method returned:", res.status, res.text - def ListLinks (self): - mc = self.mclient - links = mc.syncGetObjects (self.mch, "link") + def listLinks(self): + links = self.qmf.getObjects(_class="link") if len(links) == 0: print "No Links Found" else: print - print "Host Port Durable State Last Error" - print "===================================================================" + print "Host Port Transport Durable State Last Error" + print "=============================================================================" + for link in links: + print "%-16s%-8d%-13s%c %-18s%s" % \ + (link.host, link.port, link.transport, YN(link.durable), link.state, link.lastError) + + def mapRoutes(self): + qmf = self.qmf + print + print "Finding Linked Brokers:" + + brokerList = {} + brokerList[self.local.name()] = self.broker + print " %s... Ok" % self.local + + added = True + while added: + added = False + links = qmf.getObjects(_class="link") for link in links: - print "%-16s%-8d %c %-18s%s" % (link.host, link.port, YN(link.durable), link.state, link.lastError) + url = BrokerURL("%s:%d" % (link.host, link.port)) + if url.name() not in brokerList: + print " %s..." % url.name(), + try: + b = qmf.addBroker("%s:%d" % (link.host, link.port), _connTimeout) + brokerList[url.name()] = b + added = True + print "Ok" + except Exception, e: + print e + + print + print "Dynamic Routes:" + bridges = qmf.getObjects(_class="bridge", dynamic=True) + fedExchanges = [] + for bridge in bridges: + if bridge.src not in fedExchanges: + fedExchanges.append(bridge.src) + if len(fedExchanges) == 0: + print " none found" + print + + for ex in fedExchanges: + print " Exchange %s:" % ex + pairs = [] + for bridge in bridges: + if bridge.src == ex: + link = bridge._linkRef_ + fromUrl = "%s:%s" % (link.host, link.port) + toUrl = bridge.getBroker().getUrl() + found = False + for pair in pairs: + if pair.matches(fromUrl, toUrl): + found = True + if not found: + pairs.append(RoutePair(fromUrl, toUrl)) + for pair in pairs: + print " %s" % pair + print - def AddRoute (self, srcBroker, exchange, routingKey, tag, excludes): - self.src = Broker (srcBroker) - mc = self.mclient + print "Static Routes:" + bridges = qmf.getObjects(_class="bridge", dynamic=False) + if len(bridges) == 0: + print " none found" + print - if self.dest.name() == self.src.name(): - print "Linking broker to itself is not permitted" - sys.exit(1) + for bridge in bridges: + link = bridge._linkRef_ + fromUrl = "%s:%s" % (link.host, link.port) + toUrl = bridge.getBroker().getUrl() + leftType = "ex" + rightType = "ex" + if bridge.srcIsLocal: + arrow = "=>" + left = bridge.src + right = bridge.dest + if bridge.srcIsQueue: + leftType = "queue" + else: + arrow = "<=" + left = bridge.dest + right = bridge.src + if bridge.srcIsQueue: + rightType = "queue" + + if bridge.srcIsQueue: + print " %s(%s=%s) %s %s(%s=%s)" % \ + (toUrl, leftType, left, arrow, fromUrl, rightType, right) + else: + print " %s(%s=%s) %s %s(%s=%s) key=%s" % \ + (toUrl, leftType, left, arrow, fromUrl, rightType, right, bridge.key) + print + + for broker in brokerList: + if broker != self.local.name(): + qmf.delBroker(brokerList[broker]) - brokers = mc.syncGetObjects (self.mch, "broker") - broker = brokers[0] - link = self.getLink () + def addRoute(self, remoteBroker, exchange, routingKey, tag, excludes, dynamic=False): + if dynamic and _srclocal: + raise Exception("--src-local is not permitted on dynamic routes") + + self.addLink(remoteBroker) + link = self.getLink() if link == None: - if _verbose: - print "Inter-broker link not found, creating..." - - connectArgs = {} - connectArgs["host"] = self.src.host - connectArgs["port"] = self.src.port - connectArgs["useSsl"] = False - connectArgs["durable"] = _durable - if self.src.username == "anonymous": - connectArgs["authMechanism"] = "ANONYMOUS" - else: - connectArgs["authMechanism"] = "PLAIN" - connectArgs["username"] = self.src.username - connectArgs["password"] = self.src.password - res = mc.syncCallMethod (self.mch, broker.id, broker.classKey, "connect", connectArgs) - if _verbose: - print "Connect method returned:", res.status, res.statusText - link = self.getLink () + raise Exception("Link failed to create") + bridges = self.qmf.getObjects(_class="bridge") + for bridge in bridges: + if bridge.linkRef == link.getObjectId() and \ + bridge.dest == exchange and bridge.key == routingKey and not bridge.srcIsQueue: + if not _quiet: + raise Exception("Duplicate Route - ignoring: %s(%s)" % (exchange, routingKey)) + sys.exit(0) + + if _verbose: + print "Creating inter-broker binding..." + res = link.bridge(_durable, exchange, exchange, routingKey, tag, excludes, False, _srclocal, dynamic, _ack) + if res.status != 0: + raise Exception(res.text) + if _verbose: + print "Bridge method returned:", res.status, res.text + + def addQueueRoute(self, remoteBroker, exchange, queue): + self.addLink(remoteBroker) + link = self.getLink() if link == None: - print "Protocol Error - Missing link ID" - sys.exit (1) + raise Exception("Link failed to create") - bridges = mc.syncGetObjects (self.mch, "bridge") + bridges = self.qmf.getObjects(_class="bridge") for bridge in bridges: - if bridge.linkRef == link.id and bridge.dest == exchange and bridge.key == routingKey: + if bridge.linkRef == link.getObjectId() and \ + bridge.dest == exchange and bridge.src == queue and bridge.srcIsQueue: if not _quiet: - print "Duplicate Route - ignoring: %s(%s)" % (exchange, routingKey) - sys.exit (1) - sys.exit (0) + raise Exception("Duplicate Route - ignoring: %s(%s)" % (exchange, queue)) + sys.exit(0) if _verbose: print "Creating inter-broker binding..." - bridgeArgs = {} - bridgeArgs["durable"] = _durable - bridgeArgs["src"] = exchange - bridgeArgs["dest"] = exchange - bridgeArgs["key"] = routingKey - bridgeArgs["tag"] = tag - bridgeArgs["excludes"] = excludes - bridgeArgs["srcIsQueue"] = 0 - bridgeArgs["srcIsLocal"] = 0 - res = mc.syncCallMethod (self.mch, link.id, link.classKey, "bridge", bridgeArgs) - if res.status == 4: - print "Can't create a durable route on a non-durable link" - sys.exit(1) + res = link.bridge(_durable, queue, exchange, "", "", "", True, _srclocal, False, _ack) + if res.status != 0: + raise Exception(res.text) if _verbose: - print "Bridge method returned:", res.status, res.statusText + print "Bridge method returned:", res.status, res.text - def DelRoute (self, srcBroker, exchange, routingKey): - self.src = Broker (srcBroker) - mc = self.mclient + def delQueueRoute(self, remoteBroker, exchange, queue): + self.remote = BrokerURL(remoteBroker) + link = self.getLink() + if link == None: + if not _quiet: + raise Exception("No link found from %s to %s" % (self.remote.name(), self.local.name())) + sys.exit(0) + + bridges = self.qmf.getObjects(_class="bridge") + for bridge in bridges: + if bridge.linkRef == link.getObjectId() and \ + bridge.dest == exchange and bridge.src == queue and bridge.srcIsQueue: + if _verbose: + print "Closing bridge..." + res = bridge.close() + if res.status != 0: + raise Exception("Error closing bridge: %d - %s" % (res.status, res.text)) + if len(bridges) == 1 and _dellink: + link = self.getLink() + if link == None: + sys.exit(0) + if _verbose: + print "Last bridge on link, closing link..." + res = link.close() + if res.status != 0: + raise Exception("Error closing link: %d - %s" % (res.status, res.text)) + sys.exit(0) + if not _quiet: + raise Exception("Route not found") - link = self.getLink () + def delRoute(self, remoteBroker, exchange, routingKey, dynamic=False): + self.remote = BrokerURL(remoteBroker) + link = self.getLink() if link == None: if not _quiet: - print "No link found from %s to %s" % (self.src.name(), self.dest.name()) - sys.exit (1) - sys.exit (0) + raise Exception("No link found from %s to %s" % (self.remote.name(), self.local.name())) + sys.exit(0) - bridges = mc.syncGetObjects (self.mch, "bridge") + bridges = self.qmf.getObjects(_class="bridge") for bridge in bridges: - if bridge.linkRef == link.id and bridge.dest == exchange and bridge.key == routingKey: + if bridge.linkRef == link.getObjectId() and bridge.dest == exchange and bridge.key == routingKey \ + and bridge.dynamic == dynamic: if _verbose: print "Closing bridge..." - res = mc.syncCallMethod (self.mch, bridge.id, bridge.classKey, "close") + res = bridge.close() if res.status != 0: - print "Error closing bridge: %d - %s" % (res.status, res.statusText) - sys.exit (1) - if len (bridges) == 1 and _dellink: - link = self.getLink () + raise Exception("Error closing bridge: %d - %s" % (res.status, res.text)) + if len(bridges) == 1 and _dellink: + link = self.getLink() if link == None: - sys.exit (0) + sys.exit(0) if _verbose: print "Last bridge on link, closing link..." - res = mc.syncCallMethod (self.mch, link.id, link.classKey, "close") + res = link.close() if res.status != 0: - print "Error closing link: %d - %s" % (res.status, res.statusText) - sys.exit (1) - sys.exit (0) + raise Exception("Error closing link: %d - %s" % (res.status, res.text)) + sys.exit(0) if not _quiet: - print "Route not found" - sys.exit (1) + raise Exception("Route not found") - def ListRoutes (self): - mc = self.mclient - links = mc.syncGetObjects (self.mch, "link") - bridges = mc.syncGetObjects (self.mch, "bridge") + def listRoutes(self): + links = self.qmf.getObjects(_class="link") + bridges = self.qmf.getObjects(_class="bridge") for bridge in bridges: myLink = None for link in links: - if bridge.linkRef == link.id: + if bridge.linkRef == link.getObjectId(): myLink = link break if myLink != None: - print "%s %s:%d %s %s" % (self.dest.name(), myLink.host, myLink.port, bridge.dest, bridge.key) + if bridge.dynamic: + keyText = "<dynamic>" + else: + keyText = bridge.key + print "%s %s:%d %s %s" % (self.local.name(), myLink.host, myLink.port, bridge.dest, keyText) - def ClearAllRoutes (self): - mc = self.mclient - links = mc.syncGetObjects (self.mch, "link") - bridges = mc.syncGetObjects (self.mch, "bridge") + def clearAllRoutes(self): + links = self.qmf.getObjects(_class="link") + bridges = self.qmf.getObjects(_class="bridge") for bridge in bridges: if _verbose: myLink = None for link in links: - if bridge.linkRef == link.id: + if bridge.linkRef == link.getObjectId(): myLink = link break if myLink != None: print "Deleting Bridge: %s:%d %s %s... " % (myLink.host, myLink.port, bridge.dest, bridge.key), - res = mc.syncCallMethod (self.mch, bridge.id, bridge.classKey, "close") + res = bridge.close() if res.status != 0: - print "Error: %d - %s" % (res.status, res.statusText) + print "Error: %d - %s" % (res.status, res.text) elif _verbose: print "Ok" if _dellink: - links = mc.syncGetObjects (self.mch, "link") + links = self.qmf.getObjects(_class="link") for link in links: if _verbose: print "Deleting Link: %s:%d... " % (link.host, link.port), - res = mc.syncCallMethod (self.mch, link.id, link.classKey, "close") + res = link.close() if res.status != 0: - print "Error: %d - %s" % (res.status, res.statusText) + print "Error: %d - %s" % (res.status, res.text) elif _verbose: print "Ok" +class RoutePair: + def __init__(self, fromUrl, toUrl): + self.fromUrl = fromUrl + self.toUrl = toUrl + self.bidir = False + + def __repr__(self): + if self.bidir: + delimit = "<=>" + else: + delimit = " =>" + return "%s %s %s" % (self.fromUrl, delimit, self.toUrl) + + def matches(self, fromUrl, toUrl): + if fromUrl == self.fromUrl and toUrl == self.toUrl: + return True + if toUrl == self.fromUrl and fromUrl == self.toUrl: + self.bidir = True + return True + return False + + def YN(val): if val == 1: return 'Y' @@ -306,12 +405,22 @@ def YN(val): ## try: - longOpts = ("verbose", "quiet", "durable", "del-empty-link") - (optlist, cargs) = getopt.gnu_getopt (sys.argv[1:], "vqde", longOpts) + longOpts = ("verbose", "quiet", "durable", "del-empty-link", "src-local", "transport=", "ack=", "timeout=") + (optlist, encArgs) = getopt.gnu_getopt(sys.argv[1:], "vqdest:", longOpts) except: - Usage () + Usage() + +try: + encoding = locale.getpreferredencoding() + cargs = [a.decode(encoding) for a in encArgs] +except: + cargs = encArgs for opt in optlist: + if opt[0] == "--timeout": + _connTimeout = int(opt[1]) + if _connTimeout == 0: + _connTimeout = None if opt[0] == "-v" or opt[0] == "--verbose": _verbose = True if opt[0] == "-q" or opt[0] == "--quiet": @@ -320,52 +429,96 @@ for opt in optlist: _durable = True if opt[0] == "-e" or opt[0] == "--del-empty-link": _dellink = True - -nargs = len (cargs) + if opt[0] == "-s" or opt[0] == "--src-local": + _srclocal = True + if opt[0] == "-t" or opt[0] == "--transport": + _transport = opt[1] + if opt[0] == "--ack": + _ack = int(opt[1]) + +nargs = len(cargs) if nargs < 2: - Usage () + Usage() if nargs == 2: - destBroker = "localhost" + localBroker = "localhost" else: - destBroker = cargs[2] + if _srclocal: + localBroker = cargs[3] + remoteBroker = cargs[2] + else: + localBroker = cargs[2] + if nargs > 3: + remoteBroker = cargs[3] group = cargs[0] cmd = cargs[1] -rm = RouteManager (destBroker) -rm.ConnectToBroker () -if group == "link": - if cmd == "add": - if nargs != 4: - Usage() - rm.AddLink (cargs[3]) - elif cmd == "del": - if nargs != 4: - Usage() - rm.DelLink (cargs[3]) - elif cmd == "list": - rm.ListLinks () - -elif group == "route": - if cmd == "add": - if nargs < 6 or nargs > 8: - Usage () - - tag = "" - excludes = "" - if nargs > 6: tag = cargs[6] - if nargs > 7: excludes = cargs[7] - rm.AddRoute (cargs[3], cargs[4], cargs[5], tag, excludes) - elif cmd == "del": - if nargs != 6: - Usage () +try: + rm = RouteManager(localBroker) + if group == "link": + if cmd == "add": + if nargs != 4: + Usage() + rm.addLink(remoteBroker) + elif cmd == "del": + if nargs != 4: + Usage() + rm.delLink(remoteBroker) + elif cmd == "list": + rm.listLinks() + + elif group == "dynamic": + if cmd == "add": + if nargs < 5 or nargs > 7: + Usage() + + tag = "" + excludes = "" + if nargs > 5: tag = cargs[5] + if nargs > 6: excludes = cargs[6] + rm.addRoute(remoteBroker, cargs[4], "", tag, excludes, dynamic=True) + elif cmd == "del": + if nargs != 5: + Usage() + else: + rm.delRoute(remoteBroker, cargs[4], "", dynamic=True) + + elif group == "route": + if cmd == "add": + if nargs < 6 or nargs > 8: + Usage() + + tag = "" + excludes = "" + if nargs > 6: tag = cargs[6] + if nargs > 7: excludes = cargs[7] + rm.addRoute(remoteBroker, cargs[4], cargs[5], tag, excludes, dynamic=False) + elif cmd == "del": + if nargs != 6: + Usage() + rm.delRoute(remoteBroker, cargs[4], cargs[5], dynamic=False) + elif cmd == "map": + rm.mapRoutes() else: - rm.DelRoute (cargs[3], cargs[4], cargs[5]) - else: - if cmd == "list": - rm.ListRoutes () - elif cmd == "flush": - rm.ClearAllRoutes () + if cmd == "list": + rm.listRoutes() + elif cmd == "flush": + rm.clearAllRoutes() + else: + Usage() + + elif group == "queue": + if nargs != 6: + Usage() + if cmd == "add": + rm.addQueueRoute(remoteBroker, exchange=cargs[4], queue=cargs[5]) + elif cmd == "del": + rm.delQueueRoute(remoteBroker, exchange=cargs[4], queue=cargs[5]) else: - Usage () -rm.Disconnect () + Usage() + +except Exception,e: + print "Failed: %s - %s" % (e.__class__.__name__, e) + sys.exit(1) + +rm.disconnect() |