#!/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 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 " print " qpid-route [OPTIONS] link del " print " qpid-route [OPTIONS] link list []" print print " qpid-route [OPTIONS] route add [id] [exclude-list]" print " qpid-route [OPTIONS] route del " print " qpid-route [OPTIONS] route list []" print " qpid-route [OPTIONS] route flush []" print print "Options:" 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 print " dest-broker and src-broker are in the form: [username/password@] hostname | ip-address [:]" print " ex: localhost, 10.1.1.7:10000, broker-host:10000, guest/guest@localhost" print sys.exit (1) _verbose = False _quiet = False _durable = False _dellink = False class RouteManager: def __init__ (self, destBroker): self.dest = Broker (destBroker) self.src = None 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") for link in links: if "%s:%d" % (link.host, link.port) == self.src.name (): 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) brokers = mc.syncGetObjects (self.mch, "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 brokers = mc.syncGetObjects (self.mch, "broker") broker = brokers[0] link = self.getLink() if link == None: print "Link not found" sys.exit(1) res = mc.syncCallMethod (self.mch, link.id, link.classKey, "close") if _verbose: print "Close method returned:", res.status, res.statusText def ListLinks (self): mc = self.mclient links = mc.syncGetObjects (self.mch, "link") if len(links) == 0: print "No Links Found" else: print print "Host Port Durable State Last Error" print "===================================================================" for link in links: print "%-16s%-8d %c %-18s%s" % (link.host, link.port, YN(link.durable), link.state, link.lastError) def AddRoute (self, srcBroker, exchange, routingKey, id, excludes): 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) brokers = mc.syncGetObjects (self.mch, "broker") broker = brokers[0] 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 () if link == None: print "Protocol Error - Missing link ID" sys.exit (1) bridges = mc.syncGetObjects (self.mch, "bridge") for bridge in bridges: if bridge.linkRef == link.id and bridge.dest == exchange and bridge.key == routingKey: if not _quiet: print "Duplicate Route - ignoring: %s(%s)" % (exchange, routingKey) sys.exit (1) 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"] = id 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) if _verbose: print "Bridge method returned:", res.status, res.statusText def DelRoute (self, srcBroker, exchange, routingKey): self.src = Broker (srcBroker) mc = self.mclient 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) bridges = mc.syncGetObjects (self.mch, "bridge") for bridge in bridges: if bridge.linkRef == link.id and bridge.dest == exchange and bridge.key == routingKey: if _verbose: print "Closing bridge..." res = mc.syncCallMethod (self.mch, bridge.id, bridge.classKey, "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 () if link == None: sys.exit (0) if _verbose: print "Last bridge on link, closing link..." res = mc.syncCallMethod (self.mch, link.id, link.classKey, "close") if res.status != 0: print "Error closing link: %d - %s" % (res.status, res.statusText) sys.exit (1) sys.exit (0) if not _quiet: print "Route not found" sys.exit (1) def ListRoutes (self): mc = self.mclient links = mc.syncGetObjects (self.mch, "link") bridges = mc.syncGetObjects (self.mch, "bridge") for bridge in bridges: myLink = None for link in links: if bridge.linkRef == link.id: myLink = link break if myLink != None: print "%s %s:%d %s %s" % (self.dest.name(), myLink.host, myLink.port, bridge.dest, bridge.key) def ClearAllRoutes (self): mc = self.mclient links = mc.syncGetObjects (self.mch, "link") bridges = mc.syncGetObjects (self.mch, "bridge") for bridge in bridges: if _verbose: myLink = None for link in links: if bridge.linkRef == link.id: 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") if res.status != 0: print "Error: %d - %s" % (res.status, res.statusText) elif _verbose: print "Ok" if _dellink: links = mc.syncGetObjects (self.mch, "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") if res.status != 0: print "Error: %d - %s" % (res.status, res.statusText) elif _verbose: print "Ok" def YN(val): if val == 1: return 'Y' return 'N' ## ## Main Program ## try: longOpts = ("verbose", "quiet", "durable", "del-empty-link") (optlist, cargs) = getopt.gnu_getopt (sys.argv[1:], "vqde", longOpts) except: Usage () for opt in optlist: if opt[0] == "-v" or opt[0] == "--verbose": _verbose = True if opt[0] == "-q" or opt[0] == "--quiet": _quiet = True if opt[0] == "-d" or opt[0] == "--durable": _durable = True if opt[0] == "-e" or opt[0] == "--del-empty-link": _dellink = True nargs = len (cargs) if nargs < 2: Usage () if nargs == 2: destBroker = "localhost" else: destBroker = cargs[2] 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 () id = "" excludes = "" if nargs > 6: id = cargs[6] if nargs > 7: excludes = cargs[7] rm.AddRoute (cargs[3], cargs[4], cargs[5], id, excludes) elif cmd == "del": if nargs != 6: Usage () else: rm.DelRoute (cargs[3], cargs[4], cargs[5]) else: if cmd == "list": rm.ListRoutes () elif cmd == "flush": rm.ClearAllRoutes () else: Usage () rm.Disconnect ()