QPID-2810: clean up the broker thread properly on shutdown.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@987330 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 75 insertions, 32 deletions

diff --git a/qpid/tools/src/py/qpid-cluster b/qpid/tools/src/py/qpid-cluster
index e02cca7a88..b96d0d3694 100755
--- a/qpid/tools/src/py/qpid-cluster
+++ b/qpid/tools/src/py/qpid-cluster
@@ -90,6 +90,7 @@ class BrokerManager:
         self.brokerName = None
         self.qmf        = None
         self.broker     = None
+        self.brokers    = []
 
     def SetBroker(self, brokerUrl):
         self.url = brokerUrl
@@ -101,8 +102,18 @@ class BrokerManager:
                 self.brokerAgent = a
 
     def Disconnect(self):
-        if self.broker:
-            self.qmf.delBroker(self.broker)
+        """ Release any allocated brokers.  Ignore any failures as the tool is
+        shutting down.
+        """
+        try:
+            if self.broker:
+                self.qmf.delBroker(self.broker)
+                self.broker = None
+            while len(self.brokers):
+                b = self.brokers.pop()
+                self.qmf.delBroker(b)
+        except:
+            pass
 
     def _getClusters(self):
         packages = self.qmf.getPackages()
@@ -195,7 +206,6 @@ class BrokerManager:
         hostList = self._getHostList(memberList)
         self.qmf.delBroker(self.broker)
         self.broker = None
-        self.brokers = []
 
         idx = 0
         for host in hostList:
@@ -238,10 +248,10 @@ class BrokerManager:
         if self.config._delConn and not found:
             print "Client connection '%s' not found" % self.config._delConn
 
-        for broker in self.brokers:
+        while len(self.brokers):
+            broker = self.brokers.pop()
             self.qmf.delBroker(broker)
 
-
 def main(argv=None):
     if argv is None: argv = sys.argv
     try:
@@ -322,6 +332,7 @@ def main(argv=None):
         except KeyboardInterrupt:
             print
         except Exception,e:
+            bm.Disconnect()   # try to deallocate brokers - ignores errors
             if str(e).find("connection aborted") > 0:
                 # we expect this when asking the connected broker to shut down
                 return 0
diff --git a/qpid/tools/src/py/qpid-config b/qpid/tools/src/py/qpid-config
index a97b6cba94..cc7debc80a 100755
--- a/qpid/tools/src/py/qpid-config
+++ b/qpid/tools/src/py/qpid-config
@@ -574,13 +574,16 @@ except KeyboardInterrupt:
     print
 except IOError, e:
     print e
+    bm.Disconnect()
     sys.exit(1)
 except SystemExit, e:
+    bm.Disconnect()
     sys.exit(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()
         sys.exit(1)
  
 while True:
diff --git a/qpid/tools/src/py/qpid-printevents b/qpid/tools/src/py/qpid-printevents
index ed2155ad22..e14d85e12a 100755
--- a/qpid/tools/src/py/qpid-printevents
+++ b/qpid/tools/src/py/qpid-printevents
@@ -59,18 +59,20 @@ ex: localhost, 10.1.1.7:10000, broker-host:10000, guest/guest@localhost
   console = EventConsole()
   session = Session(console, rcvObjects=False, rcvHeartbeats=False, manageConnections=True)
   brokers = []
-  for host in arguments:
-    brokers.append(session.addBroker(host))
-
   try:
-    while (True):
-      sleep(10)
-  except KeyboardInterrupt:
-    for broker in brokers:
-      session.delBroker(broker)
-    print
-    sys.exit(0)
- 
+    for host in arguments:
+      brokers.append(session.addBroker(host))
+    try:
+      while (True):
+        sleep(10)
+    except KeyboardInterrupt:
+      print
+      sys.exit(0)
+  finally:
+    while len(brokers):
+      b = brokers.pop()
+      session.delBroker(b)
+
 if __name__ == '__main__':
   main()
 
diff --git a/qpid/tools/src/py/qpid-route b/qpid/tools/src/py/qpid-route
index ceb0c6313e..be6bdf958c 100755
--- a/qpid/tools/src/py/qpid-route
+++ b/qpid/tools/src/py/qpid-route
@@ -71,6 +71,7 @@ _connTimeout = 10
 
 class RouteManager:
     def __init__(self, localBroker):
+        self.brokerList = {}
         self.local = BrokerURL(localBroker)
         self.remote  = None
         self.qmf = Session()
@@ -79,7 +80,16 @@ class RouteManager:
         self.agent = self.broker.getBrokerAgent()
 
     def disconnect(self):
-        self.qmf.delBroker(self.broker)
+        try:
+            if self.broker:
+                self.qmf.delBroker(self.broker)
+                self.broker = None
+            while len(self.brokerList):
+                b = self.brokerList.popitem()
+                if b[0] != self.local.name():
+                    self.qmf.delBroker(b[1])
+        except:
+            pass  # ignore errors while shutting down
 
     def getLink(self):
         links = self.agent.getObjects(_class="link")
@@ -135,8 +145,7 @@ class RouteManager:
         print
         print "Finding Linked Brokers:"
 
-        brokerList = {}
-        brokerList[self.local.name()] = self.broker
+        self.brokerList[self.local.name()] = self.broker
         print "    %s... Ok" % self.local
 
         added = True
@@ -145,11 +154,11 @@ class RouteManager:
             links = self.qmf.getObjects(_class="link")
             for link in links:
                 url = BrokerURL("%s:%d" % (link.host, link.port))
-                if url.name() not in brokerList:
+                if url.name() not in self.brokerList:
                     print "    %s..." % url.name(),
                     try:
                         b = self.qmf.addBroker("%s:%d" % (link.host, link.port), _connTimeout)
-                        brokerList[url.name()] = b
+                        self.brokerList[url.name()] = b
                         added = True
                         print "Ok"
                     except Exception, e:
@@ -217,10 +226,10 @@ class RouteManager:
                     (toUrl, leftType, left, arrow, fromUrl, rightType, right, bridge.key)
         print
 
-        for broker in brokerList:
-            if broker != self.local.name():
-                self.qmf.delBroker(brokerList[broker])
-
+        while len(self.brokerList):
+            b = self.brokerList.popitem()
+            if b[0] != self.local.name():
+                self.qmf.delBroker(b[1])
 
     def addRoute(self, remoteBroker, exchange, routingKey, tag, excludes, dynamic=False):
         if dynamic and _srclocal:
@@ -463,6 +472,7 @@ else:
 group = cargs[0]
 cmd   = cargs[1]
 
+rm = None
 try:
     rm = RouteManager(localBroker)
     if group == "link":
@@ -528,6 +538,8 @@ try:
             Usage()
 
 except Exception,e:
+    if rm:
+        rm.disconnect()  # try to release broker resources
     print "Failed: %s - %s" % (e.__class__.__name__, e)
     sys.exit(1)
 
diff --git a/qpid/tools/src/py/qpid-stat b/qpid/tools/src/py/qpid-stat
index 848596158f..ae5683ec54 100755
--- a/qpid/tools/src/py/qpid-stat
+++ b/qpid/tools/src/py/qpid-stat
@@ -161,10 +161,16 @@ class BrokerManager(Console):
                 self.brokerAgent = a
 
     def Disconnect(self):
-        if self.broker:
-            self.qmf.delBroker(self.broker)
-        else:
-            for b in self.brokers: self.qmf.delBroker(b.broker)
+        """ Release any allocated brokers.  Ignore any failures as the tool is
+        shutting down.
+        """
+        try:
+            if self.broker:
+                self.qmf.delBroker(self.broker)
+            else:
+                for b in self.brokers: self.qmf.delBroker(b.broker)
+        except:
+            pass
 
     def _getCluster(self):
         packages = self.qmf.getPackages()
@@ -524,6 +530,7 @@ except KeyboardInterrupt:
     print
 except Exception,e:
     print "Failed: %s - %s" % (e.__class__.__name__, e)
+    bm.Disconnect()   # try to deallocate brokers
     raise                               # FIXME aconway 2010-03-03: 
     sys.exit(1)
 
diff --git a/qpid/tools/src/py/qpid-tool b/qpid/tools/src/py/qpid-tool
index 3a2a340080..cebd865501 100755
--- a/qpid/tools/src/py/qpid-tool
+++ b/qpid/tools/src/py/qpid-tool
@@ -191,8 +191,12 @@ class QmfData(Console):
     self.cli = cli
 
   def close(self):
-    self.closing = True
-    self.session.delBroker(self.broker)
+    try:
+      self.closing = True
+      if self.session and self.broker:
+        self.session.delBroker(self.broker)
+    except:
+      pass   # we're shutting down - ignore any errors
 
   def classCompletions(self, text):
     pass
@@ -645,4 +649,8 @@ try:
 except KeyboardInterrupt:
   print
   print "Exiting..."
-  data.close()
+except Exception, e:
+  print "Failed: %s - %s" % (e.__class__.__name__, e)
+
+# alway attempt to cleanup broker resources
+data.close()