QPID-953 from tross

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

7 files changed, 59 insertions, 18 deletions

diff --git a/qpid/python/commands/qpid-config b/qpid/python/commands/qpid-config
index 20b08933a4..03a0fd8538 100755
--- a/qpid/python/commands/qpid-config
+++ b/qpid/python/commands/qpid-config
@@ -19,7 +19,7 @@
 # under the License.
 #
 
-import os, uuid
+import os
 import getopt
 import sys
 import socket
@@ -97,10 +97,11 @@ class BrokerManager:
     def ConnectToBroker (self):
         try:
             self.spec = qpid.spec.load (_specpath)
+            self.sessionId = "%s.%d" % (os.uname()[1], os.getpid())
             self.conn = Connection (connect (self.broker.host, self.broker.port), self.spec)
             self.conn.start ()
             self.mclient  = managementClient (self.spec)
-            self.mchannel = self.mclient.addChannel (self.conn.session(str(uuid.uuid4())))
+            self.mchannel = self.mclient.addChannel (self.conn.session(self.sessionId))
         except socket.error, e:
             print "Socket Error:", e
             sys.exit (1)
@@ -108,6 +109,9 @@ class BrokerManager:
             print "Connect Failed:", e
             sys.exit (1)
 
+    def Disconnect (self):
+        self.mclient.removeChannel (self.mchannel)
+
     def Overview (self):
         self.ConnectToBroker ()
         mc  = self.mclient
@@ -374,4 +378,4 @@ else:
         bm.Unbind (cargs[1:])
     else:
         Usage ()
-
+bm.Disconnect()
diff --git a/qpid/python/commands/qpid-queue-stats b/qpid/python/commands/qpid-queue-stats
index 6e3579edfd..ff28e5b50c 100755
--- a/qpid/python/commands/qpid-queue-stats
+++ b/qpid/python/commands/qpid-queue-stats
@@ -24,7 +24,6 @@ import getopt
 import sys
 import socket
 import qpid
-import uuid
 from threading       import Condition
 from qpid.management import managementClient
 from qpid.peer       import Closed
@@ -86,14 +85,18 @@ class BrokerManager:
     def ConnectToBroker (self):
         try:
             self.spec     = qpid.spec.load (specpath)
+            self.sessionId = "%s.%d" % (os.uname()[1], os.getpid())
             self.conn     = Connection (connect (self.broker.host, self.broker.port), self.spec)
             self.conn.start ()
             self.mclient  = managementClient (self.spec, None, self.configCb, self.instCb)
-            self.mchannel = self.mclient.addChannel (self.conn.session(str(uuid.uuid4())))
+            self.mchannel = self.mclient.addChannel (self.conn.session(self.sessionId))
         except socket.error, e:
             print "Connect Error:", e
             exit (1)
 
+    def Disconnect (self):
+        self.mclient.removeChannel (self.mchannel)
+
     def configCb (self, context, classKey, row, timestamps):
         className = classKey[1]
         if className != "queue":
@@ -131,12 +134,16 @@ class BrokerManager:
             (name, deltaTime / 1000000000, obj.msgDepthLow, obj.msgDepthHigh, enqueueRate, dequeueRate)
 
 
-    def Overview (self):
+    def Display (self):
         self.ConnectToBroker ()
         print "Queue Name                                     Sec       Depth Range          Enq Rate     Deq Rate"
         print "==================================================================================================="
-        while True:
-            sleep (1)
+        try:
+            while True:
+                sleep (1)
+        except KeyboardInterrupt:
+            pass
+        self.Disconnect ()
 
 ##
 ## Main Program
@@ -156,4 +163,4 @@ for opt in optlist:
 nargs = len (cargs)
 bm  = BrokerManager ()
 bm.SetBroker (Broker (host))
-bm.Overview ()
+bm.Display ()
diff --git a/qpid/python/commands/qpid-route b/qpid/python/commands/qpid-route
index a0755641c9..a7fbb17777 100755
--- a/qpid/python/commands/qpid-route
+++ b/qpid/python/commands/qpid-route
@@ -23,7 +23,7 @@ import getopt
 import sys
 import socket
 import qpid
-import uuid
+import os
 from qpid.management import managementClient
 from qpid.peer       import Closed
 from qpid.connection import Connection
@@ -80,15 +80,19 @@ class RouteManager:
             print "Connecting to broker: %s:%d" % (broker.host, broker.port)
         try:
             self.spec    = qpid.spec.load (_specpath)
+            self.sessionId = "%s.%d" % (os.uname()[1], os.getpid())
             self.conn    = Connection (connect (broker.host, broker.port), self.spec)
             self.conn.start ()
             self.mclient = managementClient (self.spec)
-            self.mch     = self.mclient.addChannel (self.conn.session(str(uuid.uuid4())))
+            self.mch     = self.mclient.addChannel (self.conn.session(self.sessionId))
             self.mclient.syncWaitForStable (self.mch)
         except socket.error, e:
             print "Connect Error:", e
             sys.exit (1)
 
+    def Disconnect (self):
+        self.mclient.removeChannel (self.mch)
+
     def getLink (self):
         links = self.mclient.syncGetObjects (self.mch, "link")
         for link in links:
@@ -268,4 +272,4 @@ else:
         rm.ClearAllRoutes ()
     else:
         Usage ()
-
+rm.Disconnect ()
diff --git a/qpid/python/commands/qpid-tool b/qpid/python/commands/qpid-tool
index 02579b9d35..1aee3a1b7f 100755
--- a/qpid/python/commands/qpid-tool
+++ b/qpid/python/commands/qpid-tool
@@ -111,9 +111,17 @@ class Mcli (Cmd):
 
   def do_EOF (self, data):
     print "quit"
+    try:
+      self.dataObject.do_exit ()
+    except:
+      pass
     return True
 
   def do_quit (self, data):
+    try:
+      self.dataObject.do_exit ()
+    except:
+      pass
     return True
 
   def postcmd (self, stop, line):
diff --git a/qpid/python/qpid/management.py b/qpid/python/qpid/management.py
index 3a7a564e19..3f8d3c9cfe 100644
--- a/qpid/python/qpid/management.py
+++ b/qpid/python/qpid/management.py
@@ -85,6 +85,8 @@ class managementChannel:
     """ Given a channel on an established AMQP broker connection, this method
     opens a session and performs all of the declarations and bindings needed
     to participate in the management protocol. """
+    self.enabled     = True
+    self.ssn         = ssn
     self.sessionId   = ssn.name
     self.topicName   = "mgmt-%s" % self.sessionId
     self.replyName   = "repl-%s" % self.sessionId
@@ -115,16 +117,24 @@ class managementChannel:
     ssn.message_flow (destination="rdest", unit=0, value=0xFFFFFFFF)
     ssn.message_flow (destination="rdest", unit=1, value=0xFFFFFFFF)
 
+  def shutdown (self):
+    self.enabled = False
+    self.ssn.message_cancel (destination="tdest")
+    self.ssn.message_cancel (destination="rdest")
+
   def topicCb (self, msg):
     """ Receive messages via the topic queue on this channel. """
-    self.tcb (self, msg)
+    if self.enabled:
+      self.tcb (self, msg)
 
   def replyCb (self, msg):
     """ Receive messages via the reply queue on this channel. """
-    self.rcb (self, msg)
+    if self.enabled:
+      self.rcb (self, msg)
 
   def send (self, exchange, msg):
-    self.qpidChannel.message_transfer (destination=exchange, message=msg)
+    if self.enabled:
+      self.qpidChannel.message_transfer (destination=exchange, message=msg)
 
   def accept (self, msg):
     self.qpidChannel.message_accept(RangedSet(msg.id))
@@ -193,6 +203,7 @@ class managementClient:
 
   def removeChannel (self, mch):
     """ Remove a previously added channel from management. """
+    mch.shutdown ()
     self.channels.remove (mch)
 
   def callMethod (self, channel, userSequence, objId, className, methodName, args=None):
diff --git a/qpid/python/qpid/managementdata.py b/qpid/python/qpid/managementdata.py
index 327ae96a26..c908483354 100644
--- a/qpid/python/qpid/managementdata.py
+++ b/qpid/python/qpid/managementdata.py
@@ -22,7 +22,7 @@
 import qpid
 import socket
 import struct
-import uuid
+import os
 from qpid.management import managementChannel, managementClient
 from threading       import Lock
 from disp            import Display
@@ -171,6 +171,7 @@ class ManagementData:
     self.lastUnit       = None
     self.methodSeq      = 1
     self.methodsPending = {}
+    self.sessionId      = "%s.%d" % (os.uname()[1], os.getpid())
 
     self.broker = Broker (host)
     self.conn   = Connection (connect (self.broker.host, self.broker.port), self.spec)
@@ -179,10 +180,10 @@ class ManagementData:
     self.mclient = managementClient (self.spec, self.ctrlHandler, self.configHandler,
                                      self.instHandler, self.methodReply)
     self.mclient.schemaListener (self.schemaHandler)
-    self.mch = self.mclient.addChannel (self.conn.session(str(uuid.uuid4())))
+    self.mch = self.mclient.addChannel (self.conn.session(self.sessionId))
 
   def close (self):
-    self.mclient.removeChannel (self.mch)
+    pass
 
   def refName (self, oid):
     if oid == 0:
@@ -626,3 +627,6 @@ class ManagementData:
     methodName = tokens[1]
     args       = tokens[2:]
     self.callMethod (userOid, methodName, args)
+
+  def do_exit (self):
+    self.mclient.removeChannel (self.mch)
diff --git a/qpid/python/tests_0-10/management.py b/qpid/python/tests_0-10/management.py
index e893dbbd87..b2ab617244 100644
--- a/qpid/python/tests_0-10/management.py
+++ b/qpid/python/tests_0-10/management.py
@@ -50,6 +50,7 @@ class ManagementTest (TestBase010):
             self.assertEqual (res.statusText, "OK")
             self.assertEqual (res.sequence,   seq)
             self.assertEqual (res.body,       body)
+        mc.removeChannel (mch)
 
     def test_system_object (self):
         session = self.session
@@ -60,6 +61,7 @@ class ManagementTest (TestBase010):
         mc.syncWaitForStable (mch)
         systems = mc.syncGetObjects (mch, "system")
         self.assertEqual (len (systems), 1)
+        mc.removeChannel (mch)
 
     def test_standard_exchanges (self):
         session = self.session
@@ -81,6 +83,7 @@ class ManagementTest (TestBase010):
         self.assertEqual (exchange.type, "headers")
         exchange = self.findExchange (exchanges, "qpid.management")
         self.assertEqual (exchange.type, "topic")
+        mc.removeChannel (mch)
 
     def findExchange (self, exchanges, name):
         for exchange in exchanges: