QPID-1288 - New (and greatly improved) API for management console access

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

1 files changed, 953 insertions, 0 deletions

diff --git a/python/qpid/qmfconsole.py b/python/qpid/qmfconsole.py
new file mode 100644
index 0000000000..96633bca0d
--- /dev/null
+++ b/python/qpid/qmfconsole.py
@@ -0,0 +1,953 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+""" Console API for Qpid Management Framework """
+
+import os
+import qpid
+import struct
+import socket
+from qpid.peer       import Closed
+from qpid.connection import Connection, ConnectionFailed
+from qpid.datatypes  import uuid4
+from qpid.util       import connect
+from datatypes       import Message, RangedSet
+from threading       import Lock, Condition
+from codec010        import StringCodec as Codec
+from time            import time
+from cStringIO       import StringIO
+
+class Console:
+  """ To access the asynchronous operations, a class must be derived from
+  Console with overrides of any combination of the available methods. """
+
+  def newPackage(self, name):
+    """ Invoked when a QMF package is discovered. """
+    pass
+
+  def newClass(self, classKey):
+    """ Invoked when a new class is discovered.  Session.getSchema can be
+    used to obtain details about the class."""
+    pass
+
+  def newAgent(self, broker, agent):
+    """ Invoked when a QMF agent is discovered. """
+    pass
+
+  def objectProps(self, broker, id, record):
+    """ Invoked when an object is updated. """
+    global B
+    if record.getClassKey()[1] == "broker":
+      B = record
+
+  def objectStats(self, broker, id, record):
+    """ Invoked when an object is updated. """
+    pass
+
+  def event(self, broker, event):
+    """ Invoked when an event is raised. """
+    pass
+
+  def heartbeat(self, agent, timestamp):
+    """ """
+    pass
+
+  def brokerInfo(self, broker):
+    """ """
+    pass
+
+class Session:
+  """
+  An instance of the Session class represents a console session running
+  against one or more QMF brokers.  A single instance of Session is needed
+  to interact with the management framework as a console.
+  """
+  _CONTEXT_SYNC     = 1
+  _CONTEXT_STARTUP  = 2
+  _CONTEXT_MULTIGET = 3
+
+  GET_WAIT_TIME = 10
+
+  def __init__(self, console=None):
+    """
+    Initialize a session.  If the console argument is provided, the
+    more advanced asynchronous features are available.  If console is
+    defaulted, the session will operate in a simpler, synchronous manner.
+    """
+    self.console          = console
+    self.brokers          = []
+    self.packages         = {}
+    self.seqMgr           = SequenceManager()
+    self.cv               = Condition()
+    self.syncSequenceList = []
+    self.getResult        = []
+
+  def addBroker(self, host="localhost", port=5672,
+                authMech="PLAIN", authName="guest", authPass="guest"):
+    """ Connect to a Qpid broker.  Returns an object of type Broker. """
+    broker = Broker(self, host, port, authMech, authName, authPass)
+    self.brokers.append(broker)
+    return broker
+
+  def delBroker(self, broker):
+    """ Disconnect from a broker.  The 'broker' argument is the object
+    returned from the addBroker call """
+    broker._shutdown()
+    self.brokers.remove(broker)
+    del broker
+
+  def getPackages(self):
+    """ Get the list of known QMF packages """
+    for broker in self.brokers:
+      broker._waitForStable()
+    list = []
+    for package in self.packages:
+      list.append(package)
+    return list
+
+  def getClasses(self, packageName):
+    """ Get the list of known classes within a QMF package """
+    for broker in self.brokers:
+      broker._waitForStable()
+    list = []
+    if packageName in self.packages:
+      for cname, hash in self.packages[packageName]:
+        list.append((packageName, cname, hash))
+    return list
+
+  def getSchema(self, classKey):
+    """ Get the schema for a QMF class """
+    for broker in self.brokers:
+      broker._waitForStable()
+    pname, cname, hash = classKey
+    if pname in self.packages:
+      if (cname, hash) in self.packages[pname]:
+        return self.packages[pname][(cname, hash)]
+
+  def getAgents(self):
+    """ Get a list of currently known agents """
+    for broker in self.brokers:
+      broker._waitForStable()
+    pass
+
+  def getObjects(self, **kwargs):
+    """ Get a list of objects from QMF agents.
+    All arguments are passed by name(keyword).
+
+    The class for queried objects may be specified in one of the following ways:
+
+    schema = <schema> - supply a schema object returned from getSchema
+    key = <key>       - supply a classKey from the list returned by getClasses
+    name = <name>     - supply a class name as a string
+
+    If objects should be obtained from only one agent, use the following argument.
+    Otherwise, the query will go to all agents.
+
+    agent = <agent> - supply an agent from the list returned by getAgents
+
+    If the get query is to be restricted to one broker (as opposed to all connected brokers),
+    add the following argument:
+
+    broker = <broker> - supply a broker as returned by addBroker
+    """
+    if "broker" in kwargs:
+      brokerList = [].append(kwargs["broker"])
+    else:
+      brokerList = self.brokers
+    for broker in brokerList:
+      broker._waitForStable()
+
+    agentList = []
+    if "agent" in kwargs:
+      agent = kwargs["agent"]
+      if agent.broker not in brokerList:
+        raise Exception("Supplied agent is not accessible through the supplied broker")
+      agentList = append(agent)
+    else:
+      for broker in brokerList:
+        for agent in broker.getAgents():
+          agentList.append(agent)
+
+    cname = None
+    if   "schema" in kwargs: pname, cname, hash = kwargs["schema"].getKey()
+    elif "key"    in kwargs: pname, cname, hash = kwargs["key"]
+    elif "name"   in kwargs: pname, cname, hash = None, kwargs["name"], None
+    if cname == None:
+      raise Exception("No class supplied, use 'schema', 'key', or 'name' argument")
+    map = {}
+    map["_class"] = cname
+    if pname != None: map["_package"] = pname
+    if hash  != None: map["_hash"]    = hash
+
+    self.getResult = []
+    for agent in agentList:
+      broker = agent.broker
+      sendCodec = Codec(broker.conn.spec)
+      self.cv.acquire()
+      seq = self.seqMgr._reserve(self._CONTEXT_MULTIGET)
+      self.syncSequenceList.append(seq)
+      self.cv.release()
+      broker._setHeader(sendCodec, 'G', seq)
+      sendCodec.write_map(map)
+      smsg = broker._message(sendCodec.encoded, "agent.%d" % agent.bank)
+      broker._send(smsg)
+
+    starttime = time()
+    timeout = False
+    self.cv.acquire()
+    while len(self.syncSequenceList) > 0:
+      self.cv.wait(self.GET_WAIT_TIME)
+      if time() - starttime > self.GET_WAIT_TIME:
+        for pendingSeq in self.syncSequenceList:
+          self.seqMgr._release(pendingSeq)
+        self.syncSequenceList = []
+        timeout = True
+    self.cv.release()
+
+    if len(self.getResult) == 0 and timeout:
+      raise RuntimeError("No agent responded within timeout period")
+    return self.getResult
+
+  def setEventFilter(self, **kwargs):
+    """ """
+    pass
+
+  def _handleBrokerResp(self, broker, codec, seq):
+    broker.brokerId = codec.read_uuid()
+    if self.console != None:
+      self.console.brokerInfo(broker)
+
+    # Send a package request
+    # (effectively inc and dec outstanding by not doing anything)
+    sendCodec = Codec(broker.conn.spec)
+    seq = self.seqMgr._reserve(self._CONTEXT_STARTUP)
+    broker._setHeader(sendCodec, 'P', seq)
+    smsg = broker._message(sendCodec.encoded)
+    broker._send(smsg)
+
+  def _handlePackageInd(self, broker, codec, seq):
+    pname = str(codec.read_str8())
+    self.cv.acquire()
+    if pname not in self.packages:
+      self.packages[pname] = {}
+      self.cv.release()
+      if self.console != None:
+        self.console.newPackage(pname)
+    else:
+      self.cv.release()
+
+    # Send a class request
+    broker._incOutstanding()
+    sendCodec = Codec(broker.conn.spec)
+    seq = self.seqMgr._reserve(self._CONTEXT_STARTUP)
+    broker._setHeader(sendCodec, 'Q', seq)
+    sendCodec.write_str8(pname)
+    smsg = broker._message(sendCodec.encoded)
+    broker._send(smsg)
+
+  def _handleCommandComplete(self, broker, codec, seq):
+    code = codec.read_uint32()
+    text = str(codec.read_str8())
+    context = self.seqMgr._release(seq)
+    if context == self._CONTEXT_STARTUP:
+      broker._decOutstanding()
+    elif context == self._CONTEXT_SYNC and seq == broker.syncSequence:
+      broker.cv.acquire()
+      broker.syncInFlight = False
+      broker.cv.notify()
+      broker.cv.release()
+    elif context == self._CONTEXT_MULTIGET and seq in self.syncSequenceList:
+      self.cv.acquire()
+      self.syncSequenceList.remove(seq)
+      if len(self.syncSequenceList) == 0:
+        self.cv.notify()
+      self.cv.release()
+
+  def _handleClassInd(self, broker, codec, seq):
+    pname = str(codec.read_str8())
+    cname = str(codec.read_str8())
+    hash  = codec.read_bin128()
+
+    self.cv.acquire()
+    if pname not in self.packages:
+      self.cv.release()
+      return
+    if (cname, hash) not in self.packages[pname]:
+      # Send a schema request for the unknown class
+      self.cv.release()
+      broker._incOutstanding()
+      sendCodec = Codec(broker.conn.spec)
+      seq = self.seqMgr._reserve(self._CONTEXT_STARTUP)
+      broker._setHeader(sendCodec, 'S', seq)
+      sendCodec.write_str8(pname)
+      sendCodec.write_str8(cname)
+      sendCodec.write_bin128(hash)
+      smsg = broker._message(sendCodec.encoded)
+      broker._send(smsg)
+    else:
+      self.cv.release()
+
+  def _handleMethodResp(self, broker, codec, seq):
+    code = codec.read_uint32()
+    text = str(codec.read_str8())
+    outArgs = {}
+    obj, method = self.seqMgr._release(seq)
+    if code == 0:
+      for arg in method.arguments:
+        if arg.dir.find("O") != -1:
+          outArgs[arg.name] = obj._decodeValue(codec, arg.type)
+    broker.cv.acquire()
+    broker.syncResult = MethodResult(code, text, outArgs)
+    broker.syncInFlight = False
+    broker.cv.notify()
+    broker.cv.release()
+
+  def _handleHeartbeatInd(self, broker, codec, seq):
+    timestamp = codec.read_uint64()
+    pass
+
+  def _handleEventInd(self, broker, codec, seq):
+    pass
+
+  def _handleSchemaResp(self, broker, codec, seq):
+    pname = str(codec.read_str8())
+    cname = str(codec.read_str8())
+    hash  = codec.read_bin128()
+    classKey = (pname, cname, hash)
+    _class = SchemaClass(classKey, codec)
+    self.cv.acquire()
+    self.packages[pname][(cname, hash)] = _class
+    self.cv.release()
+    broker._decOutstanding()
+    if self.console != None:
+      self.console.newClass(classKey)
+
+  def _handleContentInd(self, broker, codec, seq, prop=False, stat=False):
+    pname = str(codec.read_str8())
+    cname = str(codec.read_str8())
+    hash  = codec.read_bin128()
+    classKey = (pname, cname, hash)
+    self.cv.acquire()
+    if pname not in self.packages:
+      self.cv.release()
+      return
+    if (cname, hash) not in self.packages[pname]:
+      self.cv.release()
+      return
+    self.cv.release()
+    schema = self.packages[pname][(cname, hash)]
+    object = Object(self, broker, schema, codec, prop, stat)
+
+    self.cv.acquire()
+    if seq in self.syncSequenceList:
+      self.getResult.append(object)
+      self.cv.release()
+      return
+    self.cv.release()
+
+    if self.console != None:
+      if prop:
+        self.console.objectProps(broker, object.getObjectId(), object)
+      if stat:
+        self.console.objectStats(broker, object.getObjectId(), object)
+
+class Package:
+  """ """
+  def __init__(self, name):
+    self.name = name
+
+class ClassKey:
+  """ """
+  def __init__(self, package, className, hash):
+    self.package = package
+    self.className = className
+    self.hash = hash
+
+class SchemaClass:
+  """ """
+  def __init__(self, key, codec):
+    self.classKey = key
+    self.properties = []
+    self.statistics = []
+    self.methods    = []
+    self.events     = []
+
+    propCount   = codec.read_uint16()
+    statCount   = codec.read_uint16()
+    methodCount = codec.read_uint16()
+    eventCount  = codec.read_uint16()
+
+    for idx in range(propCount):
+      self.properties.append(SchemaProperty(codec))
+    for idx in range(statCount):
+      self.statistics.append(SchemaStatistic(codec))
+    for idx in range(methodCount):
+      self.methods.append(SchemaMethod(codec))
+    for idx in range(eventCount):
+      self.events.append(SchemaEvent(codec))
+
+  def __repr__(self):
+    pname, cname, hash = self.classKey
+    result = "Class: %s:%s " % (pname, cname)
+    result += "(%08x-%04x-%04x-%04x-%04x%08x)" % struct.unpack ("!LHHHHL", hash)
+    return result
+
+  def getKey(self):
+    """ """
+    return self.classKey
+
+  def getProperties(self):
+    """ """
+    return self.properties
+
+  def getStatistics(self):
+    """ """
+    return self.statistics
+
+  def getMethods(self):
+    """ """
+    return self.methods
+
+  def getEvents(self):
+    """ """
+    return self.events
+
+class SchemaProperty:
+  """ """
+  def __init__(self, codec):
+    map = codec.read_map()
+    self.name     = str(map["name"])
+    self.type     = map["type"]
+    self.access   = map["access"]
+    self.index    = map["index"] != 0
+    self.optional = map["optional"] != 0
+    self.unit     = None
+    self.min      = None
+    self.max      = None
+    self.maxlan   = None
+    self.desc     = None
+
+    for key, value in map.items():
+      if   key == "unit"   : self.unit   = str(value)
+      elif key == "min"    : self.min    = value
+      elif key == "max"    : self.max    = value
+      elif key == "maxlen" : self.maxlen = value
+      elif key == "desc"   : self.desc   = str(value)
+
+class SchemaStatistic:
+  """ """
+  def __init__(self, codec):
+    map = codec.read_map()
+    self.name     = str(map["name"])
+    self.type     = map["type"]
+    self.unit     = None
+    self.desc     = None
+
+    for key, value in map.items():
+      if   key == "unit" : self.unit = str(value)
+      elif key == "desc" : self.desc = str(value)
+
+class SchemaMethod:
+  """ """
+  def __init__(self, codec):
+    map = codec.read_map()
+    self.name = str(map["name"])
+    argCount  = map["argCount"]
+    if "desc" in map:
+      self.desc = str(map["desc"])
+    else:
+      self.desc = None
+    self.arguments = []
+
+    for idx in range(argCount):
+      self.arguments.append(SchemaArgument(codec, methodArg=True))
+
+class SchemaEvent:
+  """ """
+  def __init__(self, codec):
+    map = codec.read_map()
+    self.name = str(map["name"])
+    argCount  = map["argCount"]
+    if "desc" in map:
+      self.desc = str(map["desc"])
+    else:
+      self.desc = None
+    self.arguments = []
+
+    for idx in range(argCount):
+      self.arguments.append(SchemaArgument(codec, methodArg=False))
+
+class SchemaArgument:
+  """ """
+  def __init__(self, codec, methodArg):
+    map = codec.read_map()
+    self.name    = str(map["name"])
+    self.type    = map["type"]
+    if methodArg:
+      self.dir   = str(map["dir"].upper())
+    self.unit    = None
+    self.min     = None
+    self.max     = None
+    self.maxlen  = None
+    self.desc    = None
+    self.default = None
+
+    for key, value in map.items():
+      if   key == "unit"    : self.unit    = str(value)
+      elif key == "min"     : self.min     = value
+      elif key == "max"     : self.max     = value
+      elif key == "maxlen"  : self.maxlen  = value
+      elif key == "desc"    : self.desc    = str(value)
+      elif key == "default" : self.default = str(value)
+
+class ObjectId(object):
+  """ Object that represents QMF object identifiers """
+  def __init__(self, codec, first=0, second=0):
+    if codec:
+      self.first  = codec.read_uint64()
+      self.second = codec.read_uint64()
+    else:
+      self.first = first
+      self.second = second
+
+  def __cmp__(self, other):
+    if other == None:
+      return 1
+    if self.first < other.first:
+      return -1
+    if self.first > other.first:
+      return 1
+    if self.second < other.second:
+      return -1
+    if self.second > other.second:
+      return 1
+    return 0
+
+  def __repr__(self):
+    return "%08x-%04x-%04x-%04x-%04x%08x" % ((self.first  & 0xFFFFFFFF00000000) >> 32,
+                                             (self.first  & 0x00000000FFFF0000) >> 16,
+                                             (self.first  & 0x000000000000FFFF),
+                                             (self.second & 0xFFFF000000000000) >> 48,
+                                             (self.second & 0x0000FFFF00000000) >> 32,
+                                             (self.second & 0x00000000FFFFFFFF))
+
+  def index(self):
+    return (self.first, self.second)
+
+  def getFlags(self):
+    return (self.first & 0xF000000000000000) >> 60
+
+  def getSequence(self):
+    return (self.first & 0x0FFF000000000000) >> 48
+
+  def getBroker(self):
+    return (self.first & 0x0000FFFFF0000000) >> 28
+
+  def getBank(self):
+    return self.first & 0x000000000FFFFFFF
+
+  def getObject(self):
+    return self.second
+
+  def isDurable(self):
+    return self.getSequence() == 0
+
+  def encode(self, codec):
+    codec.write_uint64(self.first)
+    codec.write_uint64(self.second)
+
+
+class Object(object):
+  """ """
+  def __init__(self, session, broker, schema, codec, prop, stat):
+    """ """
+    self.session = session
+    self.broker  = broker
+    self.schema  = schema
+    self.currentTime = codec.read_uint64()
+    self.createTime  = codec.read_uint64()
+    self.deleteTime  = codec.read_uint64()
+    self.objectId    = ObjectId(codec)
+    self.properties  = []
+    self.statistics  = []
+    if prop:
+      notPresent = self._parsePresenceMasks(codec, schema)
+      for property in schema.getProperties():
+        if property.name in notPresent:
+          self.properties.append((property, None))
+        else:
+          self.properties.append((property, self._decodeValue(codec, property.type)))
+    if stat:
+      for statistic in schema.getStatistics():
+        self.statistics.append((statistic, self._decodeValue(codec, statistic.type)))
+
+  def getObjectId(self):
+    """ """
+    return self.objectId
+
+  def getClassKey(self):
+    """ """
+    return self.schema.getKey()
+
+  def getSchema(self):
+    """ """
+    return self.schema
+
+  def getTimestamps(self):
+    """ """
+    return self.currentTime, self.createTime, self.deleteTime
+
+  def getIndex(self):
+    """ """
+    result = ""
+    for property, value in self.properties:
+      if property.index:
+        if result != "":
+          result += ":"
+        result += str(value)
+    return result
+
+  def __repr__(self):
+    return self.getIndex()
+
+  def __getattr__(self, name):
+    for method in self.schema.getMethods():
+      if name == method.name:
+        return lambda *args, **kwargs : self._invoke(name, args, kwargs)
+    for property, value in self.properties:
+      if name == property.name:
+        return value
+    for statistic, value in self.statistics:
+      if name == statistic.name:
+        return value
+
+  def _invoke(self, name, args, kwargs):
+    for method in self.schema.getMethods():
+      if name == method.name:
+        aIdx = 0
+        sendCodec = Codec(self.broker.conn.spec)
+        seq = self.session.seqMgr._reserve((self, method))
+        self.broker._setHeader(sendCodec, 'M', seq)
+        self.objectId.encode(sendCodec)
+        pname, cname, hash = self.schema.getKey()
+        sendCodec.write_str8(pname)
+        sendCodec.write_str8(cname)
+        sendCodec.write_bin128(hash)
+        sendCodec.write_str8(name)
+        for arg in method.arguments:
+          if arg.dir.find("I") != -1:
+            self._encodeValue(sendCodec, args[aIdx], arg.type)
+            aIdx += 1
+        smsg = self.broker._message(sendCodec.encoded, "agent." + str(self.objectId.getBank()))
+        self.broker._send(smsg)
+        self.broker.cv.acquire()
+        self.broker.syncInFlight = True
+        starttime = time()
+        while self.broker.syncInFlight:
+          self.broker.cv.wait(self.broker.SYNC_TIME)
+          if time() - starttime > self.broker.SYNC_TIME:
+            self.broker.cv.release()
+            self.session.seqMgr._release(seq)
+            raise RuntimeError("Timed out waiting for method to respond")
+        self.broker.cv.release()
+        return self.broker.syncResult
+      else:
+        raise Exception("Invalid Method (software defect)")
+
+  def _parsePresenceMasks(self, codec, schema):
+    excludeList = []
+    bit = 0
+    for property in schema.getProperties():
+      if property.optional:
+        if bit == 0:
+          mask = codec.read_uint8()
+          bit = 1
+        if (mask & bit) == 0:
+          excludeList.append(property.name)
+        bit *= 2
+        if bit == 256:
+          bit = 0
+    return excludeList
+
+  def _decodeValue(self, codec, typecode):
+    """ Decode, from the codec, a value based on its typecode. """
+    if   typecode == 1:  data = codec.read_uint8()     # U8
+    elif typecode == 2:  data = codec.read_uint16()    # U16
+    elif typecode == 3:  data = codec.read_uint32()    # U32
+    elif typecode == 4:  data = codec.read_uint64()    # U64
+    elif typecode == 6:  data = str(codec.read_str8()) # SSTR
+    elif typecode == 7:  data = codec.read_vbin32()    # LSTR
+    elif typecode == 8:  data = codec.read_int64()     # ABSTIME
+    elif typecode == 9:  data = codec.read_uint64()    # DELTATIME
+    elif typecode == 10: data = ObjectId(codec)        # REF
+    elif typecode == 11: data = codec.read_uint8()     # BOOL
+    elif typecode == 12: data = codec.read_float()     # FLOAT
+    elif typecode == 13: data = codec.read_double()    # DOUBLE
+    elif typecode == 14: data = codec.read_uuid()      # UUID
+    elif typecode == 15: data = codec.read_map()       # FTABLE
+    elif typecode == 16: data = codec.read_int8()      # S8
+    elif typecode == 17: data = codec.read_int16()     # S16
+    elif typecode == 18: data = codec.read_int32()     # S32
+    elif typecode == 19: data = codec.read_int64()     # S63
+    else:
+      raise ValueError("Invalid type code: %d" % typecode)
+    return data
+
+  def _encodeValue(self, codec, value, typecode):
+    """ Encode, into the codec, a value based on its typecode. """
+    if   typecode == 1:  codec.write_uint8  (int(value))    # U8
+    elif typecode == 2:  codec.write_uint16 (int(value))    # U16
+    elif typecode == 3:  codec.write_uint32 (long(value))   # U32
+    elif typecode == 4:  codec.write_uint64 (long(value))   # U64
+    elif typecode == 6:  codec.write_str8   (value)         # SSTR
+    elif typecode == 7:  codec.write_vbin32 (value)         # LSTR
+    elif typecode == 8:  codec.write_int64  (long(value))   # ABSTIME
+    elif typecode == 9:  codec.write_uint64 (long(value))   # DELTATIME
+    elif typecode == 10: value.encode       (codec)         # REF
+    elif typecode == 11: codec.write_uint8  (int(value))    # BOOL
+    elif typecode == 12: codec.write_float  (float(value))  # FLOAT
+    elif typecode == 13: codec.write_double (double(value)) # DOUBLE
+    elif typecode == 14: codec.write_uuid   (value)         # UUID
+    elif typecode == 15: codec.write_map    (value)         # FTABLE
+    elif typecode == 16: codec.write_int8   (int(value))    # S8
+    elif typecode == 17: codec.write_int16  (int(value))    # S16
+    elif typecode == 18: codec.write_int32  (int(value))    # S32
+    elif typecode == 19: codec.write_int64  (int(value))    # S64
+    else:
+      raise ValueError ("Invalid type code: %d" % typecode)
+
+class MethodResult(object):
+  """ """
+  def __init__(self, status, text, outArgs):
+    """ """
+    self.status  = status
+    self.text    = text
+    self.outArgs = outArgs
+
+  def __getattr__(self, name):
+    if name in self.outArgs:
+      return self.outArgs[name]
+
+  def __repr__(self):
+    return "%s (%d) - %s" % (self.text, self.status, self.outArgs)
+
+class Broker:
+  """ """
+  SYNC_TIME         = 10
+
+  def __init__(self, session, host, port, authMech, authUser, authPass):
+    self.session = session
+    self.agents = []
+    self.agents.append(Agent(self, 0))
+    self.topicBound = False
+    self.cv = Condition()
+    self.syncInFlight = False
+    self.syncRequest = 0
+    self.syncResult = None
+    self.reqsOutstanding = 1
+    self.brokerId  = None
+    err = None
+    try:
+      self.amqpSessionId = "%s.%d" % (os.uname()[1], os.getpid())
+      self.conn = Connection(connect(host, port), username=authUser, password=authPass)
+      self.conn.start()
+      self.replyName = "reply-%s" % self.amqpSessionId
+      self.amqpSession = self.conn.session(self.amqpSessionId)
+      self.amqpSession.auto_sync = False
+      self.amqpSession.queue_declare(queue=self.replyName, exclusive=True, auto_delete=True)
+      self.amqpSession.exchange_bind(exchange="amq.direct",
+                                 queue=self.replyName, binding_key=self.replyName)
+      self.amqpSession.message_subscribe(queue=self.replyName, destination="rdest")
+      self.amqpSession.incoming("rdest").listen(self._replyCb, self._exceptionCb)
+      self.amqpSession.message_set_flow_mode(destination="rdest", flow_mode=1)
+      self.amqpSession.message_flow(destination="rdest", unit=0, value=0xFFFFFFFF)
+      self.amqpSession.message_flow(destination="rdest", unit=1, value=0xFFFFFFFF)
+
+      if self.session.console != None:
+        self.topicName = "topic-%s" % self.amqpSessionId
+        self.amqpSession.queue_declare(queue=self.topicName, exclusive=True, auto_delete=True)
+        self.amqpSession.message_subscribe(queue=self.topicName, destination="tdest")
+        self.amqpSession.incoming("tdest").listen(self._replyCb)
+        self.amqpSession.message_set_flow_mode(destination="tdest", flow_mode=1)
+        self.amqpSession.message_flow(destination="tdest", unit=0, value=0xFFFFFFFF)
+        self.amqpSession.message_flow(destination="tdest", unit=1, value=0xFFFFFFFF)
+
+      codec = Codec(self.conn.spec)
+      self._setHeader(codec, 'B')
+      msg = self._message(codec.encoded)
+      self._send(msg)
+
+    except socket.error, e:
+      err = "Socket Error %s - %s" % (e[0], e[1])
+    except Closed, e:
+      err = "Connect Failed %d - %s" % (e[0], e[1])
+    except ConnectionFailed, e:
+      err = "Connect Failed %d - %s" % (e[0], e[1])
+
+    if err != None:
+      raise Exception(err)
+
+  def getBrokerId(self):
+    """ Get broker's unique identifier (UUID) """
+    return self.brokerId
+
+  def getSessionId(self):
+    """ Get the identifier of the AMQP session to the broker """
+    return self.amqpSessionId
+
+  def getAgents(self):
+    """ Get the list of agents reachable via this broker """
+    return self.agents
+
+  def _setHeader(self, codec, opcode, seq=0):
+    """ Compose the header of a management message. """
+    codec.write_uint8(ord('A'))
+    codec.write_uint8(ord('M'))
+    codec.write_uint8(ord('1'))
+    codec.write_uint8(ord(opcode))
+    codec.write_uint32(seq)
+
+  def _checkHeader(self, codec):
+    """ Check the header of a management message and extract the opcode and class. """
+    octet = chr(codec.read_uint8())
+    if octet != 'A':
+      return None, None
+    octet = chr(codec.read_uint8())
+    if octet != 'M':
+      return None, None
+    octet = chr(codec.read_uint8())
+    if octet != '1':
+      return None, None
+    opcode = chr(codec.read_uint8())
+    seq    = codec.read_uint32()
+    return opcode, seq
+
+  def _message (self, body, routing_key="broker"):
+    dp = self.amqpSession.delivery_properties()
+    dp.routing_key = routing_key
+    mp = self.amqpSession.message_properties()
+    mp.content_type = "x-application/qmf"
+    mp.reply_to = self.amqpSession.reply_to("amq.direct", self.replyName)
+    return Message(dp, mp, body)
+
+  def _send(self, msg, dest="qpid.management"):
+    self.amqpSession.message_transfer(destination=dest, message=msg)
+
+  def _shutdown(self):
+    self.amqpSession.incoming("rdest").stop()
+    if self.session.console != None:
+      self.amqpSession.incoming("tdest").stop()
+    self.amqpSession.close()
+
+  def _waitForStable(self):
+    self.cv.acquire()
+    if self.reqsOutstanding == 0:
+      self.cv.release()
+      return
+    self.syncInFlight = True
+    starttime = time()
+    while self.reqsOutstanding != 0:
+      self.cv.wait(self.SYNC_TIME)
+      if time() - starttime > self.SYNC_TIME:
+        self.cv.release()
+        raise RuntimeError("Timed out waiting for broker to synchronize")
+    self.cv.release()
+
+  def _incOutstanding(self):
+    self.cv.acquire()
+    self.reqsOutstanding += 1
+    self.cv.release()
+
+  def _decOutstanding(self):
+    self.cv.acquire()
+    self.reqsOutstanding -= 1
+    if self.reqsOutstanding == 0 and not self.topicBound and self.session != None:
+      self.topicBound = True
+      self.amqpSession.exchange_bind(exchange="qpid.management", queue=self.topicName, binding_key="mgmt.#")
+    if self.reqsOutstanding == 0 and self.syncInFlight:
+      self.syncInFlight = False
+      self.cv.notify()
+    self.cv.release()
+
+  def _replyCb(self, msg):
+    codec = Codec(self.conn.spec, msg.body)
+    opcode, seq = self._checkHeader(codec)
+    if opcode == None:
+      return
+    if   opcode == 'b': self.session._handleBrokerResp      (self, codec, seq)
+    elif opcode == 'p': self.session._handlePackageInd      (self, codec, seq)
+    elif opcode == 'z': self.session._handleCommandComplete (self, codec, seq)
+    elif opcode == 'q': self.session._handleClassInd        (self, codec, seq)
+    elif opcode == 'm': self.session._handleMethodResp      (self, codec, seq)
+    elif opcode == 'h': self.session._handleHeartbeatInd    (self, codec, seq)
+    elif opcode == 'e': self.session._handleEventInd        (self, codec, seq)
+    elif opcode == 's': self.session._handleSchemaResp      (self, codec, seq)
+    elif opcode == 'c': self.session._handleContentInd      (self, codec, seq, prop=True)
+    elif opcode == 'i': self.session._handleContentInd      (self, codec, seq, stat=True)
+    elif opcode == 'g': self.session._handleContentInd      (self, codec, seq, prop=True, stat=True)
+
+  def _exceptionCb(self, data):
+    pass
+
+class Agent:
+  """ """
+  def __init__(self, broker, bank):
+    self.broker = broker
+    self.bank   = bank
+
+class Event:
+  """ """
+  def __init__(self):
+    pass
+
+class SequenceManager:
+  """ Manage sequence numbers for asynchronous method calls """
+  def __init__(self):
+    self.lock     = Lock()
+    self.sequence = 0
+    self.pending  = {}
+
+  def _reserve(self, data):
+    """ Reserve a unique sequence number """
+    self.lock.acquire()
+    result = self.sequence
+    self.sequence = self.sequence + 1
+    self.pending[result] = data
+    self.lock.release()
+    return result
+
+  def _release(self, seq):
+    """ Release a reserved sequence number """
+    data = None
+    self.lock.acquire()
+    if seq in self.pending:
+      data = self.pending[seq]
+      del self.pending[seq]
+    self.lock.release()
+    return data
+
+
+# TEST
+
+c = Console()
+s = Session(c)
+b = s.addBroker()
+cl = s.getClasses("org.apache.qpid.broker")
+sch = s.getSchema(cl[0])
+