Author: cctrieloff
Date: Thu Apr 24 08:24:51 2008
New Revision: 651290
URL: http://svn.apache.org/viewvc?rev=651290&view=rev
Log:
QPID-953 from tross
Modified:
incubator/qpid/trunk/qpid/python/commands/qpid-config
incubator/qpid/trunk/qpid/python/commands/qpid-queue-stats
incubator/qpid/trunk/qpid/python/commands/qpid-route
incubator/qpid/trunk/qpid/python/commands/qpid-tool
incubator/qpid/trunk/qpid/python/qpid/management.py
incubator/qpid/trunk/qpid/python/qpid/managementdata.py
incubator/qpid/trunk/qpid/python/tests_0-10/management.py
Modified: incubator/qpid/trunk/qpid/python/commands/qpid-config
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/commands/qpid-config?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/commands/qpid-config (original)
+++ incubator/qpid/trunk/qpid/python/commands/qpid-config Thu Apr 24 08:24:51
2008
@@ -19,7 +19,7 @@
# under the License.
#
-import os, uuid
+import os
import getopt
import sys
import socket
@@ -97,10 +97,11 @@
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 @@
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 @@
bm.Unbind (cargs[1:])
else:
Usage ()
-
+bm.Disconnect()
Modified: incubator/qpid/trunk/qpid/python/commands/qpid-queue-stats
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/commands/qpid-queue-stats?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/commands/qpid-queue-stats (original)
+++ incubator/qpid/trunk/qpid/python/commands/qpid-queue-stats Thu Apr 24
08:24:51 2008
@@ -24,7 +24,6 @@
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 @@
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 @@
(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 @@
nargs = len (cargs)
bm = BrokerManager ()
bm.SetBroker (Broker (host))
-bm.Overview ()
+bm.Display ()
Modified: incubator/qpid/trunk/qpid/python/commands/qpid-route
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/commands/qpid-route?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/commands/qpid-route (original)
+++ incubator/qpid/trunk/qpid/python/commands/qpid-route Thu Apr 24 08:24:51
2008
@@ -23,7 +23,7 @@
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 @@
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 @@
rm.ClearAllRoutes ()
else:
Usage ()
-
+rm.Disconnect ()
Modified: incubator/qpid/trunk/qpid/python/commands/qpid-tool
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/commands/qpid-tool?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/commands/qpid-tool (original)
+++ incubator/qpid/trunk/qpid/python/commands/qpid-tool Thu Apr 24 08:24:51 2008
@@ -111,9 +111,17 @@
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):
Modified: incubator/qpid/trunk/qpid/python/qpid/management.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/management.py?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/management.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/management.py Thu Apr 24 08:24:51 2008
@@ -85,6 +85,8 @@
""" 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 @@
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 @@
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):
Modified: incubator/qpid/trunk/qpid/python/qpid/managementdata.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/managementdata.py?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/managementdata.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/managementdata.py Thu Apr 24 08:24:51
2008
@@ -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 @@
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 @@
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 @@
methodName = tokens[1]
args = tokens[2:]
self.callMethod (userOid, methodName, args)
+
+ def do_exit (self):
+ self.mclient.removeChannel (self.mch)
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/management.py
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/management.py?rev=651290&r1=651289&r2=651290&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/management.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/management.py Thu Apr 24
08:24:51 2008
@@ -50,6 +50,7 @@
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 @@
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 @@
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:
