diff options
Diffstat (limited to 'python/qpid')
| -rw-r--r-- | python/qpid/management.py | 18 | ||||
| -rw-r--r-- | python/qpid/managementdata.py | 13 |
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 |