summaryrefslogtreecommitdiff
path: root/python/qpid
diff options
context:
space:
mode:
authorGordon Sim <gsim@apache.org>2008-05-12 19:45:22 +0000
committerGordon Sim <gsim@apache.org>2008-05-12 19:45:22 +0000
commit6f145c117cebc1c75e993ad67facd11b8363f3c9 (patch)
tree05f02953d4ae7e9533b0dfccdf44a3f609ec8ef4 /python/qpid
parent277e5e1f1e7bc26e2c8aff065e84ef0861a851c2 (diff)
downloadqpid-python-6f145c117cebc1c75e993ad67facd11b8363f3c9.tar.gz
QPID-1052: Patch from Ted Ross
This patch contains the following: 1) The session-id reported by the management API now matches the session.name in the session table 2) management.py API has a new callback for closed connections 3) qpid-tool uses the closed-connection handler to notify the user of a lost connection git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@655619 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'python/qpid')
-rw-r--r--python/qpid/management.py18
-rw-r--r--python/qpid/managementdata.py13
2 files changed, 26 insertions, 5 deletions
diff --git a/python/qpid/management.py b/python/qpid/management.py
index d8f09d14ab..0e7233fad2 100644
--- a/python/qpid/management.py
+++ b/python/qpid/management.py
@@ -81,7 +81,7 @@ class methodResult:
class managementChannel:
""" This class represents a connection to an AMQP broker. """
- def __init__ (self, ssn, topicCb, replyCb, cbContext, _detlife=0):
+ def __init__ (self, ssn, topicCb, replyCb, exceptionCb, cbContext, _detlife=0):
""" 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. """
@@ -93,6 +93,7 @@ class managementChannel:
self.qpidChannel = ssn
self.tcb = topicCb
self.rcb = replyCb
+ self.ecb = exceptionCb
self.context = cbContext
self.reqsOutstanding = 0
@@ -104,7 +105,7 @@ class managementChannel:
ssn.message_subscribe (queue=self.topicName, destination="tdest")
ssn.message_subscribe (queue=self.replyName, destination="rdest")
- ssn.incoming ("tdest").listen (self.topicCb)
+ ssn.incoming ("tdest").listen (self.topicCb, self.exceptionCb)
ssn.incoming ("rdest").listen (self.replyCb)
ssn.message_set_flow_mode (destination="tdest", flow_mode=1)
@@ -130,6 +131,10 @@ class managementChannel:
if self.enabled:
self.rcb (self, msg)
+ def exceptionCb (self, data):
+ if self.ecb != None:
+ self.ecb (data)
+
def send (self, exchange, msg):
if self.enabled:
self.qpidChannel.message_transfer (destination=exchange, message=msg)
@@ -160,12 +165,13 @@ class managementClient:
#========================================================
# User API - interacts with the class's user
#========================================================
- def __init__ (self, amqpSpec, ctrlCb=None, configCb=None, instCb=None, methodCb=None):
+ def __init__ (self, amqpSpec, ctrlCb=None, configCb=None, instCb=None, methodCb=None, closeCb=None):
self.spec = amqpSpec
self.ctrlCb = ctrlCb
self.configCb = configCb
self.instCb = instCb
self.methodCb = methodCb
+ self.closeCb = closeCb
self.schemaCb = None
self.eventCb = None
self.channels = []
@@ -189,7 +195,7 @@ class managementClient:
def addChannel (self, channel, cbContext=None):
""" Register a new channel. """
- mch = managementChannel (channel, self.topicCb, self.replyCb, cbContext)
+ mch = managementChannel (channel, self.topicCb, self.replyCb, self.exceptCb, cbContext)
self.channels.append (mch)
self.incOutstanding (mch)
@@ -312,6 +318,10 @@ class managementClient:
self.parse (ch, codec, hdr[0], hdr[1])
ch.accept(msg)
+ def exceptCb (self, data):
+ if self.closeCb != None:
+ self.closeCb (data)
+
#========================================================
# Internal Functions
#========================================================
diff --git a/python/qpid/managementdata.py b/python/qpid/managementdata.py
index 1d99cc11bc..bdc299767d 100644
--- a/python/qpid/managementdata.py
+++ b/python/qpid/managementdata.py
@@ -160,11 +160,20 @@ class ManagementData:
finally:
self.lock.release ()
+ def closeHandler (self, reason):
+ print "Connection to broker lost:", reason
+ self.operational = False
+ if self.cli != None:
+ self.cli.setPromptMessage ("Broker Disconnected")
+
def schemaHandler (self, context, className, configs, insts, methods, events):
""" Callback for schema updates """
if className not in self.schema:
self.schema[className] = (configs, insts, methods, events)
+ def setCli (self, cliobj):
+ self.cli = cliobj
+
def __init__ (self, disp, host, username="guest", password="guest",
specfile="../../specs/amqp.0-10.xml"):
self.spec = qpid.spec.load (specfile)
@@ -184,9 +193,11 @@ class ManagementData:
self.conn.start ()
self.mclient = managementClient (self.spec, self.ctrlHandler, self.configHandler,
- self.instHandler, self.methodReply)
+ self.instHandler, self.methodReply, self.closeHandler)
self.mclient.schemaListener (self.schemaHandler)
self.mch = self.mclient.addChannel (self.conn.session(self.sessionId))
+ self.operational = True
+ self.cli = None
def close (self):
pass