path: root/qpid/extras/qmf/src/py/qmf/console.py

#
# 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 platform
import qpid
import struct
import socket
import re
from qpid.datatypes  import UUID
from qpid.datatypes  import timestamp
from qpid.datatypes  import datetime
from qpid.exceptions import Closed
from qpid.session    import SessionDetached
from qpid.connection import Connection, ConnectionFailed, Timeout
from qpid.datatypes  import Message, RangedSet, UUID
from qpid.util       import connect, ssl, URL
from qpid.codec010   import StringCodec as Codec
from threading       import Lock, Condition, Thread, Semaphore
from Queue           import Queue, Empty
from time            import time, strftime, gmtime, sleep
from cStringIO       import StringIO

#import qpid.log
#qpid.log.enable(name="qpid.io.cmd", level=qpid.log.DEBUG)

#===================================================================================================
# CONSOLE
#===================================================================================================
class Console:
  """ To access the asynchronous operations, a class must be derived from
  Console with overrides of any combination of the available methods. """

  def brokerConnected(self, broker):
    """ Invoked when a connection is established to a broker """
    pass

  def brokerConnectionFailed(self, broker):
    """ Invoked when a connection to a broker fails """
    pass

  def brokerDisconnected(self, broker):
    """ Invoked when the connection to a broker is lost """
    pass

  def newPackage(self, name):
    """ Invoked when a QMF package is discovered. """
    pass

  def newClass(self, kind, classKey):
    """ Invoked when a new class is discovered.  Session.getSchema can be
    used to obtain details about the class."""
    pass

  def newAgent(self, agent):
    """ Invoked when a QMF agent is discovered. """
    pass

  def delAgent(self, agent):
    """ Invoked when a QMF agent disconects. """
    pass

  def objectProps(self, broker, record):
    """ Invoked when an object is updated. """
    pass

  def objectStats(self, broker, 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):
    """ Invoked when an agent heartbeat is received. """
    pass

  def brokerInfo(self, broker):
    """ Invoked when the connection sequence reaches the point where broker information is available. """
    pass

  def methodResponse(self, broker, seq, response):
    """ Invoked when a method response from an asynchronous method call is received. """
    pass


#===================================================================================================
# BrokerURL
#===================================================================================================
class BrokerURL(URL):
  def __init__(self, text):
    URL.__init__(self, text)
    if self.port is None:
      if self.scheme == URL.AMQPS:
        self.port = 5671
      else:
        self.port = 5672
    self.authName = None
    self.authPass = None
    if self.user:
      self.authName = str(self.user)
    if self.password:
      self.authPass = str(self.password)

  def name(self):
    return self.host + ":" + str(self.port)

  def match(self, host, port):
    return socket.getaddrinfo(self.host, self.port)[0][4] == socket.getaddrinfo(host, port)[0][4]


#===================================================================================================
# Object
#===================================================================================================
class Object(object):
  """
  This class defines a 'proxy' object representing a real managed object on an agent.
  Actions taken on this proxy are remotely affected on the real managed object.
  """
  def __init__(self, agent, schema, codec=None, prop=None, stat=None, v2Map=None, agentName=None, kwargs={}):
    self._agent   = agent
    self._session = None
    self._broker  = None
    if agent:
      self._session = agent.session
      self._broker  = agent.broker
    self._schema  = schema
    self._properties  = []
    self._statistics  = []
    self._currentTime = None
    self._createTime  = None
    self._deleteTime  = 0
    self._objectId    = None
    if v2Map:
      self.v2Init(v2Map, agentName)
      return

    if self._agent:
      self._currentTime = codec.read_uint64()
      self._createTime  = codec.read_uint64()
      self._deleteTime  = codec.read_uint64()
      self._objectId    = ObjectId(codec)
    if codec:
      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._session._decodeValue(codec, property.type, self._broker)))
      if stat:
        for statistic in schema.getStatistics():
          self._statistics.append((statistic, self._session._decodeValue(codec, statistic.type, self._broker)))
    else:
      for property in schema.getProperties():
        if property.optional:
          self._properties.append((property, None))
        else:
          self._properties.append((property, self._session._defaultValue(property, self._broker, kwargs)))
      for statistic in schema.getStatistics():
          self._statistics.append((statistic, self._session._defaultValue(statistic, self._broker, kwargs)))

  def v2Init(self, omap, agentName):
    if omap.__class__ != dict:
      raise Exception("QMFv2 object data must be a map/dict")
    if '_values' not in omap:
      raise Exception("QMFv2 object must have '_values' element")

    values = omap['_values']
    for prop in self._schema.getProperties():
      if prop.name in values:
        if prop.type == 10: # Reference
          self._properties.append((prop, ObjectId(values[prop.name], agentName=agentName)))
        else:
          self._properties.append((prop, values[prop.name]))
    for stat in self._schema.getStatistics():
      if stat.name in values:
        self._statistics.append((stat, values[stat.name]))
    if '_subtypes' in omap:
      self._subtypes = omap['_subtypes']
    if '_object_id' in omap:
      self._objectId = ObjectId(omap['_object_id'], agentName=agentName)
    else:
      self._objectId = None

    self._currentTime = omap.get("_update_ts", 0)
    self._createTime = omap.get("_create_ts", 0)
    self._deleteTime = omap.get("_delete_ts", 0)

  def getAgent(self):
    """ Return the agent from which this object was sent """
    return self._agent

  def getBroker(self):
    """ Return the broker from which this object was sent """
    return self._broker

  def getV2RoutingKey(self):
    """ Get the QMFv2 routing key to address this object """
    return self._agent.getV2RoutingKey()

  def getObjectId(self):
    """ Return the object identifier for this object """
    return self._objectId

  def getClassKey(self):
    """ Return the class-key that references the schema describing this object. """
    return self._schema.getKey()

  def getSchema(self):
    """ Return the schema that describes this object. """
    return self._schema

  def getMethods(self):
    """ Return a list of methods available for this object. """
    return self._schema.getMethods()

  def getTimestamps(self):
    """ Return the current, creation, and deletion times for this object. """
    return self._currentTime, self._createTime, self._deleteTime

  def isDeleted(self):
    """ Return True iff this object has been deleted. """
    return self._deleteTime != 0

  def isManaged(self):
    """ Return True iff this object is a proxy for a managed object on an agent. """
    return self._objectId and self._agent

  def getIndex(self):
    """ Return a string describing this object's primary key. """
    if self._objectId.isV2:
      return self._objectId.getObject()
    result = u""
    for prop, value in self._properties:
      if prop.index:
        if result != u"":
          result += u":"
        try:
          valstr = unicode(self._session._displayValue(value, prop.type))
        except Exception, e:
          valstr = u"<undecodable>"
        result += valstr
    return result

  def getProperties(self):
    """ Return a list of object properties """
    return self._properties

  def getStatistics(self):
    """ Return a list of object statistics """
    return self._statistics

  def mergeUpdate(self, newer):
    """ Replace properties and/or statistics with a newly received update """
    if not self.isManaged():
      raise Exception("Object is not managed")
    if self._objectId != newer._objectId:
      raise Exception("Objects with different object-ids")
    if len(newer.getProperties()) > 0:
      self._properties = newer.getProperties()
    if len(newer.getStatistics()) > 0:
      self._statistics = newer.getStatistics()
    self._currentTime = newer._currentTime
    self._deleteTime = newer._deleteTime

  def update(self):
    """ Contact the agent and retrieve the lastest property and statistic values for this object. """
    if not self.isManaged():
      raise Exception("Object is not managed")
    obj = self._agent.getObjects(_objectId=self._objectId)
    if obj:
      self.mergeUpdate(obj[0])
    else:
      raise Exception("Underlying object no longer exists")

  def __repr__(self):
    if self.isManaged():
      id = self.getObjectId().__repr__()
    else:
      id = "unmanaged"
    key = self.getClassKey()
    return key.getPackageName() + ":" + key.getClassName() +\
        "[" + id + "] " + self.getIndex().encode("utf8")

  def __getattr__(self, name):
    for method in self._schema.getMethods():
      if name == method.name:
        return lambda *args, **kwargs : self._invoke(name, args, kwargs)
    for prop, value in self._properties:
      if name == prop.name:
        return value
      if name == "_" + prop.name + "_" and prop.type == 10:  # Dereference references
        deref = self._agent.getObjects(_objectId=value)
        if len(deref) != 1:
          return None
        else:
          return deref[0]
    for stat, value in self._statistics:
      if name == stat.name:
        return value

    #
    # Check to see if the name is in the schema.  If so, return None (i.e. this is a not-present attribute)
    #
    for prop in self._schema.getProperties():
      if name == prop.name:
        return None
    for stat in self._schema.getStatistics():
      if name == stat.name:
        return None
    raise Exception("Type Object has no attribute '%s'" % name)

  def __setattr__(self, name, value):
    if name[0] == '_':
      super.__setattr__(self, name, value)
      return

    for prop, unusedValue in self._properties:
      if name == prop.name:
        newprop = (prop, value)
        newlist = []
        for old, val in self._properties:
          if name == old.name:
            newlist.append(newprop)
          else:
            newlist.append((old, val))
        self._properties = newlist
        return
    super.__setattr__(self, name, value)

  def _sendMethodRequest(self, name, args, kwargs, synchronous=False, timeWait=None):
    for method in self._schema.getMethods():
      if name == method.name:
        aIdx = 0
        sendCodec = Codec()
        seq = self._session.seqMgr._reserve((method, synchronous))

        count = 0
        for arg in method.arguments:
          if arg.dir.find("I") != -1:
            count += 1
        if count != len(args):
          raise Exception("Incorrect number of arguments: expected %d, got %d" % (count, len(args)))

        if self._agent.isV2:
          #
          # Compose and send a QMFv2 method request
          #
          call = {}
          call['_object_id'] = self._objectId.asMap()
          call['_method_name'] = name
          argMap = {}
          for arg in method.arguments:
            if arg.dir.find("I") != -1:
              argMap[arg.name] = args[aIdx]
              aIdx += 1
          call['_arguments'] = argMap

          dp = self._broker.amqpSession.delivery_properties()
          dp.routing_key = self.getV2RoutingKey()
          mp = self._broker.amqpSession.message_properties()
          mp.content_type = "amqp/map"
          if self._broker.saslUser:
            mp.user_id = self._broker.saslUser
          mp.correlation_id = str(seq)
          mp.app_id = "qmf2"
          mp.reply_to = self._broker.amqpSession.reply_to("qmf.default.direct", self._broker.v2_direct_queue)
          mp.application_headers = {'qmf.opcode':'_method_request'}
          sendCodec.write_map(call)
          smsg = Message(dp, mp, sendCodec.encoded)
          exchange = "qmf.default.direct"

        else:
          #
          # Associate this sequence with the agent hosting the object so we can correctly
          # route the method-response
          #
          agent = self._broker.getAgent(self._broker.getBrokerBank(), self._objectId.getAgentBank())
          self._broker._setSequence(seq, agent)

          #
          # Compose and send a QMFv1 method request
          #
          self._broker._setHeader(sendCodec, 'M', seq)
          self._objectId.encode(sendCodec)
          self._schema.getKey().encode(sendCodec)
          sendCodec.write_str8(name)

          for arg in method.arguments:
            if arg.dir.find("I") != -1:
              self._session._encodeValue(sendCodec, args[aIdx], arg.type)
              aIdx += 1
          smsg = self._broker._message(sendCodec.encoded, "agent.%d.%s" %
                                       (self._objectId.getBrokerBank(), self._objectId.getAgentBank()))
          exchange = "qpid.management"

        if synchronous:
          try:
            self._broker.cv.acquire()
            self._broker.syncInFlight = True
          finally:
            self._broker.cv.release()
        self._broker._send(smsg, exchange)
        return seq
    return None

  def _invoke(self, name, args, kwargs):
    if not self.isManaged():
      raise Exception("Object is not managed")
    if "_timeout" in kwargs:
      timeout = kwargs["_timeout"]
    else:
      timeout = self._broker.SYNC_TIME

    if "_async" in kwargs and kwargs["_async"]:
      sync = False
      if "_timeout" not in kwargs:
        timeout = None
    else:
      sync = True

    seq = self._sendMethodRequest(name, args, kwargs, sync, timeout)
    if seq:
      if not sync:
        return seq
      self._broker.cv.acquire()
      try:
        starttime = time()
        while self._broker.syncInFlight and self._broker.error == None:
          self._broker.cv.wait(timeout)
          if time() - starttime > timeout:
            raise RuntimeError("Timed out waiting for method to respond")
      finally:
        self._session.seqMgr._release(seq)
        self._broker.cv.release()
      if self._broker.error != None:
        errorText = self._broker.error
        self._broker.error = None
        raise Exception(errorText)
      return self._broker.syncResult
    raise Exception("Invalid Method (software defect) [%s]" % name)

  def _encodeUnmanaged(self, codec):
    codec.write_uint8(20) 
    codec.write_str8(self._schema.getKey().getPackageName())
    codec.write_str8(self._schema.getKey().getClassName())
    codec.write_bin128(self._schema.getKey().getHash())

    # emit presence masks for optional properties
    mask = 0
    bit  = 0
    for prop, value in self._properties:
      if prop.optional:
        if bit == 0:
          bit = 1
        if value:
          mask |= bit
        bit = bit << 1
        if bit == 256:
          bit = 0
          codec.write_uint8(mask)
          mask = 0
    if bit != 0:
      codec.write_uint8(mask)    
    
    # encode properties
    for prop, value in self._properties:
      if value != None: 
        self._session._encodeValue(codec, value, prop.type)

    # encode statistics
    for stat, value in self._statistics:
      self._session._encodeValue(codec, value, stat.type)

  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    


#===================================================================================================
# Session
#===================================================================================================
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

  DEFAULT_GET_WAIT_TIME = 60

  ENCODINGS = {
    str: 7,
    timestamp: 8,
    datetime: 8,  
    int: 9,
    long: 9,
    float: 13,
    UUID: 14,      
    Object: 20,    
    list: 21
    }  


  def __init__(self, console=None, rcvObjects=True, rcvEvents=True, rcvHeartbeats=True,
               manageConnections=False, userBindings=False):
    """
    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.

    The rcvObjects, rcvEvents, and rcvHeartbeats arguments are meaningful only if 'console'
    is provided.  They control whether object updates, events, and agent-heartbeats are
    subscribed to.  If the console is not interested in receiving one or more of the above,
    setting the argument to False will reduce tha bandwidth used by the API.

    If manageConnections is set to True, the Session object will manage connections to
    the brokers.  This means that if a broker is unreachable, it will retry until a connection
    can be established.  If a connection is lost, the Session will attempt to reconnect.

    If manageConnections is set to False, the user is responsible for handing failures.  In
    this case, an unreachable broker will cause addBroker to raise an exception.

    If userBindings is set to False (the default) and rcvObjects is True, the console will
    receive data for all object classes.  If userBindings is set to True, the user must select
    which classes the console shall receive by invoking the bindPackage or bindClass methods.
    This allows the console to be configured to receive only information that is relavant to
    a particular application.  If rcvObjects id False, userBindings has no meaning.
    """
    self.console           = console
    self.brokers           = []
    self.schemaCache       = SchemaCache()
    self.seqMgr            = SequenceManager()
    self.cv                = Condition()
    self.syncSequenceList  = []
    self.getResult         = []
    self.getSelect         = []
    self.error             = None
    self.rcvObjects        = rcvObjects
    self.rcvEvents         = rcvEvents
    self.rcvHeartbeats     = rcvHeartbeats
    self.userBindings      = userBindings
    if self.console == None:
      self.rcvObjects    = False
      self.rcvEvents     = False
      self.rcvHeartbeats = False
    self.v1BindingKeyList, self.v2BindingKeyList = self._bindingKeys()
    self.manageConnections = manageConnections
    # callback filters:
    self.agent_filter = []  # (vendor, product, instance) || v1-agent-label-str
    self.class_filter = []  # (pkg, class)
    self.event_filter = []  # (pkg, event)
    self.agent_heartbeat_min = 10 # minimum agent heartbeat timeout interval
    self.agent_heartbeat_miss = 3 # # of heartbeats to miss before deleting agent

    if self.userBindings and not self.console:
      raise Exception("userBindings can't be set unless a console is provided.")

  def close(self):
    """ Releases all resources held by the session.  Must be called by the
    application when it is done with the Session object.
    """
    self.cv.acquire()
    try:
      while len(self.brokers):
        b = self.brokers.pop()
        try:
          b._shutdown()
        except:
          pass
    finally:
      self.cv.release()

  def _getBrokerForAgentAddr(self, agent_addr):
    try:
      self.cv.acquire()
      key = (1, agent_addr)
      for b in self.brokers:
        if key in b.agents:
          return b
    finally:
      self.cv.release()
    return None


  def _getAgentForAgentAddr(self, agent_addr):
    try:
      self.cv.acquire()
      key = agent_addr
      for b in self.brokers:
        if key in b.agents:
          return b.agents[key]
    finally:
      self.cv.release()
    return None


  def __repr__(self):
    return "QMF Console Session Manager (brokers: %d)" % len(self.brokers)


  def addBroker(self, target="localhost", timeout=None, mechanisms=None):
    """ Connect to a Qpid broker.  Returns an object of type Broker.
    Will raise an exception if the session is not managing the connection and
    the connection setup to the broker fails.
    """
    url = BrokerURL(target)
    broker = Broker(self, url.host, url.port, mechanisms, url.authName, url.authPass,
                    ssl = url.scheme == URL.AMQPS, connTimeout=timeout)

    self.brokers.append(broker)
    return broker


  def delBroker(self, broker):
    """ Disconnect from a broker, and deallocate the broker proxy object.  The
    'broker' argument is the object returned from the addBroker call.  Errors
    are ignored.
    """
    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()
    return self.schemaCache.getPackages()


  def getClasses(self, packageName):
    """ Get the list of known classes within a QMF package """
    for broker in self.brokers:
      broker._waitForStable()
    return self.schemaCache.getClasses(packageName)


  def getSchema(self, classKey):
    """ Get the schema for a QMF class """
    for broker in self.brokers:
      broker._waitForStable()
    return self.schemaCache.getSchema(classKey)


  def bindPackage(self, packageName):
    """ Filter object and event callbacks to only those elements of the
    specified package. Also filters newPackage and newClass callbacks to the
    given package. Only valid if userBindings is True.
    """
    if not self.userBindings:
      raise Exception("userBindings option must be set for this Session.")
    if not self.rcvObjects and not self.rcvEvents:
      raise Exception("Session needs to be configured to receive events or objects.")
    v1keys = ["console.obj.*.*.%s.#" % packageName, "console.event.*.*.%s.#" % packageName]
    v2keys = ["agent.ind.data.%s.#" % packageName.replace(".", "_"),
              "agent.ind.event.%s.#" % packageName.replace(".", "_"),]
    if (packageName, None) not in self.class_filter:
      self.class_filter.append((packageName, None))
    if (packageName, None) not in self.event_filter:
      self.event_filter.append((packageName, None))
    self.v1BindingKeyList.extend(v1keys)
    self.v2BindingKeyList.extend(v2keys)
    for broker in self.brokers:
      if broker.isConnected():
        for v1key in v1keys:
          broker.amqpSession.exchange_bind(exchange="qpid.management", queue=broker.topicName, binding_key=v1key)
        if broker.brokerSupportsV2:
          for v2key in v2keys:
            # data indications should arrive on the unsolicited indication queue
            broker.amqpSession.exchange_bind(exchange="qmf.default.topic", queue=broker.v2_topic_queue_ui, binding_key=v2key)


  def bindClass(self, pname, cname=None):
    """ Filter object callbacks to only those objects of the specified package
    and optional class. Will also filter newPackage/newClass callbacks to the
    specified package and class.  Only valid if userBindings is True and
    rcvObjects is True.
    """
    if not self.userBindings:
      raise Exception("userBindings option must be set for this Session.")
    if not self.rcvObjects:
      raise Exception("Session needs to be configured with rcvObjects=True.")
    if cname is not None:
      v1key = "console.obj.*.*.%s.%s.#" % (pname, cname)
      v2key = "agent.ind.data.%s.%s.#" % (pname.replace(".", "_"), cname.replace(".", "_"))
    else:
      v1key = "console.obj.*.*.%s.#" % pname
      v2key = "agent.ind.data.%s.#" % pname.replace(".", "_")
    self.v1BindingKeyList.append(v1key)
    self.v2BindingKeyList.append(v2key)
    if (pname, cname) not in self.class_filter:
      self.class_filter.append((pname, cname))
    for broker in self.brokers:
      if broker.isConnected():
        broker.amqpSession.exchange_bind(exchange="qpid.management", queue=broker.topicName, binding_key=v1key)
        if broker.brokerSupportsV2:
          # data indications should arrive on the unsolicited indication queue
          broker.amqpSession.exchange_bind(exchange="qmf.default.topic", queue=broker.v2_topic_queue_ui, binding_key=v2key)

      
  def bindClassKey(self, classKey):
    """ Filter object callbacks to only those objects of the specified
    class. Will also filter newPackage/newClass callbacks to the specified
    package and class.  Only valid if userBindings is True and rcvObjects is
    True.
    """
    pname = classKey.getPackageName()
    cname = classKey.getClassName()
    self.bindClass(pname, cname)

  def bindEvent(self, pname, ename=None):
    """ Filter event callbacks only from a particular class by package and
    event name, or all events in a package if ename=None.  Will also filter
    newPackage/newClass callbacks to the specified package and class. Only
    valid if userBindings is True and rcvEvents is True.
    """
    if not self.userBindings:
      raise Exception("userBindings option must be set for this Session.")
    if not self.rcvEvents:
      raise Exception("Session needs to be configured with rcvEvents=True.")
    if ename is not None:
      v1key = "console.event.*.*.%s.%s.#" % (pname, ename)
      v2key = "agent.ind.event.%s.%s.#" % (pname.replace(".", "_"), ename.replace(".", "_"))
    else:
      v1key = "console.event.*.*.%s.#" % pname
      v2key = "agent.ind.event.%s.#" % pname.replace(".", "_")
    self.v1BindingKeyList.append(v1key)
    self.v2BindingKeyList.append(v2key)
    if (pname, ename) not in self.event_filter:
      self.event_filter.append((pname, ename))
    for broker in self.brokers:
      if broker.isConnected():
        broker.amqpSession.exchange_bind(exchange="qpid.management", queue=broker.topicName, binding_key=v1key)
        if broker.brokerSupportsV2:
          # event indications should arrive on the unsolicited indication queue
          broker.amqpSession.exchange_bind(exchange="qmf.default.topic", queue=broker.v2_topic_queue_ui, binding_key=v2key)

  def bindEventKey(self, eventKey):
    """ Filter event callbacks only from a particular class key. Will also
    filter newPackage/newClass callbacks to the specified package and
    class. Only valid if userBindings is True and rcvEvents is True.
    """
    pname = eventKey.getPackageName()
    ename = eventKey.getClassName()
    self.bindEvent(pname, ename)

  def bindAgent(self, vendor=None, product=None, instance=None, label=None):
    """ Receive heartbeats, newAgent and delAgent callbacks only for those
    agent(s) that match the passed identification criteria:
    V2 agents: vendor, optionally product and instance strings
    V1 agents: the label string.
    Only valid if userBindings is True.
    """
    if not self.userBindings:
      raise Exception("Session not configured for binding specific agents.")
    if vendor is None and label is None:
      raise Exception("Must specify at least a vendor (V2 agents)"
                      " or label (V1 agents).")

    if vendor:   # V2 agent identification
      if product is not None:
        v2key = "agent.ind.heartbeat.%s.%s.#" % (vendor.replace(".", "_"), product.replace(".", "_"))
      else:
        v2key = "agent.ind.heartbeat.%s.#" % vendor.replace(".", "_")
      self.v2BindingKeyList.append(v2key)

      # allow wildcards - only add filter if a non-wildcarded component is given
      if vendor == "*":
        vendor = None
      if product == "*":
        product = None
      if instance == "*":
        instance = None
      if vendor or product or instance:
        if (vendor, product, instance) not in self.agent_filter:
          self.agent_filter.append((vendor, product, instance))

      for broker in self.brokers:
        if broker.isConnected():
          if broker.brokerSupportsV2:
            # heartbeats should arrive on the heartbeat queue
            broker.amqpSession.exchange_bind(exchange="qmf.default.topic",
                                             queue=broker.v2_topic_queue_hb,
                                             binding_key=v2key)
    elif label != "*": # non-wildcard V1 agent label
      # V1 format heartbeats do not have any agent identifier in the routing
      # key, so we cannot filter them by bindings.
      if label not in self.agent_filter:
        self.agent_filter.append(label)


  def getAgents(self, broker=None):
    """ Get a list of currently known agents """
    brokerList = []
    if broker == None:
      for b in self.brokers:
        brokerList.append(b)
    else:
      brokerList.append(broker)

    for b in brokerList:
      b._waitForStable()
    agentList = []
    for b in brokerList:
      for a in b.getAgents():
        agentList.append(a)
    return agentList


  def makeObject(self, classKey, **kwargs):
    """ Create a new, unmanaged object of the schema indicated by classKey """
    schema = self.getSchema(classKey)
    if schema == None:
      raise Exception("Schema not found for classKey")
    return Object(None, schema, None, True, True, kwargs)


  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.
    _class = <name>    - supply a class name as a string.  If the class name exists
                         in multiple packages, a _package argument may also be supplied.
    _objectId = <id>   - get the object referenced by the object-id

    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.

    The default timeout for this synchronous operation is 60 seconds.  To change the timeout,
    use the following argument:

    _timeout = <time in seconds>

    If additional arguments are supplied, they are used as property selectors.  For example,
    if the argument name="test" is supplied, only objects whose "name" property is "test"
    will be returned in the result.
    """
    if "_broker" in kwargs:
      brokerList = []
      brokerList.append(kwargs["_broker"])
    else:
      brokerList = self.brokers
    for broker in brokerList:
      broker._waitForStable()
      if broker.isConnected():
        if "_package" not in kwargs or "_class" not in kwargs or \
              kwargs["_package"] != "org.apache.qpid.broker" or \
              kwargs["_class"] != "agent":
          self.getObjects(_package = "org.apache.qpid.broker", _class = "agent",
                     _agent = broker.getAgent(1,0))

    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")
      if agent.broker.isConnected():
        agentList.append(agent)
    else:
      if "_objectId" in kwargs:
        oid = kwargs["_objectId"]
        for broker in brokerList:
          for agent in broker.getAgents():
            if agent.getBrokerBank() == oid.getBrokerBank() and agent.getAgentBank() == oid.getAgentBank():
              agentList.append(agent)
      else:
        for broker in brokerList:
          for agent in broker.getAgents():
            if agent.broker.isConnected():
              agentList.append(agent)

    if len(agentList) == 0:
      return []

    #
    # We now have a list of agents to query, start the queries and gather the results.
    #
    request = SessionGetRequest(len(agentList))
    for agent in agentList:
      agent.getObjects(request, **kwargs)
    timeout = 60
    if '_timeout' in kwargs:
        timeout = kwargs['_timeout']
    request.wait(timeout)
    return request.result


  def addEventFilter(self, **kwargs):
    """Filter unsolicited events based on package and event name.
    QMF v2 also can filter on vendor, product, and severity values.

    By default, a console receives unsolicted events by binding to:

        qpid.management/console.event.#  (v1)

        qmf.default.topic/agent.ind.event.#  (v2)

    A V1 event filter binding uses the pattern:

        qpid.management/console.event.*.*[.<package>[.<event>]].#

    A V2 event filter binding uses the pattern:

        qmf.default.topic/agent.ind.event.<Vendor|*>.<Product|*>.<severity|*>.<package|*>.<event|*>.#
    """
    package = kwargs.get("package", "*")
    event = kwargs.get("event", "*")
    vendor = kwargs.get("vendor", "*")
    product = kwargs.get("product", "*")
    severity = kwargs.get("severity", "*")

    if package == "*" and event != "*":
      raise Exception("'package' parameter required if 'event' parameter"
                      " supplied")

    # V1 key - can only filter on package (and event)
    if package == "*":
      key = "console.event.*.*." + str(package)
      if event != "*":
        key += "." + str(event)
      key += ".#"

      if key not in self.v1BindingKeyList:
        self.v1BindingKeyList.append(key)
        try:
          # remove default wildcard binding
          self.v1BindingKeyList.remove("console.event.#")
        except:
          pass

    # V2 key - escape any "." in the filter strings

    key = "agent.ind.event." + str(package).replace(".", "_") \
        + "." + str(event).replace(".", "_") \
        + "." + str(severity).replace(".", "_") \
        + "." + str(vendor).replace(".", "_") \
        + "." + str(product).replace(".", "_") \
        + ".#"

    if key not in self.v2BindingKeyList:
      self.v2BindingKeyList.append(key)
      try:
        # remove default wildcard binding
        self.v2BindingKeyList.remove("agent.ind.event.#")
      except:
        pass

    if package != "*":
      if event != "*":
        f = (package, event)
      else:
        f = (package, None)
      if f not in self.event_filter:
        self.event_filter.append(f)


  def addAgentFilter(self, vendor, product=None):
    """ Deprecate - use bindAgent() instead
    """
    self.addHeartbeatFilter(vendor=vendor, product=product)

  def addHeartbeatFilter(self, **kwargs):
    """ Deprecate - use bindAgent() instead.
    """
    vendor = kwargs.get("vendor")
    product = kwargs.get("product")
    if vendor is None:
      raise Exception("vendor parameter required!")

    # V1 heartbeats do not have any agent identifier - we cannot
    # filter them by agent.

    # build the binding key - escape "."s...
    key = "agent.ind.heartbeat." + str(vendor).replace(".", "_")
    if product is not None:
      key += "." + str(product).replace(".", "_")
    key += ".#"

    if key not in self.v2BindingKeyList:
      self.v2BindingKeyList.append(key) 
      self.agent_filter.append((vendor, product, None))

    # be sure we don't ever filter the local broker
    local_broker_key = "agent.ind.heartbeat." + "org.apache".replace(".", "_") \
        + "." + "qpidd".replace(".", "_") + ".#"
    if local_broker_key not in self.v2BindingKeyList:
      self.v2BindingKeyList.append(local_broker_key)

    # remove the wildcard key if present
    try:
      self.v2BindingKeyList.remove("agent.ind.heartbeat.#")
    except:
      pass

  def _bindingKeys(self):
    v1KeyList = []
    v2KeyList = []
    v1KeyList.append("schema.#")
    # note well: any binding that starts with 'agent.ind.heartbeat' will be
    # bound to the heartbeat queue, otherwise it will be bound to the
    # unsolicited indication queue.  See _decOutstanding() for the binding.
    if not self.userBindings:
      if self.rcvObjects and self.rcvEvents and self.rcvHeartbeats:
        v1KeyList.append("console.#")
        v2KeyList.append("agent.ind.data.#")
        v2KeyList.append("agent.ind.event.#")
        v2KeyList.append("agent.ind.heartbeat.#")
      else:
        # need heartbeats for V2 newAgent()/delAgent()
        v2KeyList.append("agent.ind.heartbeat.#")
        if self.rcvObjects:
          v1KeyList.append("console.obj.#")
          v2KeyList.append("agent.ind.data.#")
        else:
          v1KeyList.append("console.obj.*.*.org.apache.qpid.broker.agent")
        if self.rcvEvents:
          v1KeyList.append("console.event.#")
          v2KeyList.append("agent.ind.event.#")
        else:
          v1KeyList.append("console.event.*.*.org.apache.qpid.broker.agent")
        if self.rcvHeartbeats:
          v1KeyList.append("console.heartbeat.#")
    else:
      # mandatory bindings
      v1KeyList.append("console.obj.*.*.org.apache.qpid.broker.agent")
      v1KeyList.append("console.event.*.*.org.apache.qpid.broker.agent")
      v1KeyList.append("console.heartbeat.#")  # no way to turn this on later
      v2KeyList.append("agent.ind.heartbeat.org_apache.qpidd.#")

    return (v1KeyList, v2KeyList)


  def _handleBrokerConnect(self, broker):
    if self.console:
      for agent in broker.getAgents():
        self._newAgentCallback(agent)
      self.console.brokerConnected(broker)


  def _handleBrokerDisconnect(self, broker):
    if self.console:
      for agent in broker.getAgents():
        self._delAgentCallback(agent)
      self.console.brokerDisconnected(broker)


  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()
    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())
    notify = self.schemaCache.declarePackage(pname)
    if notify and self.console != None:
      self._newPackageCallback(pname)

    # Send a class request
    broker._incOutstanding()
    sendCodec = Codec()
    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, agent):
    code = codec.read_uint32()
    text = codec.read_str8()
    context = self.seqMgr._release(seq)
    if context == self._CONTEXT_STARTUP:
      broker._decOutstanding()
    elif context == self._CONTEXT_SYNC and seq == broker.syncSequence:
      try:
        broker.cv.acquire()
        broker.syncInFlight = False
        broker.cv.notify()
      finally:
        broker.cv.release()
    elif context == self._CONTEXT_MULTIGET and seq in self.syncSequenceList:
      try:
        self.cv.acquire()
        self.syncSequenceList.remove(seq)
        if len(self.syncSequenceList) == 0:
          self.cv.notify()
      finally:
        self.cv.release()

    if agent:
      agent._handleV1Completion(seq, code, text)


  def _handleClassInd(self, broker, codec, seq):
    kind  = codec.read_uint8()
    classKey = ClassKey(codec)
    classKey._setType(kind)
    schema = self.schemaCache.getSchema(classKey)

    if not schema:
      # Send a schema request for the unknown class
      broker._incOutstanding()
      sendCodec = Codec()
      seq = self.seqMgr._reserve(self._CONTEXT_STARTUP)
      broker._setHeader(sendCodec, 'S', seq)
      classKey.encode(sendCodec)
      smsg = broker._message(sendCodec.encoded)
      broker._send(smsg)


  def _handleHeartbeatInd(self, broker, codec, seq, msg):
    brokerBank = 1
    agentBank = 0
    dp = msg.get("delivery_properties")
    if dp:
      key = dp["routing_key"]
      if key:
        keyElements = key.split(".")
        if len(keyElements) == 4:
          brokerBank = int(keyElements[2])
          agentBank = int(keyElements[3])
      else:
        # If there's no routing key in the delivery properties,
        # assume the message is from the broker.
        brokerBank = 1
        agentBank = 0

    agent = broker.getAgent(brokerBank, agentBank)
    if self.rcvHeartbeats and self.console != None and agent != None:
      timestamp = codec.read_uint64()
      self._heartbeatCallback(agent, timestamp)


  def _handleSchemaResp(self, broker, codec, seq, agent_addr):
    kind  = codec.read_uint8()
    classKey = ClassKey(codec)
    classKey._setType(kind)
    _class = SchemaClass(kind, classKey, codec, self)
    new_pkg, new_cls = self.schemaCache.declareClass(classKey, _class)
    ctx = self.seqMgr._release(seq)
    if ctx:
      broker._decOutstanding()
    if self.console != None:
      if new_pkg:
        self._newPackageCallback(classKey.getPackageName())
      if new_cls:
        self._newClassCallback(kind, classKey)

    if agent_addr and (agent_addr.__class__ == str or agent_addr.__class__ == unicode):
      agent = self._getAgentForAgentAddr(agent_addr)
      if agent:
        agent._schemaInfoFromV2Agent()


  def _v2HandleHeartbeatInd(self, broker, mp, ah, content):
    try:
      agentName = ah["qmf.agent"]
      values = content["_values"]

      if '_timestamp' in values:
        timestamp = values["_timestamp"]
      else:
        timestamp = values['timestamp']

      if '_heartbeat_interval' in values:
        interval = values['_heartbeat_interval']
      else:
        interval = values['heartbeat_interval']

      epoch = 0
      if '_epoch' in values:
        epoch = values['_epoch']
      elif 'epoch' in values:
        epoch = values['epoch']
    except Exception,e:
      return

    ##
    ## For now, ignore heartbeats from messaging brokers.  We already have the "local-broker"
    ## agent in our list.
    ##
    if '_vendor' in values and values['_vendor'] == 'apache.org' and \
          '_product' in values and values['_product'] == 'qpidd':
      return

    if self.agent_filter:
      # only allow V2 agents that satisfy the filter
      v = agentName.split(":", 2)
      if len(v) != 3 or ((v[0], None, None) not in self.agent_filter
                         and (v[0], v[1], None) not in self.agent_filter
                         and (v[0], v[1], v[2]) not in self.agent_filter):
        return

    agent = broker.getAgent(1, agentName)
    if agent == None:
      agent = Agent(broker, agentName, "QMFv2 Agent", True, interval)
      agent.setEpoch(epoch)
      broker._addAgent(agentName, agent)
    else:
      agent.touch()
    if self.rcvHeartbeats and self.console and agent:
      self._heartbeatCallback(agent, timestamp)
    agent.update_schema_timestamp(values.get("_schema_updated", 0))


  def _v2HandleAgentLocateRsp(self, broker, mp, ah, content):
    self._v2HandleHeartbeatInd(broker, mp, ah, content)


  def _handleError(self, error):
    try:
      self.cv.acquire()
      if len(self.syncSequenceList) > 0:
        self.error = error
      self.syncSequenceList = []
      self.cv.notify()
    finally:
      self.cv.release()


  def _selectMatch(self, object):
    """ Check the object against self.getSelect to check for a match """
    for key, value in self.getSelect:
      for prop, propval in object.getProperties():
        if key == prop.name and value != propval:
          return False
    return True
  

  def _decodeValue(self, codec, typecode, broker=None):
    """ 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 = codec.read_str8()       # SSTR
    elif typecode == 7:  data = codec.read_str16()      # 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() != 0 # 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 == 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
    elif typecode == 15: data = codec.read_map()        # FTABLE
    elif typecode == 20:                                # OBJECT
      # Peek at the type, and if it is still 20 pull it decode. If
      # Not, call back into self.
      inner_type_code = codec.read_uint8()
      if inner_type_code == 20:
          classKey = ClassKey(codec)
          schema = self.schemaCache.getSchema(classKey)
          if not schema:
            return None
          data = Object(self, broker, schema, codec, True, True, False)
      else:
          data = self._decodeValue(codec, inner_type_code, broker)
    elif typecode == 21: data = codec.read_list()       # List
    elif typecode == 22:                                #Array
        #taken from codec10.read_array
        sc = Codec(codec.read_vbin32()) 
        count = sc.read_uint32()
        type = sc.read_uint8()
        data = []
        while count > 0:
          data.append(self._decodeValue(sc,type,broker))
          count -= 1
    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_str16  (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 (float(value))  # DOUBLE
    elif typecode == 14: codec.write_uuid   (value.bytes)   # UUID
    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
    elif typecode == 20: value._encodeUnmanaged(codec)      # OBJECT
    elif typecode == 15: codec.write_map    (value)         # FTABLE
    elif typecode == 21: codec.write_list   (value)         # List
    elif typecode == 22:                                    # Array
        sc = Codec()
        self._encodeValue(sc, len(value), 3)
        if len(value) > 0:
            ltype = self.encoding(value[0])
            self._encodeValue(sc,ltype,1)
            for o in value:
              self._encodeValue(sc, o, ltype)
        codec.write_vbin32(sc.encoded)
    else:
      raise ValueError ("Invalid type code: %d" % typecode)


  def encoding(self, value):
      return self._encoding(value.__class__)
      

  def _encoding(self, klass):
    if Session.ENCODINGS.has_key(klass):
      return self.ENCODINGS[klass]
    for base in klass.__bases__:
      result = self._encoding(base)
      if result != None:
        return result
            

  def _displayValue(self, value, typecode):
    """ """
    if   typecode == 1:  return unicode(value)
    elif typecode == 2:  return unicode(value)
    elif typecode == 3:  return unicode(value)
    elif typecode == 4:  return unicode(value)
    elif typecode == 6:  return value
    elif typecode == 7:  return value
    elif typecode == 8:  return unicode(strftime("%c", gmtime(value / 1000000000)))
    elif typecode == 9:  return unicode(value)
    elif typecode == 10: return unicode(value.__repr__())
    elif typecode == 11:
      if value: return u"T"
      else:     return u"F"
    elif typecode == 12: return unicode(value)
    elif typecode == 13: return unicode(value)
    elif typecode == 14: return unicode(value.__repr__())
    elif typecode == 15: return unicode(value.__repr__())
    elif typecode == 16: return unicode(value)
    elif typecode == 17: return unicode(value)
    elif typecode == 18: return unicode(value)
    elif typecode == 19: return unicode(value)
    elif typecode == 20: return unicode(value.__repr__())
    elif typecode == 21: return unicode(value.__repr__())
    elif typecode == 22: return unicode(value.__repr__())
    else:
      raise ValueError ("Invalid type code: %d" % typecode)
    

  def _defaultValue(self, stype, broker=None, kwargs={}):
    """ """
    typecode = stype.type
    if   typecode == 1:  return 0
    elif typecode == 2:  return 0
    elif typecode == 3:  return 0
    elif typecode == 4:  return 0
    elif typecode == 6:  return ""
    elif typecode == 7:  return ""
    elif typecode == 8:  return 0
    elif typecode == 9:  return 0
    elif typecode == 10: return ObjectId(None)
    elif typecode == 11: return False
    elif typecode == 12: return 0.0
    elif typecode == 13: return 0.0
    elif typecode == 14: return UUID(bytes=[0 for i in range(16)])
    elif typecode == 15: return {}
    elif typecode == 16: return 0
    elif typecode == 17: return 0
    elif typecode == 18: return 0
    elif typecode == 19: return 0
    elif typecode == 21: return []
    elif typecode == 22: return []
    elif typecode == 20:
      try:
        if "classKeys" in kwargs:
          keyList = kwargs["classKeys"]
        else:
          keyList = None
        classKey = self._bestClassKey(stype.refPackage, stype.refClass, keyList)
        if classKey:
          return self.makeObject(classKey, broker, kwargs)
      except:
        pass
      return None
    else:
      raise ValueError ("Invalid type code: %d" % typecode)


  def _bestClassKey(self, pname, cname, preferredList):
    """ """
    if pname == None or cname == None:
      if len(preferredList) == 0:
        return None
      return preferredList[0]
    for p in preferredList:
      if p.getPackageName() == pname and p.getClassName() == cname:
        return p
    clist = self.getClasses(pname)
    for c in clist:
      if c.getClassName() == cname:
        return c
    return None
    

  def _sendMethodRequest(self, broker, schemaKey, objectId, name, argList):
    """ This function can be used to send a method request to an object given only the
    broker, schemaKey, and objectId.  This is an uncommon usage pattern as methods are
    normally invoked on the object itself.
    """
    schema = self.getSchema(schemaKey)
    if not schema:
      raise Exception("Schema not present (Key=%s)" % str(schemaKey))
    for method in schema.getMethods():
      if name == method.name:
        #
        # Count the arguments supplied and validate that the number is what is expected
        # based on the schema.
        #
        count = 0
        for arg in method.arguments:
          if arg.dir.find("I") != -1:
            count += 1
        if count != len(argList):
          raise Exception("Incorrect number of arguments: expected %d, got %d" % (count, len(argList)))

        aIdx = 0
        sendCodec = Codec()
        seq = self.seqMgr._reserve((method, False))

        if objectId.isV2:
          #
          # Compose and send a QMFv2 method request
          #
          call = {}
          call['_object_id'] = objectId.asMap()
          call['_method_name'] = name
          args = {}
          for arg in method.arguments:
            if arg.dir.find("I") != -1:
              args[arg.name] = argList[aIdx]
              aIdx += 1
          call['_arguments'] = args

          dp = broker.amqpSession.delivery_properties()
          dp.routing_key = objectId.getV2RoutingKey()
          mp = broker.amqpSession.message_properties()
          mp.content_type = "amqp/map"
          if broker.saslUser:
            mp.user_id = broker.saslUser
          mp.correlation_id = str(seq)
          mp.app_id = "qmf2"
          mp.reply_to = broker.amqpSession.reply_to("qmf.default.direct", broker.v2_direct_queue)
          mp.application_headers = {'qmf.opcode':'_method_request'}
          sendCodec.write_map(call)
          msg = Message(dp, mp, sendCodec.encoded)
          broker._send(msg, "qmf.default.direct")

        else:
          #
          # Associate this sequence with the agent hosting the object so we can correctly
          # route the method-response
          #
          agent = broker.getAgent(broker.getBrokerBank(), objectId.getAgentBank())
          broker._setSequence(seq, agent)

          #
          # Compose and send a QMFv1 method request
          #
          broker._setHeader(sendCodec, 'M', seq)
          objectId.encode(sendCodec)
          schemaKey.encode(sendCodec)
          sendCodec.write_str8(name)

          for arg in method.arguments:
            if arg.dir.find("I") != -1:
              self._encodeValue(sendCodec, argList[aIdx], arg.type)
              aIdx += 1
          smsg = broker._message(sendCodec.encoded, "agent.%d.%s" %
                                 (objectId.getBrokerBank(), objectId.getAgentBank()))
          broker._send(smsg)
        return seq
    return None

  def _newPackageCallback(self, pname):
    """
    Invokes the console.newPackage() callback if the callback is present and
    the package is not filtered.
    """
    if self.console:
      if len(self.class_filter) == 0 and len(self.event_filter) == 0:
        self.console.newPackage(pname)
      else:
        for x in self.class_filter:
          if x[0] == pname:
            self.console.newPackage(pname)
            return

        for x in self.event_filter:
          if x[0] == pname:
            self.console.newPackage(pname)
            return


  def _newClassCallback(self, ctype, ckey):
    """
    Invokes the console.newClass() callback if the callback is present and the
    class is not filtered.
    """
    if self.console:
      if ctype == ClassKey.TYPE_DATA:
        if (len(self.class_filter) == 0
            or (ckey.getPackageName(), ckey.getClassName()) in self.class_filter):
          self.console.newClass(ctype, ckey)
      elif ctype == ClassKey.TYPE_EVENT:
        if (len(self.event_filter) == 0
            or (ckey.getPackageName(), ckey.getClassName()) in self.event_filter):
          self.console.newClass(ctype, ckey)
      else: # old class keys did not contain type info, check both filters
        if ((len(self.class_filter) == 0 and len(self.event_filter) == 0)
            or (ckey.getPackageName(), ckey.getClassName()) in self.class_filter
            or (ckey.getPackageName(), ckey.getClassName()) in self.event_filter):
          self.console.newClass(ctype, ckey)

  def _agentAllowed(self, agentName, isV2):
    """ True if the agent is NOT filtered.
    """
    if self.agent_filter:
      if isV2:
        v = agentName.split(":", 2)
        return ((len(v) > 2 and (v[0], v[1], v[2]) in self.agent_filter)
                or (len(v) > 1 and (v[0], v[1], None) in self.agent_filter)
                or (v and (v[0], None, None) in self.agent_filter));
      else:
        return agentName in self.agent_filter
    return True

  def _heartbeatCallback(self, agent, timestamp):
    """
    Invokes the console.heartbeat() callback if the callback is present and the
    agent is not filtered.
    """
    if self.console and self.rcvHeartbeats:
      if ((agent.isV2 and self._agentAllowed(agent.agentBank, True))
          or ((not agent.isV2) and self._agentAllowed(agent.label, False))):
        self.console.heartbeat(agent, timestamp)

  def _newAgentCallback(self, agent):
    """
    Invokes the console.newAgent() callback if the callback is present and the
    agent is not filtered.
    """
    if self.console:
      if ((agent.isV2 and self._agentAllowed(agent.agentBank, True))
          or ((not agent.isV2) and self._agentAllowed(agent.label, False))):
        self.console.newAgent(agent)

  def _delAgentCallback(self, agent):
    """
    Invokes the console.delAgent() callback if the callback is present and the
    agent is not filtered.
    """
    if self.console:
      if ((agent.isV2 and self._agentAllowed(agent.agentBank, True))
          or ((not agent.isV2) and self._agentAllowed(agent.label, False))):
        self.console.delAgent(agent)

#===================================================================================================
# SessionGetRequest
#===================================================================================================
class SessionGetRequest(object):
  """
  This class is used to track get-object queries at the Session level.
  """
  def __init__(self, agentCount):
    self.agentCount = agentCount
    self.result = []
    self.cv = Condition()
    self.waiting = True

  def __call__(self, **kwargs):
    """
    Callable entry point for gathering collected objects.
    """
    try:
      self.cv.acquire()
      if 'qmf_object' in kwargs:
        self.result.append(kwargs['qmf_object'])
      elif 'qmf_complete' in kwargs or 'qmf_exception' in kwargs:
        self.agentCount -= 1
        if self.agentCount == 0:
          self.waiting = None
          self.cv.notify()
    finally:
      self.cv.release()

  def wait(self, timeout):
    starttime = time()
    try:
      self.cv.acquire()
      while self.waiting:
        if (time() - starttime) > timeout:
          raise Exception("Timed out after %d seconds" % timeout)
        self.cv.wait(1)
    finally:
      self.cv.release()


#===================================================================================================
# SchemaCache
#===================================================================================================
class SchemaCache(object):
  """
  The SchemaCache is a data structure that stores learned schema information.
  """
  def __init__(self):
    """
    Create a map of schema packages and a lock to protect this data structure.
    Note that this lock is at the bottom of any lock hierarchy.  If it is held, no other
    lock in the system should attempt to be acquired.
    """
    self.packages = {}
    self.lock = Lock()

  def getPackages(self):
    """ Get the list of known QMF packages """
    list = []
    try:
      self.lock.acquire()
      for package in self.packages:
        list.append(package)
    finally:
      self.lock.release()
    return list

  def getClasses(self, packageName):
    """ Get the list of known classes within a QMF package """
    list = []
    try:
      self.lock.acquire()
      if packageName in self.packages:
        for pkey in self.packages[packageName]:
          if isinstance(self.packages[packageName][pkey], SchemaClass):
            list.append(self.packages[packageName][pkey].getKey())
          elif self.packages[packageName][pkey] is not None:
            # schema not present yet, but we have schema type
            list.append(ClassKey({"_package_name": packageName,
                                  "_class_name": pkey[0],
                                  "_hash": pkey[1],
                                  "_type": self.packages[packageName][pkey]}))
    finally:
      self.lock.release()
    return list

  def getSchema(self, classKey):
    """ Get the schema for a QMF class, return None if schema not available """
    pname = classKey.getPackageName()
    pkey = classKey.getPackageKey()
    try:
      self.lock.acquire()
      if pname in self.packages:
        if (pkey in self.packages[pname] and
            isinstance(self.packages[pname][pkey], SchemaClass)):
          # hack: value may be schema type info if schema not available
          return self.packages[pname][pkey]
    finally:
      self.lock.release()
    return None

  def declarePackage(self, pname):
    """ Maybe add a package to the cache.  Return True if package was added, None if it pre-existed. """
    try:
      self.lock.acquire()
      if pname in self.packages:
        return None
      self.packages[pname] = {}
    finally:
      self.lock.release()
    return True

  def declareClass(self, classKey, classDef=None):
    """ Add a class definition to the cache, if supplied.  Return a pair
    indicating if the package or class is new.
    """
    new_package = False
    new_class = False
    pname = classKey.getPackageName()
    pkey = classKey.getPackageKey()
    try:
      self.lock.acquire()
      if pname not in self.packages:
        self.packages[pname] = {}
        new_package = True
      packageMap = self.packages[pname]
      if pkey not in packageMap or not isinstance(packageMap[pkey], SchemaClass):
        if classDef is not None:
          new_class = True
          packageMap[pkey] = classDef
        elif classKey.getType() is not None:
          # hack: don't indicate "new_class" to caller unless the classKey type
          # information is present.  "new_class" causes the console.newClass()
          # callback to be invoked, which -requires- a valid classKey type!
          new_class = True
          # store the type for the getClasses() method:
          packageMap[pkey] = classKey.getType()

    finally:
      self.lock.release()
    return (new_package, new_class)


#===================================================================================================
# ClassKey
#===================================================================================================
class ClassKey:
  """ A ClassKey uniquely identifies a class from the schema. """

  TYPE_DATA = "_data"
  TYPE_EVENT = "_event"

  def __init__(self, constructor):
    if constructor.__class__ == str:
      # construct from __repr__ string
      try:
        # supports two formats:
        # type present = P:C:T(H)
        # no type present = P:C(H)
        tmp = constructor.split(":")
        if len(tmp) == 3:
          self.pname, self.cname, rem = tmp
          self.type, hsh = rem.split("(")
        else:
          self.pname, rem = tmp
          self.cname, hsh = rem.split("(")
          self.type = None
        hsh = hsh.strip(")")
        hexValues = hsh.split("-")
        h0 = int(hexValues[0], 16)
        h1 = int(hexValues[1], 16)
        h2 = int(hexValues[2], 16)
        h3 = int(hexValues[3], 16)
        h4 = int(hexValues[4][0:4], 16)
        h5 = int(hexValues[4][4:12], 16)
        self.hash = UUID(bytes=struct.pack("!LHHHHL", h0, h1, h2, h3, h4, h5))
      except:
        raise Exception("Invalid ClassKey format")
    elif constructor.__class__ == dict:
      # construct from QMFv2 map
      try:
        self.pname = constructor['_package_name']
        self.cname = constructor['_class_name']
        self.hash  = constructor['_hash']
        self.type  = constructor.get('_type')
      except:
        raise Exception("Invalid ClassKey map format %s" % str(constructor))
    else:
      # construct from codec
      codec = constructor
      self.pname = str(codec.read_str8())
      self.cname = str(codec.read_str8())
      self.hash  = UUID(bytes=codec.read_bin128())
      # old V1 codec did not include "type"
      self.type = None

  def encode(self, codec):
    # old V1 codec did not include "type"
    codec.write_str8(self.pname)
    codec.write_str8(self.cname)
    codec.write_bin128(self.hash.bytes)

  def asMap(self):
    m = {'_package_name': self.pname,
            '_class_name': self.cname,
            '_hash': self.hash}
    if self.type is not None:
      m['_type'] = self.type
    return m

  def getPackageName(self):
    return self.pname

  def getClassName(self):
    return self.cname

  def getHash(self):
    return self.hash

  def getType(self):
    return self.type

  def getHashString(self):
    return str(self.hash)

  def getPackageKey(self):
    return (self.cname, self.hash)

  def __repr__(self):
    if self.type is None:
      return self.pname + ":" + self.cname + "(" + self.getHashString() + ")"
    return self.pname + ":" + self.cname + ":" + self.type +  "(" + self.getHashString()  + ")"

  def _setType(self, _type):
    if _type == 2 or _type == ClassKey.TYPE_EVENT:
      self.type = ClassKey.TYPE_EVENT
    else:
      self.type = ClassKey.TYPE_DATA

  def __hash__(self):
    ss = self.pname + self.cname + self.getHashString()
    return ss.__hash__()

  def __eq__(self, other):
    return self.__repr__() == other.__repr__()

#===================================================================================================
# SchemaClass
#===================================================================================================
class SchemaClass:
  """ """
  CLASS_KIND_TABLE = 1
  CLASS_KIND_EVENT = 2

  def __init__(self, kind, key, codec, session):
    self.kind = kind
    self.classKey = key
    self.properties = []
    self.statistics = []
    self.methods = []
    self.arguments = []
    self.session = session

    hasSupertype = 0  #codec.read_uint8()
    if self.kind == self.CLASS_KIND_TABLE:
      propCount   = codec.read_uint16()
      statCount   = codec.read_uint16()
      methodCount = codec.read_uint16()
      if hasSupertype == 1:
        self.superTypeKey = ClassKey(codec)
      else:
        self.superTypeKey = None ;
      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))

    elif self.kind == self.CLASS_KIND_EVENT:
      argCount = codec.read_uint16()
      if (hasSupertype):
        self.superTypeKey = ClassKey(codec)
      else:
        self.superTypeKey = None ;      
      for idx in range(argCount):
        self.arguments.append(SchemaArgument(codec, methodArg=False))

  def __repr__(self):
    if self.kind == self.CLASS_KIND_TABLE:
      kindStr = "Table"
    elif self.kind == self.CLASS_KIND_EVENT:
      kindStr = "Event"
    else:
      kindStr = "Unsupported"
    result = "%s Class: %s " % (kindStr, self.classKey.__repr__())
    return result

  def getKey(self):
    """ Return the class-key for this class. """
    return self.classKey

  def getProperties(self):
    """ Return the list of properties for the class. """
    if (self.superTypeKey == None):
        return self.properties
    else:
        return self.properties + self.session.getSchema(self.superTypeKey).getProperties() 

  def getStatistics(self):
    """ Return the list of statistics for the class. """
    if (self.superTypeKey == None):
        return self.statistics
    else:
        return self.statistics + self.session.getSchema(self.superTypeKey).getStatistics()     

  def getMethods(self):
    """ Return the list of methods for the class. """
    if (self.superTypeKey == None):
        return self.methods
    else:
        return self.methods + self.session.getSchema(self.superTypeKey).getMethods()    

  def getArguments(self):
    """ Return the list of events for the class. """
    """ Return the list of methods for the class. """
    if (self.superTypeKey == None):
        return self.arguments
    else:
        return self.arguments + self.session.getSchema(self.superTypeKey).getArguments()  


#===================================================================================================
# SchemaProperty
#===================================================================================================
class SchemaProperty:
  """ """
  def __init__(self, codec):
    map = codec.read_map()
    self.name     = str(map["name"])
    self.type     = map["type"]
    self.access   = str(map["access"])
    self.index    = map["index"] != 0
    self.optional = map["optional"] != 0
    self.refPackage = None
    self.refClass   = None
    self.unit       = None
    self.min        = None
    self.max        = None
    self.maxlen     = None
    self.desc       = None

    for key, value in map.items():
      if   key == "unit"       : self.unit = value
      elif key == "min"        : self.min = value
      elif key == "max"        : self.max = value
      elif key == "maxlen"     : self.maxlen = value
      elif key == "desc"       : self.desc = value
      elif key == "refPackage" : self.refPackage = value
      elif key == "refClass"   : self.refClass = value

  def __repr__(self):
    return self.name


#===================================================================================================
# SchemaStatistic
#===================================================================================================
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 = value
      elif key == "desc" : self.desc = value

  def __repr__(self):
    return self.name


#===================================================================================================
# SchemaMethod
#===================================================================================================
class SchemaMethod:
  """ """
  def __init__(self, codec):
    map = codec.read_map()
    self.name = str(map["name"])
    argCount  = map["argCount"]
    if "desc" in map:
      self.desc = map["desc"]
    else:
      self.desc = None
    self.arguments = []

    for idx in range(argCount):
      self.arguments.append(SchemaArgument(codec, methodArg=True))

  def __repr__(self):
    result = self.name + "("
    first = True
    for arg in self.arguments:
      if arg.dir.find("I") != -1:
        if first:
          first = False
        else:
          result += ", "
        result += arg.name
    result += ")"
    return result


#===================================================================================================
# SchemaArgument
#===================================================================================================
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
    self.refPackage = None
    self.refClass   = None

    for key, value in map.items():
      if   key == "unit"    : self.unit    = value
      elif key == "min"     : self.min     = value
      elif key == "max"     : self.max     = value
      elif key == "maxlen"  : self.maxlen  = value
      elif key == "desc"    : self.desc    = value
      elif key == "default" : self.default = value
      elif key == "refPackage" : self.refPackage = value
      elif key == "refClass"   : self.refClass = value


#===================================================================================================
# ObjectId
#===================================================================================================
class ObjectId:
  """ Object that represents QMF object identifiers """
  def __init__(self, constructor, first=0, second=0, agentName=None):
    if  constructor.__class__ == dict:
      self.isV2 = True
      self.agentName = agentName
      self.agentEpoch = 0
      if '_agent_name' in constructor:  self.agentName = constructor['_agent_name']
      if '_agent_epoch' in constructor: self.agentEpoch = constructor['_agent_epoch']
      if '_object_name' not in constructor:
        raise Exception("QMFv2 OBJECT_ID must have the '_object_name' field.")
      self.objectName = constructor['_object_name']
    else:
      self.isV2 = None
      if not constructor:
        first = first
        second = second
      else:
        first  = constructor.read_uint64()
        second = constructor.read_uint64()
      self.agentName = str(first & 0x000000000FFFFFFF)
      self.agentEpoch = (first & 0x0FFF000000000000) >> 48
      self.objectName = str(second)

  def _create(cls, agent_name, object_name, epoch=0):
    oid = {"_agent_name": agent_name,
           "_object_name": object_name,
           "_agent_epoch": epoch}
    return cls(oid)
  create = classmethod(_create)

  def __cmp__(self, other):    
    if other == None or not isinstance(other, ObjectId) :
      return 1

    if self.objectName < other.objectName:
      return -1
    if self.objectName > other.objectName:
      return 1

    if self.agentName < other.agentName:
      return -1
    if self.agentName > other.agentName:
      return 1

    if self.agentEpoch < other.agentEpoch:
      return -1
    if self.agentEpoch > other.agentEpoch:
      return 1
    return 0

  def __repr__(self):
    return "%d-%d-%d-%s-%s" % (self.getFlags(), self.getSequence(),
                               self.getBrokerBank(), self.getAgentBank(), self.getObject())

  def index(self):
    return self.__repr__()

  def getFlags(self):
    return 0

  def getSequence(self):
    return self.agentEpoch

  def getBrokerBank(self):
    return 1

  def getAgentBank(self):
    return self.agentName

  def getV2RoutingKey(self):
    if self.agentName == '0':
      return "broker"
    return self.agentName

  def getObject(self):
    return self.objectName

  def isDurable(self):
    return self.getSequence() == 0

  def encode(self, codec):
    first = (self.agentEpoch << 48) + (1 << 28)
    second = 0

    try:
      first += int(self.agentName)
    except:
      pass

    try:
      second = int(self.objectName)
    except:
      pass

    codec.write_uint64(first)
    codec.write_uint64(second)

  def asMap(self):
    omap = {'_agent_name': self.agentName, '_object_name': self.objectName}
    if self.agentEpoch != 0:
      omap['_agent_epoch'] = self.agentEpoch
    return omap

  def __hash__(self):
    return self.__repr__().__hash__()

  def __eq__(self, other):
    return self.__repr__().__eq__(other)


#===================================================================================================
# MethodResult
#===================================================================================================
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)


#===================================================================================================
# Broker
#===================================================================================================
class Broker(Thread):
  """ This object represents a connection (or potential connection) to a QMF broker. """
  SYNC_TIME = 60
  nextSeq = 1

  # for connection recovery
  DELAY_MIN = 1
  DELAY_MAX = 128
  DELAY_FACTOR = 2

  class _q_item:
    """ Broker-private class to encapsulate data sent to the broker thread
    queue.
    """
    type_wakeup = 0
    type_v1msg = 1
    type_v2msg = 2

    def __init__(self, typecode, data):
      self.typecode = typecode
      self.data = data

  def __init__(self, session, host, port, authMechs, authUser, authPass, ssl=False, connTimeout=None):
    """ Create a broker proxy and setup a connection to the broker.  Will raise
    an exception if the connection fails and the session is not configured to
    retry connection setup (manageConnections = False).

    Spawns a thread to manage the broker connection.  Call _shutdown() to
    shutdown the thread when releasing the broker.
    """
    Thread.__init__(self)
    self.session  = session
    self.host = host
    self.port = port
    self.mechanisms = authMechs
    self.ssl = ssl
    if connTimeout is not None:
	connTimeout = float(connTimeout)
    self.connTimeout = connTimeout
    self.authUser = authUser
    self.authPass = authPass
    self.saslUser = None
    self.cv = Condition()
    self.seqToAgentMap = {}
    self.error = None
    self.conn_exc = None  # exception hit by _tryToConnect()
    self.brokerId = None
    self.connected = False
    self.brokerAgent = None
    self.brokerSupportsV2 = None
    self.rcv_queue = Queue() # for msg received on session
    self.conn = None
    self.amqpSession = None
    self.amqpSessionId = "%s.%d.%d" % (platform.uname()[1], os.getpid(), Broker.nextSeq)
    Broker.nextSeq += 1
    self.last_age_check = time()

    # thread control
    self.setDaemon(True)
    self.setName("Thread for broker: %s:%d" % (host, port))
    self.canceled = False
    self.ready = Semaphore(0)
    self.start()
    if not self.session.manageConnections:
      # wait for connection setup to complete in subthread.
      # On failure, propagate exception to caller
      self.ready.acquire()
      if self.conn_exc:
        self._shutdown()   # wait for the subthread to clean up...
        raise self.conn_exc
      # connection up - wait for stable...
      try:
        self._waitForStable()
        agent = self.getBrokerAgent()
        if agent:
          agent.getObjects(_class="agent")
      except:
        self._shutdown()   # wait for the subthread to clean up...
        raise


  def isConnected(self):
    """ Return True if there is an active connection to the broker. """
    return self.connected

  def getError(self):
    """ Return the last error message seen while trying to connect to the broker. """
    return self.error

  def getBrokerId(self):
    """ Get broker's unique identifier (UUID) """
    return self.brokerId

  def getBrokerBank(self):
    """ Return the broker-bank value.  This is the value that the broker assigns to
        objects within its control.  This value appears as a field in the ObjectId
        of objects created by agents controlled by this broker. """
    return 1

  def getAgent(self, brokerBank, agentBank):
    """ Return the agent object associated with a particular broker and agent bank value."""
    bankKey = str(agentBank)
    try:
      self.cv.acquire()
      if bankKey in self.agents:
        return self.agents[bankKey]
    finally:
      self.cv.release()
    return None

  def getBrokerAgent(self):
    return self.brokerAgent

  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 """
    try:
      self.cv.acquire()
      return self.agents.values()
    finally:
      self.cv.release()

  def getAmqpSession(self):
    """ Get the AMQP session object for this connected broker. """
    return self.amqpSession

  def getUrl(self):
    """ """
    return "%s:%d" % (self.host, self.port)

  def getFullUrl(self, noAuthIfGuestDefault=True):
    """ """
    ssl = ""
    if self.ssl:
      ssl = "s"
    auth = "%s/%s@" % (self.authUser, self.authPass)
    if self.authUser == "" or \
          (noAuthIfGuestDefault and self.authUser == "guest" and self.authPass == "guest"):
      auth = ""
    return "amqp%s://%s%s:%d" % (ssl, auth, self.host, self.port or 5672)

  def __repr__(self):
    if self.connected:
      return "Broker connected at: %s" % self.getUrl()
    else:
      return "Disconnected Broker"

  def _setSequence(self, sequence, agent):
    try:
      self.cv.acquire()
      self.seqToAgentMap[sequence] = agent
    finally:
      self.cv.release()

  def _clearSequence(self, sequence):
    try:
      self.cv.acquire()
      self.seqToAgentMap.pop(sequence)
    finally:
      self.cv.release()

  def _tryToConnect(self):
    """ Connect to the broker.  Returns True if connection setup completes
    successfully, otherwise returns False and sets self.error/self.conn_exc
    with error info.  Does not raise exceptions.
    """
    self.error = None
    self.conn_exc = None
    try:
      try:
        self.cv.acquire()
        self.agents = {}
      finally:
        self.cv.release()

      self.topicBound = False
      self.syncInFlight = False
      self.syncRequest = 0
      self.syncResult = None
      self.reqsOutstanding = 1

      try:
        if self.amqpSession:
          self.amqpSession.close()
      except:
        pass
      self.amqpSession = None

      try:
        if self.conn:
          self.conn.close()
      except:
        pass
      self.conn = None

      sock = connect(self.host, self.port)
      sock.settimeout(5)
      oldTimeout = sock.gettimeout()
      sock.settimeout(self.connTimeout)
      if self.ssl:
        connSock = ssl(sock)
      else:
        connSock = sock
      self.conn = Connection(connSock, username=self.authUser, password=self.authPass,
                             mechanism = self.mechanisms, host=self.host, service="qpidd")
      def aborted():
        raise Timeout("Waiting for connection to be established with broker")
      oldAborted = self.conn.aborted
      self.conn.aborted = aborted
      self.conn.start()
      sock.settimeout(oldTimeout)
      self.conn.aborted = oldAborted
      uid = self.conn.user_id
      if uid.__class__ == tuple and len(uid) == 2:
        self.saslUser = uid[1]
      else:
        self.saslUser = self.authUser

      # prevent topic queues from filling up (and causing the agents to
      # disconnect) by discarding the oldest queued messages when full.
      topic_queue_options = {"qpid.policy_type":"ring"}

      self.replyName = "reply-%s" % self.amqpSessionId
      self.amqpSession = self.conn.session(self.amqpSessionId)
      self.amqpSession.timeout = self.SYNC_TIME
      self.amqpSession.auto_sync = True
      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",
                                         accept_mode=self.amqpSession.accept_mode.none,
                                         acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
      self.amqpSession.incoming("rdest").listen(self._v1Cb, self._exceptionCb)
      self.amqpSession.message_set_flow_mode(destination="rdest", flow_mode=self.amqpSession.flow_mode.window)
      self.amqpSession.message_flow(destination="rdest", unit=self.amqpSession.credit_unit.byte, value=0xFFFFFFFFL)
      self.amqpSession.message_flow(destination="rdest", unit=self.amqpSession.credit_unit.message, value=200)

      self.topicName = "topic-%s" % self.amqpSessionId
      self.amqpSession.queue_declare(queue=self.topicName, exclusive=True,
                                     auto_delete=True,
                                     arguments=topic_queue_options)
      self.amqpSession.message_subscribe(queue=self.topicName, destination="tdest",
                                         accept_mode=self.amqpSession.accept_mode.none,
                                         acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
      self.amqpSession.incoming("tdest").listen(self._v1Cb, self._exceptionCb)
      self.amqpSession.message_set_flow_mode(destination="tdest", flow_mode=self.amqpSession.flow_mode.window)
      self.amqpSession.message_flow(destination="tdest", unit=self.amqpSession.credit_unit.byte, value=0xFFFFFFFFL)
      self.amqpSession.message_flow(destination="tdest", unit=self.amqpSession.credit_unit.message, value=200)

      ##
      ## Check to see if the broker has QMFv2 exchanges configured
      ##
      direct_result = self.amqpSession.exchange_query("qmf.default.direct")
      topic_result = self.amqpSession.exchange_query("qmf.default.topic")
      self.brokerSupportsV2 = not (direct_result.not_found or topic_result.not_found)

      try:
        self.cv.acquire()
        self.agents = {}
        self.brokerAgent = Agent(self, 0, "BrokerAgent", isV2=self.brokerSupportsV2)
        self.agents['0'] = self.brokerAgent
      finally:
        self.cv.release()

      ##
      ## Set up connectivity for QMFv2
      ##
      if self.brokerSupportsV2:
        # set up 3 queues:
        # 1 direct queue - for responses destined to this console.
        # 2 topic queues - one for heartbeats (hb), one for unsolicited data
        # and event indications (ui).
        self.v2_direct_queue = "qmfc-v2-%s" % self.amqpSessionId
        self.amqpSession.queue_declare(queue=self.v2_direct_queue, exclusive=True, auto_delete=True)
        self.v2_topic_queue_ui = "qmfc-v2-ui-%s" % self.amqpSessionId
        self.amqpSession.queue_declare(queue=self.v2_topic_queue_ui,
                                       exclusive=True, auto_delete=True,
                                       arguments=topic_queue_options)
        self.v2_topic_queue_hb = "qmfc-v2-hb-%s" % self.amqpSessionId
        self.amqpSession.queue_declare(queue=self.v2_topic_queue_hb,
                                       exclusive=True, auto_delete=True,
                                       arguments=topic_queue_options)

        self.amqpSession.exchange_bind(exchange="qmf.default.direct",
                                       queue=self.v2_direct_queue, binding_key=self.v2_direct_queue)
        ## Other bindings here...

        self.amqpSession.message_subscribe(queue=self.v2_direct_queue, destination="v2dest",
                                           accept_mode=self.amqpSession.accept_mode.none,
                                           acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
        self.amqpSession.incoming("v2dest").listen(self._v2Cb, self._exceptionCb)
        self.amqpSession.message_set_flow_mode(destination="v2dest", flow_mode=self.amqpSession.flow_mode.window)
        self.amqpSession.message_flow(destination="v2dest", unit=self.amqpSession.credit_unit.byte, value=0xFFFFFFFFL)
        self.amqpSession.message_flow(destination="v2dest", unit=self.amqpSession.credit_unit.message, value=50)

        self.amqpSession.message_subscribe(queue=self.v2_topic_queue_ui, destination="v2TopicUI",
                                           accept_mode=self.amqpSession.accept_mode.none,
                                           acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
        self.amqpSession.incoming("v2TopicUI").listen(self._v2Cb, self._exceptionCb)
        self.amqpSession.message_set_flow_mode(destination="v2TopicUI", flow_mode=self.amqpSession.flow_mode.window)
        self.amqpSession.message_flow(destination="v2TopicUI", unit=self.amqpSession.credit_unit.byte, value=0xFFFFFFFFL)
        self.amqpSession.message_flow(destination="v2TopicUI", unit=self.amqpSession.credit_unit.message, value=25)


        self.amqpSession.message_subscribe(queue=self.v2_topic_queue_hb, destination="v2TopicHB",
                                           accept_mode=self.amqpSession.accept_mode.none,
                                           acquire_mode=self.amqpSession.acquire_mode.pre_acquired)
        self.amqpSession.incoming("v2TopicHB").listen(self._v2Cb, self._exceptionCb)
        self.amqpSession.message_set_flow_mode(destination="v2TopicHB", flow_mode=self.amqpSession.flow_mode.window)
        self.amqpSession.message_flow(destination="v2TopicHB", unit=self.amqpSession.credit_unit.byte, value=0xFFFFFFFFL)
        self.amqpSession.message_flow(destination="v2TopicHB", unit=self.amqpSession.credit_unit.message, value=100)

      codec = Codec()
      self._setHeader(codec, 'B')
      msg = self._message(codec.encoded)
      self._send(msg)

      return True  # connection complete

    except Exception, e:
      self.error = "Exception during connection setup: %s - %s" % (e.__class__.__name__, e)
      self.conn_exc = e
      if self.session.console:
        self.session.console.brokerConnectionFailed(self)
      return False     # connection failed

  def _updateAgent(self, obj):
    """
    Just received an object of class "org.apache.qpid.broker:agent", which
    represents a V1 agent.  Add or update the list of agent proxies.
    """
    bankKey = str(obj.agentBank)
    agent = None
    if obj._deleteTime == 0:
      try:
        self.cv.acquire()
        if bankKey not in self.agents:
          # add new agent only if label is not filtered
          if len(self.session.agent_filter) == 0 or obj.label in self.session.agent_filter:
            agent = Agent(self, obj.agentBank, obj.label)
            self.agents[bankKey] = agent
      finally:
        self.cv.release()
      if agent and self.session.console:
        self.session._newAgentCallback(agent)
    else:
      try:
        self.cv.acquire()
        agent = self.agents.pop(bankKey, None)
        if agent:
          agent.close()
      finally:
        self.cv.release()
      if agent and self.session.console:
        self.session._delAgentCallback(agent)

  def _addAgent(self, name, agent):
    try:
      self.cv.acquire()
      self.agents[name] = agent
    finally:
      self.cv.release()
    if self.session.console:
      self.session._newAgentCallback(agent)

  def _ageAgents(self):
    if (time() - self.last_age_check) < self.session.agent_heartbeat_min:
      # don't age if it's too soon
      return
    self.cv.acquire()
    try:
      to_delete = []
      to_notify = []
      for key in self.agents:
        if self.agents[key].isOld():
          to_delete.append(key)
      for key in to_delete:
        agent = self.agents.pop(key)
        agent.close()
        to_notify.append(agent)
      self.last_age_check = time()
    finally:
      self.cv.release()
    if self.session.console:
      for agent in to_notify:
        self.session._delAgentCallback(agent)

  def _v2SendAgentLocate(self, predicate=[]):
    """
    Broadcast an agent-locate request to cause all agents in the domain to tell us who they are.
    """
    # @todo: send locate only to those agents in agent_filter?
    dp = self.amqpSession.delivery_properties()
    dp.routing_key = "console.request.agent_locate"
    mp = self.amqpSession.message_properties()
    mp.content_type = "amqp/list"
    if self.saslUser:
      mp.user_id = self.saslUser
    mp.app_id = "qmf2"
    mp.reply_to = self.amqpSession.reply_to("qmf.default.direct", self.v2_direct_queue)
    mp.application_headers = {'qmf.opcode':'_agent_locate_request'}
    sendCodec = Codec()
    sendCodec.write_list(predicate)
    msg = Message(dp, mp, sendCodec.encoded)
    self._send(msg, "qmf.default.topic")

  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('2'))
    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. """
    try:
      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 != '2':
        return None, None
      opcode = chr(codec.read_uint8())
      seq    = codec.read_uint32()
      return opcode, seq
    except:
      return None, None

  def _message (self, body, routing_key="broker", ttl=None):
    dp = self.amqpSession.delivery_properties()
    dp.routing_key = routing_key
    if ttl:
      dp.ttl = ttl
    mp = self.amqpSession.message_properties()
    mp.content_type = "x-application/qmf"
    if self.saslUser:
      mp.user_id = self.saslUser
    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 _disconnect(self, err_info=None):
    """ Called when the remote broker has disconnected. Re-initializes all
    state associated with the broker.
    """
    # notify any waiters, and callback
    self.cv.acquire()
    try:
      if err_info is not None:
        self.error = err_info
      _agents = self.agents
      self.agents = {}
      for agent in _agents.itervalues():
        agent.close()
      self.syncInFlight = False
      self.reqsOutstanding = 0
      self.cv.notifyAll()
    finally:
      self.cv.release()

    if self.session.console:
      for agent in _agents.itervalues():
        self.session._delAgentCallback(agent)

  def _shutdown(self, _timeout=10):
    """ Disconnect from a broker, and release its resources.   Errors are
    ignored.
    """
    if self.isAlive():
      # kick the thread
      self.canceled = True
      self.rcv_queue.put(Broker._q_item(Broker._q_item.type_wakeup, None))
      self.join(_timeout)

    # abort any pending transactions and delete agents
    self._disconnect("broker shutdown")

    try:
      if self.amqpSession:
        self.amqpSession.close();
    except:
      pass
    self.amqpSession = None
    try:
      if self.conn:
        self.conn.close()
    except:
      pass
    self.conn = None
    self.connected = False

  def _waitForStable(self):
    try:
      self.cv.acquire()
      if not self.connected:
        return
      if self.reqsOutstanding == 0:
        return
      self.syncInFlight = True
      starttime = time()
      while self.reqsOutstanding != 0:
        self.cv.wait(self.SYNC_TIME)
        if time() - starttime > self.SYNC_TIME:
          raise RuntimeError("Timed out waiting for broker to synchronize")
    finally:
      self.cv.release()

  def _incOutstanding(self):
    try:
      self.cv.acquire()
      self.reqsOutstanding += 1
    finally:
      self.cv.release()

  def _decOutstanding(self):
    try:
      self.cv.acquire()
      self.reqsOutstanding -= 1
      if self.reqsOutstanding == 0 and not self.topicBound:
        self.topicBound = True
        for key in self.session.v1BindingKeyList:
          self.amqpSession.exchange_bind(exchange="qpid.management",
                                         queue=self.topicName, binding_key=key)
        if self.brokerSupportsV2:
          # do not drop heartbeat indications when under load from data
          # or event indications.  Put heartbeats on their own dedicated
          # queue.
          #
          for key in self.session.v2BindingKeyList:
            if key.startswith("agent.ind.heartbeat"):
              self.amqpSession.exchange_bind(exchange="qmf.default.topic",
                                             queue=self.v2_topic_queue_hb,
                                             binding_key=key)
            else:
              self.amqpSession.exchange_bind(exchange="qmf.default.topic",
                                             queue=self.v2_topic_queue_ui,
                                             binding_key=key)
          # solicit an agent locate now, after we bind to agent.ind.data,
          # because the agent locate will cause the agent to publish a
          # data indication - and now we're able to receive it!
          self._v2SendAgentLocate()


      if self.reqsOutstanding == 0 and self.syncInFlight:
        self.syncInFlight = False
        self.cv.notify()
    finally:
      self.cv.release()

  def _v1Cb(self, msg):
    """ Callback from session receive thread for V1 messages
    """
    self.rcv_queue.put(Broker._q_item(Broker._q_item.type_v1msg, msg))

  def _v1Dispatch(self, msg):
    try:
      self._v1DispatchProtected(msg)
    except Exception, e:
      print "EXCEPTION in Broker._v1Cb:", e
      import traceback
      traceback.print_exc()

  def _v1DispatchProtected(self, msg):
    """
    This is the general message handler for messages received via the QMFv1 exchanges.
    """
    try:
      agent = None
      agent_addr = None
      mp = msg.get("message_properties")
      ah = mp.application_headers
      if ah and 'qmf.agent' in ah:
        agent_addr = ah['qmf.agent']

      if not agent_addr:
        #
        # See if we can determine the agent identity from the routing key
        #
        dp = msg.get("delivery_properties")
        rkey = None
        if dp and dp.routing_key:
          rkey = dp.routing_key
          items = rkey.split('.')
          if len(items) >= 4:
            if items[0] == 'console' and items[3].isdigit():
              agent_addr = str(items[3])  # The QMFv1 Agent Bank
      if agent_addr != None and agent_addr in self.agents:
        agent = self.agents[agent_addr]

      codec = Codec(msg.body)
      alreadyTried = None
      while True:
        opcode, seq = self._checkHeader(codec)

        if not agent and not alreadyTried:
          alreadyTried = True
          try:
            self.cv.acquire()
            if seq in self.seqToAgentMap:
              agent = self.seqToAgentMap[seq]
          finally:
            self.cv.release()

        if   opcode == None: break
        if   opcode == 'b': self.session._handleBrokerResp      (self, codec, seq)
        elif opcode == 'p': self.session._handlePackageInd      (self, codec, seq)
        elif opcode == 'q': self.session._handleClassInd        (self, codec, seq)
        elif opcode == 's': self.session._handleSchemaResp      (self, codec, seq, agent_addr)
        elif opcode == 'h': self.session._handleHeartbeatInd    (self, codec, seq, msg)
        elif opcode == 'z': self.session._handleCommandComplete (self, codec, seq, agent)
        elif agent:
          agent._handleQmfV1Message(opcode, seq, mp, ah, codec)
          agent.touch() # mark agent as being alive

    finally:  # always ack the message!
      try:
        # ignore failures as the session may be shutting down...
        self.amqpSession.receiver._completed.add(msg.id)
        self.amqpSession.channel.session_completed(self.amqpSession.receiver._completed)
      except:
        pass


  def _v2Cb(self, msg):
    """ Callback from session receive thread for V2 messages
    """
    self.rcv_queue.put(Broker._q_item(Broker._q_item.type_v2msg, msg))

  def _v2Dispatch(self, msg):
    try:
      self._v2DispatchProtected(msg)
    except Exception, e:
      print "EXCEPTION in Broker._v2Cb:", e
      import traceback
      traceback.print_exc()

  def _v2DispatchProtected(self, msg):
    """
    This is the general message handler for messages received via QMFv2 exchanges.
    """
    try:
      mp = msg.get("message_properties")
      ah = mp["application_headers"]
      codec = Codec(msg.body)
  
      if 'qmf.opcode' in ah:
        opcode = ah['qmf.opcode']
        if mp.content_type == "amqp/list":
          try:
            content = codec.read_list()
            if not content:
              content = []
          except:
            # malformed list - ignore
            content = None
        elif mp.content_type == "amqp/map":
          try:
            content = codec.read_map()
            if not content:
              content = {}
          except:
            # malformed map - ignore
            content = None
        else:
          content = None

        if content != None:
          ##
          ## Directly handle agent heartbeats and agent locate responses as these are broker-scope (they are
          ## used to maintain the broker's list of agent proxies.
          ##
          if   opcode == '_agent_heartbeat_indication': self.session._v2HandleHeartbeatInd(self, mp, ah, content)
          elif opcode == '_agent_locate_response':      self.session._v2HandleAgentLocateRsp(self, mp, ah, content)
          else:
            ##
            ## All other opcodes are agent-scope and are forwarded to the agent proxy representing the sender
            ## of the message.
            ##
            agent_addr = ah['qmf.agent']
            if agent_addr == 'broker':
              agent_addr = '0'
            if agent_addr in self.agents:
              agent = self.agents[agent_addr]
              agent._handleQmfV2Message(opcode, mp, ah, content)
              agent.touch()

    finally:  # always ack the message!
      try:
        # ignore failures as the session may be shutting down...
        self.amqpSession.receiver._completed.add(msg.id)
        self.amqpSession.channel.session_completed(self.amqpSession.receiver._completed)
      except:
        pass

  def _exceptionCb(self, data):
    """ Exception notification callback from session receive thread.
    """
    self.cv.acquire()
    try:
      self.connected = False
      self.error = "exception received from messaging layer: %s" % str(data)
    finally:
      self.cv.release()
    self.rcv_queue.put(Broker._q_item(Broker._q_item.type_wakeup, None))

  def run(self):
    """ Main body of the running thread. """

    # First, attempt a connection.  In the unmanaged case,
    # failure to connect needs to cause the Broker()
    # constructor to raise an exception.
    delay = self.DELAY_MIN
    while not self.canceled:
      if self._tryToConnect(): # connection up
        break
      # unmanaged connection - fail & wake up constructor
      if not self.session.manageConnections:
        self.ready.release()
        return
      # managed connection - try again
      count = 0
      while not self.canceled and count < delay:
        sleep(1)
        count += 1
      if delay < self.DELAY_MAX:
        delay *= self.DELAY_FACTOR

    if self.canceled:
      self.ready.release()
      return

    # connection successful!
    self.cv.acquire()
    try:
      self.connected = True
    finally:
      self.cv.release()

    self.session._handleBrokerConnect(self)
    self.ready.release()

    while not self.canceled:

      try:
        item = self.rcv_queue.get(timeout=self.session.agent_heartbeat_min)
      except Empty:
        item = None

      while not self.canceled and item is not None:

        if not self.connected:
          # connection failure
          while item:
            # drain the queue
            try:
              item = self.rcv_queue.get(block=False)
            except Empty:
              item = None
              break

          self._disconnect()  # clean up any pending agents
          self.session._handleError(self.error)
          self.session._handleBrokerDisconnect(self)

          if not self.session.manageConnections:
            return  # do not attempt recovery

          # retry connection setup
          delay = self.DELAY_MIN
          while not self.canceled:
            if self._tryToConnect():
              break
            # managed connection - try again
            count = 0
            while not self.canceled and count < delay:
              sleep(1)
              count += 1
            if delay < self.DELAY_MAX:
              delay *= self.DELAY_FACTOR

          if self.canceled:
            return

          # connection successful!
          self.cv.acquire()
          try:
            self.connected = True
          finally:
            self.cv.release()

          self.session._handleBrokerConnect(self)

        elif item.typecode == Broker._q_item.type_v1msg:
          self._v1Dispatch(item.data)
        elif item.typecode == Broker._q_item.type_v2msg:
          self._v2Dispatch(item.data)

        try:
          item = self.rcv_queue.get(block=False)
        except Empty:
          item = None

      # queue drained, age the agents...
      if not self.canceled:
        self._ageAgents()

#===================================================================================================
# Agent
#===================================================================================================
class Agent:
  """
  This class represents a proxy for a remote agent being managed
  """
  def __init__(self, broker, agentBank, label, isV2=False, interval=0):
    self.broker = broker
    self.session = broker.session
    self.schemaCache = self.session.schemaCache
    self.brokerBank = broker.getBrokerBank()
    self.agentBank = str(agentBank)
    self.label = label
    self.isV2 = isV2
    self.heartbeatInterval = 0
    if interval:
      if interval < self.session.agent_heartbeat_min:
        self.heartbeatInterval = self.session.agent_heartbeat_min
      else:
        self.heartbeatInterval = interval
    self.lock = Lock()
    self.seqMgr = self.session.seqMgr
    self.contextMap = {}
    self.unsolicitedContext = RequestContext(self, self)
    self.lastSeenTime = time()
    self.closed = None
    self.epoch = 0
    self.schema_timestamp = None


  def _checkClosed(self):
    if self.closed:
      raise Exception("Agent is disconnected")


  def __call__(self, **kwargs):
    """
    This is the handler for unsolicited stuff received from the agent
    """
    if 'qmf_object' in kwargs:
      if self.session.console:
        obj = kwargs['qmf_object']
        if len(self.session.class_filter) == 0:
          self.session.console.objectProps(self.broker, obj)
        elif obj.getClassKey():
          # slow path: check classKey against event_filter
          pname = obj.getClassKey().getPackageName()
          cname = obj.getClassKey().getClassName()
          if ((pname, cname) in self.session.class_filter
              or (pname, None) in self.session.class_filter):
            self.session.console.objectProps(self.broker, obj)
    elif 'qmf_object_stats' in kwargs:
      if self.session.console:
        obj = kwargs['qmf_object_stats']
        if len(self.session.class_filter) == 0:
          self.session.console.objectStats(self.broker, obj)
        elif obj.getClassKey():
          # slow path: check classKey against event_filter
          pname = obj.getClassKey().getPackageName()
          cname = obj.getClassKey().getClassName()
          if ((pname, cname) in self.session.class_filter
              or (pname, None) in self.session.class_filter):
            self.session.console.objectStats(self.broker, obj)
    elif 'qmf_event' in kwargs:
      if self.session.console:
        event = kwargs['qmf_event']
        if len(self.session.event_filter) == 0:
          self.session.console.event(self.broker, event)
        elif event.classKey:
          # slow path: check classKey against event_filter
          pname = event.classKey.getPackageName()
          ename = event.classKey.getClassName()
          if ((pname, ename) in self.session.event_filter
              or (pname, None) in self.session.event_filter):
            self.session.console.event(self.broker, event)
    elif 'qmf_schema_id' in kwargs:
      ckey = kwargs['qmf_schema_id']
      new_pkg, new_cls = self.session.schemaCache.declareClass(ckey)
      if self.session.console:
        if new_pkg:
          self.session._newPackageCallback(ckey.getPackageName())
        if new_cls:
          # translate V2's string based type value to legacy
          # integer value for backward compatibility
          cls_type = ckey.getType()
          if str(cls_type) == ckey.TYPE_DATA:
            cls_type = 1
          elif str(cls_type) == ckey.TYPE_EVENT:
            cls_type = 2
          self.session._newClassCallback(cls_type, ckey)

  def touch(self):
    if self.heartbeatInterval:
      self.lastSeenTime = time()


  def setEpoch(self, epoch):
    self.epoch = epoch

  def update_schema_timestamp(self, timestamp):
    """ Check the latest schema timestamp from the agent V2 heartbeat.  Issue a
    query for all packages & classes should the timestamp change.
    """
    self.lock.acquire()
    try:
      if self.schema_timestamp == timestamp:
        return
      self.schema_timestamp = timestamp

      context = RequestContext(self, self)
      sequence = self.seqMgr._reserve(context)

      self.contextMap[sequence] = context
      context.setSequence(sequence)

    finally:
      self.lock.release()

    self._v2SendSchemaIdQuery(sequence, {})


  def epochMismatch(self, epoch):
    if epoch == 0 or self.epoch == 0:
      return None
    if epoch == self.epoch:
      return None
    return True


  def isOld(self):
    if self.heartbeatInterval == 0:
      return None
    if time() - self.lastSeenTime > (self.session.agent_heartbeat_miss * self.heartbeatInterval):
      return True
    return None


  def close(self):
    self.closed = True
    copy = {}
    try:
      self.lock.acquire()
      for seq in self.contextMap:
        copy[seq] = self.contextMap[seq]
    finally:
      self.lock.release()

    for seq in copy:
      context = copy[seq]
      context.cancel("Agent disconnected")
      self.seqMgr._release(seq)


  def __repr__(self):
    if self.isV2:
      ver = "v2"
    else:
      ver = "v1"
    return "Agent(%s) at bank %d.%s (%s)" % (ver, self.brokerBank, self.agentBank, self.label)


  def getBroker(self):
    return self.broker


  def getBrokerBank(self):
    return self.brokerBank


  def getAgentBank(self):
    return self.agentBank


  def getV2RoutingKey(self):
    if self.agentBank == '0':
      return 'broker'
    return self.agentBank


  def getObjects(self, notifiable=None, **kwargs):
    """ Get a list of objects from QMF agents.
    All arguments are passed by name(keyword).

    If 'notifiable' is None (default), this call will block until completion or timeout.
    If supplied, notifiable is assumed to be a callable object that will be called when the
    list of queried objects arrives.  The single argument to the call shall be a list of
    the returned objects.

    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.
    _class = <name>    - supply a class name as a string.  If the class name exists
                         in multiple packages, a _package argument may also be supplied.
    _objectId = <id>   - get the object referenced by the object-id

    The default timeout for this synchronous operation is 60 seconds.  To change the timeout,
    use the following argument:

    _timeout = <time in seconds>

    If additional arguments are supplied, they are used as property selectors.  For example,
    if the argument name="test" is supplied, only objects whose "name" property is "test"
    will be returned in the result.
    """
    self._checkClosed()
    if notifiable:
      if not callable(notifiable):
        raise Exception("notifiable object must be callable")

    #
    # Isolate the selectors from the kwargs
    #
    selectors = {}
    for key in kwargs:
      value = kwargs[key]
      if key[0] != '_':
        selectors[key] = value

    #
    # Allocate a context to track this asynchronous request.
    #
    context = RequestContext(self, notifiable, selectors)
    sequence = self.seqMgr._reserve(context)
    try:
      self.lock.acquire()
      self.contextMap[sequence] = context
      context.setSequence(sequence)
    finally:
      self.lock.release()

    #
    # Compose and send the query message to the agent using the appropriate protocol for the
    # agent's QMF version.
    #
    if self.isV2:
      self._v2SendGetQuery(sequence, kwargs)
    else:
      self.broker._setSequence(sequence, self)
      self._v1SendGetQuery(sequence, kwargs)

    #
    # If this is a synchronous call, block and wait for completion.
    #
    if not notifiable:
      timeout = 60
      if '_timeout' in kwargs:
        timeout = kwargs['_timeout']
      context.waitForSignal(timeout)
      if context.exception:
        raise Exception(context.exception)
      result = context.queryResults
      return result


  def _clearContext(self, sequence):
    try:
      self.lock.acquire()
      try:
        self.contextMap.pop(sequence)
      except KeyError:
        pass   # @todo - shouldn't happen, log a warning.
    finally:
      self.lock.release()


  def _schemaInfoFromV2Agent(self):
    """
    We have just received new schema information from this agent.  Check to see if there's
    more work that can now be done.
    """
    try:
      self.lock.acquire()
      copy_of_map = {}
      for item in self.contextMap:
        copy_of_map[item] = self.contextMap[item]
    finally:
      self.lock.release()

    self.unsolicitedContext.reprocess()
    for context in copy_of_map:
      copy_of_map[context].reprocess()


  def _handleV1Completion(self, sequence, code, text):
    """
    Called if one of this agent's V1 commands completed
    """
    context = None
    try:
      self.lock.acquire()
      if sequence in self.contextMap:
        context = self.contextMap[sequence]
    finally:
      self.lock.release()

    if context:
      if code != 0:
        ex = "Error %d: %s" % (code, text)
        context.setException(ex)
      context.signal()
    self.broker._clearSequence(sequence)


  def _v1HandleMethodResp(self, codec, seq):
    """
    Handle a QMFv1 method response
    """
    code = codec.read_uint32()
    text = codec.read_str16()
    outArgs = {}
    self.broker._clearSequence(seq)
    pair = self.seqMgr._release(seq)
    if pair == None:
      return
    method, synchronous = pair
    if code == 0:
      for arg in method.arguments:
        if arg.dir.find("O") != -1:
          outArgs[arg.name] = self.session._decodeValue(codec, arg.type, self.broker)
    result = MethodResult(code, text, outArgs)
    if synchronous:
      try:
        self.broker.cv.acquire()
        self.broker.syncResult = result
        self.broker.syncInFlight = False
        self.broker.cv.notify()
      finally:
        self.broker.cv.release()
    else:
      if self.session.console:
        self.session.console.methodResponse(self.broker, seq, result)


  def _v1HandleEventInd(self, codec, seq):
    """
    Handle a QMFv1 event indication
    """
    event = Event(self, codec)
    self.unsolicitedContext.doEvent(event)


  def _v1HandleContentInd(self, codec, sequence, prop=False, stat=False):
    """
    Handle a QMFv1 content indication
    """
    classKey = ClassKey(codec)
    schema = self.schemaCache.getSchema(classKey)
    if not schema:
      return

    obj = Object(self, schema, codec, prop, stat)
    if classKey.getPackageName() == "org.apache.qpid.broker" and classKey.getClassName() == "agent" and prop:
      self.broker._updateAgent(obj)

    context = self.unsolicitedContext
    try:
      self.lock.acquire()
      if sequence in self.contextMap:
        context = self.contextMap[sequence]
    finally:
      self.lock.release()

    context.addV1QueryResult(obj, prop, stat)


  def _v2HandleDataInd(self, mp, ah, content):
    """
    Handle a QMFv2 data indication from the agent.  Note: called from context
    of the Broker thread.
    """
    if content.__class__ != list:
      return

    if mp.correlation_id:
      try:
        self.lock.acquire()
        sequence = int(mp.correlation_id)
        if sequence not in self.contextMap:
          return
        context = self.contextMap[sequence]
      finally:
        self.lock.release()
    else:
      context = self.unsolicitedContext

    kind = "_data"
    if "qmf.content" in ah:
      kind = ah["qmf.content"]
    if kind == "_data":
      for omap in content:
        context.addV2QueryResult(omap)
      context.processV2Data()
      if 'partial' not in ah:
        context.signal()

    elif kind == "_event":
      for omap in content:
        event = Event(self, v2Map=omap)
        if event.classKey is None or event.schema:
          # schema optional or present
          context.doEvent(event)
        else:
          # schema not optional and not present
          if context.addPendingEvent(event):
            self._v2SendSchemaRequest(event.classKey)

    elif kind == "_schema_id":
      for sid in content:
        try:
          ckey = ClassKey(sid)
        except:
          # @todo: log error
          ckey = None
        if ckey is not None:
          # @todo: for now, the application cannot directly send a query for
          # _schema_id.  This request _must_ have been initiated by the framework
          # in order to update the schema cache.
          context.notifiable(qmf_schema_id=ckey)


  def _v2HandleMethodResp(self, mp, ah, content):
    """
    Handle a QMFv2 method response from the agent
    """
    context = None
    sequence = None
    if mp.correlation_id:
      try:
        self.lock.acquire()
        seq = int(mp.correlation_id)
      finally:
        self.lock.release()
    else:
      return

    pair = self.seqMgr._release(seq)
    if pair == None:
      return
    method, synchronous = pair

    result = MethodResult(0, 'OK', content['_arguments'])
    if synchronous:
      try:
        self.broker.cv.acquire()
        self.broker.syncResult = result
        self.broker.syncInFlight = False
        self.broker.cv.notify()
      finally:
        self.broker.cv.release()
    else:
      if self.session.console:
        self.session.console.methodResponse(self.broker, seq, result)

  def _v2HandleException(self, mp, ah, content):
    """
    Handle a QMFv2 exception
    """
    context = None
    if mp.correlation_id:
      try:
        self.lock.acquire()
        seq = int(mp.correlation_id)
      finally:
        self.lock.release()
    else:
      return

    values = {}
    if '_values' in content:
      values = content['_values']

    code = 7
    text = "error"
    if 'error_code' in values:
      code = values['error_code']
    if 'error_text' in values:
      text = values['error_text']

    pair = self.seqMgr._release(seq)
    if pair == None:
      return

    if pair.__class__ == RequestContext:
      pair.cancel(text)
      return

    method, synchronous = pair

    result = MethodResult(code, text, {})
    if synchronous:
      try:
        self.broker.cv.acquire()
        self.broker.syncResult = result
        self.broker.syncInFlight = False
        self.broker.cv.notify()
      finally:
        self.broker.cv.release()
    else:
      if self.session.console:
        self.session.console.methodResponse(self.broker, seq, result)


  def _v1SendGetQuery(self, sequence, kwargs):
    """
    Send a get query to a QMFv1 agent.
    """
    #
    # Build the query map
    #
    query = {}
    if '_class' in kwargs:
      query['_class'] = kwargs['_class']
      if '_package' in kwargs:
        query['_package'] = kwargs['_package']
    elif '_key' in kwargs:
      key = kwargs['_key']
      query['_class'] = key.getClassName()
      query['_package'] = key.getPackageName()
    elif '_objectId' in kwargs:
      query['_objectid'] = kwargs['_objectId'].__repr__()

    #
    # Construct and transmit the message
    #
    sendCodec = Codec()
    self.broker._setHeader(sendCodec, 'G', sequence)
    sendCodec.write_map(query)
    smsg = self.broker._message(sendCodec.encoded, "agent.%d.%s" % (self.brokerBank, self.agentBank))
    self.broker._send(smsg)


  def _v2SendQuery(self, query, sequence):
    """
    Given a query map, construct and send a V2 Query message.
    """
    dp = self.broker.amqpSession.delivery_properties()
    dp.routing_key = self.getV2RoutingKey()
    mp = self.broker.amqpSession.message_properties()
    mp.content_type = "amqp/map"
    if self.broker.saslUser:
      mp.user_id = self.broker.saslUser
    mp.correlation_id = str(sequence)
    mp.app_id = "qmf2"
    mp.reply_to = self.broker.amqpSession.reply_to("qmf.default.direct", self.broker.v2_direct_queue)
    mp.application_headers = {'qmf.opcode':'_query_request'}
    sendCodec = Codec()
    sendCodec.write_map(query)
    msg = Message(dp, mp, sendCodec.encoded)
    self.broker._send(msg, "qmf.default.direct")


  def _v2SendGetQuery(self, sequence, kwargs):
    """
    Send a get query to a QMFv2 agent.
    """
    #
    # Build the query map
    #
    query = {'_what': 'OBJECT'}
    if '_class' in kwargs:
      schemaMap = {'_class_name': kwargs['_class']}
      if '_package' in kwargs:
        schemaMap['_package_name'] = kwargs['_package']
      query['_schema_id'] = schemaMap
    elif '_key' in kwargs:
      query['_schema_id'] = kwargs['_key'].asMap()
    elif '_objectId' in kwargs:
      query['_object_id'] = kwargs['_objectId'].asMap()

    self._v2SendQuery(query, sequence)


  def _v2SendSchemaIdQuery(self, sequence, kwargs):
    """
    Send a query for all schema ids to a QMFv2 agent.
    """
    #
    # Build the query map
    #
    query = {'_what': 'SCHEMA_ID'}
    # @todo - predicate support. For now, return all known schema ids.

    self._v2SendQuery(query, sequence)


  def _v2SendSchemaRequest(self, schemaId):
    """
    Send a query to an agent to request details on a particular schema class.
    IMPORTANT:  This function currently sends a QMFv1 schema-request to the address of
                the agent.  The agent will send its response to amq.direct/<our-key>.
                Eventually, this will be converted to a proper QMFv2 schema query.
    """
    sendCodec = Codec()
    seq = self.seqMgr._reserve(None)
    self.broker._setHeader(sendCodec, 'S', seq)
    schemaId.encode(sendCodec)
    smsg = self.broker._message(sendCodec.encoded, self.agentBank)
    self.broker._send(smsg, "qmf.default.direct")


  def _handleQmfV1Message(self, opcode, seq, mp, ah, codec):
    """
    Process QMFv1 messages arriving from an agent.  Note well: this method is
    called from the context of the Broker thread.
    """
    if   opcode == 'm': self._v1HandleMethodResp(codec, seq)
    elif opcode == 'e': self._v1HandleEventInd(codec, seq)
    elif opcode == 'c': self._v1HandleContentInd(codec, seq, prop=True)
    elif opcode == 'i': self._v1HandleContentInd(codec, seq, stat=True)
    elif opcode == 'g': self._v1HandleContentInd(codec, seq, prop=True, stat=True)


  def _handleQmfV2Message(self, opcode, mp, ah, content):
    """
    Process QMFv2 messages arriving from an agent.  Note well: this method is
    called from the context of the Broker thread.
    """
    if   opcode == '_data_indication': self._v2HandleDataInd(mp, ah, content)
    elif opcode == '_query_response':  self._v2HandleDataInd(mp, ah, content)
    elif opcode == '_method_response': self._v2HandleMethodResp(mp, ah, content)
    elif opcode == '_exception':       self._v2HandleException(mp, ah, content)


#===================================================================================================
# RequestContext
#===================================================================================================
class RequestContext(object):
  """
  This class tracks an asynchronous request sent to an agent.
  TODO: Add logic for client-side selection and filtering deleted objects from get-queries
  """
  def __init__(self, agent, notifiable, selectors={}):
    self.sequence = None
    self.agent = agent
    self.schemaCache = self.agent.schemaCache
    self.notifiable = notifiable
    self.selectors = selectors
    self.startTime = time()
    self.rawQueryResults = []
    self.queryResults = []
    self.pendingEvents = {}
    self.exception = None
    self.waitingForSchema = None
    self.pendingSignal = None
    self.cv = Condition()
    self.blocked = notifiable == None


  def setSequence(self, sequence):
    self.sequence = sequence


  def addV1QueryResult(self, data, has_props, has_stats):
    values = {}
    if has_props:
      for prop, val in data.getProperties():
        values[prop.name] = val
    if has_stats:
      for stat, val in data.getStatistics():
        values[stat.name] = val
    for key in values:
      val = values[key]
      if key in self.selectors and val != self.selectors[key]:
        return

    if self.notifiable:
      if has_props:
        self.notifiable(qmf_object=data)
      if has_stats:
        self.notifiable(qmf_object_stats=data)
    else:
      self.queryResults.append(data)


  def addV2QueryResult(self, data):
    values = data['_values']
    for key in values:
      val = values[key]
      if key in self.selectors:
        sel_val = self.selectors[key]
        if sel_val.__class__ == ObjectId:
          val = ObjectId(val, agentName=self.agent.getAgentBank())
        if val != sel_val:
          return
    self.rawQueryResults.append(data)

  def addPendingEvent(self, event):
    """ Stores a received event that is pending a schema.  Returns True if this
    event is the first instance of a given schema identifier.
    """
    self.cv.acquire()
    try:
      if event.classKey in self.pendingEvents:
        self.pendingEvents[event.classKey].append((event, time()))
        return False
      self.pendingEvents[event.classKey] = [(event, time())]
      return True
    finally:
      self.cv.release()

  def processPendingEvents(self):
    """ Walk the pending events looking for schemas that are now
    available. Remove any events that now have schema, and process them.
    """
    keysToDelete = []
    events = []
    self.cv.acquire()
    try:
      for key in self.pendingEvents.iterkeys():
        schema = self.schemaCache.getSchema(key)
        if schema:
          keysToDelete.append(key)
          for item in self.pendingEvents[key]:
            # item is (timestamp, event-obj) tuple.
            # hack: I have no idea what a valid lifetime for an event
            # should be. 60 seconds???
            if (time() - item[1]) < 60:
              item[0].schema = schema
              events.append(item[0])
      for key in keysToDelete:
        self.pendingEvents.pop(key)
    finally:
      self.cv.release()
    for event in events:
      self.doEvent(event)

  def doEvent(self, data):
    if self.notifiable:
      self.notifiable(qmf_event=data)


  def setException(self, ex):
    self.exception = ex


  def getAge(self):
    return time() - self.startTime


  def cancel(self, exception):
    self.setException(exception)
    try:
      self.cv.acquire()
      self.blocked = None
      self.waitingForSchema = None
      self.cv.notify()
    finally:
      self.cv.release()
    self._complete()


  def waitForSignal(self, timeout):
    try:
      self.cv.acquire()
      while self.blocked:
        if (time() - self.startTime) > timeout:
          self.exception = "Request timed out after %d seconds" % timeout
          return
        self.cv.wait(1)
    finally:
      self.cv.release()


  def signal(self):
    try:
      self.cv.acquire()
      if self.waitingForSchema:
        self.pendingSignal = True
        return
      else:
        self.blocked = None
        self.cv.notify()
    finally:
      self.cv.release()
    self._complete()


  def _complete(self):
    if self.notifiable:
      if self.exception:
        self.notifiable(qmf_exception=self.exception)
      else:
        self.notifiable(qmf_complete=True)

    if self.sequence:
      self.agent._clearContext(self.sequence)


  def processV2Data(self):
    """
    Attempt to make progress on the entries in the raw_query_results queue.  If an entry has a schema
    that is in our schema cache, process it.  Otherwise, send a request for the schema information
    to the agent that manages the object.
    """
    schemaId = None
    queryResults = []
    try:
      self.cv.acquire()
      if self.waitingForSchema:
        return
      while (not self.waitingForSchema) and len(self.rawQueryResults) > 0:
        head = self.rawQueryResults[0]
        schemaId = self._getSchemaIdforV2ObjectLH(head)
        schema = self.schemaCache.getSchema(schemaId)
        if schema:
          obj = Object(self.agent, schema, v2Map=head, agentName=self.agent.agentBank)
          queryResults.append(obj)
          self.rawQueryResults.pop(0)
        else:
          self.waitingForSchema = True
    finally:
      self.cv.release()

    if self.waitingForSchema:
      self.agent._v2SendSchemaRequest(schemaId)

    for result in queryResults:
      key = result.getClassKey()
      if key.getPackageName() == "org.apache.qpid.broker" and key.getClassName() == "agent":
        self.agent.broker._updateAgent(result)
      if self.notifiable:
        self.notifiable(qmf_object=result)
      else:
        self.queryResults.append(result)

    complete = None
    try:
      self.cv.acquire()
      if not self.waitingForSchema and self.pendingSignal:
        self.blocked = None
        self.cv.notify()
        complete = True
    finally:
      self.cv.release()

    if complete:
      self._complete()


  def reprocess(self):
    """
    New schema information has been added to the schema-cache.  Clear our 'waiting' status
    and see if we can make more progress on any pending inbound events/objects.
    """
    try:
      self.cv.acquire()
      self.waitingForSchema = None
    finally:
      self.cv.release()
    self.processV2Data()
    self.processPendingEvents()

  def _getSchemaIdforV2ObjectLH(self, data):
    """
    Given a data map, extract the schema-identifier.
    """
    if data.__class__ != dict:
      return None
    if '_schema_id' in data:
      return ClassKey(data['_schema_id'])
    return None


#===================================================================================================
# Event
#===================================================================================================
class Event:
  """ """
  def __init__(self, agent, codec=None, v2Map=None):
    self.agent = agent
    self.session = agent.session
    self.broker  = agent.broker

    if isinstance(v2Map,dict):
      self.classKey = None
      self.schema = None
      try:
        self.arguments = v2Map["_values"]
        self.timestamp = long(v2Map["_timestamp"])
        self.severity = v2Map["_severity"]
        if "_schema_id" in v2Map:
          self.classKey = ClassKey(v2Map["_schema_id"])
          self.classKey._setType(ClassKey.TYPE_EVENT)
      except:
        raise Exception("Invalid event object: %s " % str(v2Map))
      if self.classKey is not None:
        self.schema = self.session.schemaCache.getSchema(self.classKey)

    elif codec is not None:
      self.classKey = ClassKey(codec)
      self.classKey._setType(ClassKey.TYPE_EVENT)
      self.timestamp = codec.read_int64()
      self.severity = codec.read_uint8()
      self.arguments = {}
      self.schema = self.session.schemaCache.getSchema(self.classKey)
      if not self.schema:
        return
      for arg in self.schema.arguments:
        self.arguments[arg.name] = self.session._decodeValue(codec, arg.type,
                                                             self.broker)
    else:
      raise Exception("No constructor for event object.")


  def __repr__(self):
    if self.schema == None:
      return "<uninterpretable>"
    out = strftime("%c", gmtime(self.timestamp / 1000000000))
    out += " " + self._sevName() + " " + self.classKey.getPackageName() + ":" + self.classKey.getClassName()
    out += " broker=" + self.broker.getUrl()
    for arg in self.schema.arguments:
      disp = self.session._displayValue(self.arguments[arg.name], arg.type).encode("utf8")
      if " " in disp:
        disp = "\"" + disp + "\""
      out += " " + arg.name + "=" + disp
    return out

  def _sevName(self):
    if self.severity == 0 : return "EMER "
    if self.severity == 1 : return "ALERT"
    if self.severity == 2 : return "CRIT "
    if self.severity == 3 : return "ERROR"
    if self.severity == 4 : return "WARN "
    if self.severity == 5 : return "NOTIC"
    if self.severity == 6 : return "INFO "
    if self.severity == 7 : return "DEBUG"
    return "INV-%d" % self.severity

  def getClassKey(self):
    return self.classKey

  def getArguments(self):
    return self.arguments

  def getTimestamp(self):
    return self.timestamp

  def getSchema(self):
    return self.schema


#===================================================================================================
# SequenceManager
#===================================================================================================
class SequenceManager:
  """ Manage sequence numbers for asynchronous method calls """
  def __init__(self):
    self.lock     = Lock()
    self.sequence = long(time())  # pseudo-randomize the start
    self.pending  = {}

  def _reserve(self, data):
    """ Reserve a unique sequence number """
    try:
      self.lock.acquire()
      result = self.sequence
      self.sequence = self.sequence + 1
      self.pending[result] = data
    finally:
      self.lock.release()
    return result

  def _release(self, seq):
    """ Release a reserved sequence number """
    data = None
    try:
      self.lock.acquire()
      if seq in self.pending:
        data = self.pending[seq]
        del self.pending[seq]
    finally:
      self.lock.release()
    return data


#===================================================================================================
# DebugConsole
#===================================================================================================
class DebugConsole(Console):
  """ """
  def brokerConnected(self, broker):
    print "brokerConnected:", broker

  def brokerConnectionFailed(self, broker):
    print "brokerConnectionFailed:", broker

  def brokerDisconnected(self, broker):
    print "brokerDisconnected:", broker

  def newPackage(self, name):
    print "newPackage:", name

  def newClass(self, kind, classKey):
    print "newClass:", kind, classKey

  def newAgent(self, agent):
    print "newAgent:", agent

  def delAgent(self, agent):
    print "delAgent:", agent

  def objectProps(self, broker, record):
    print "objectProps:", record

  def objectStats(self, broker, record):
    print "objectStats:", record

  def event(self, broker, event):
    print "event:", event

  def heartbeat(self, agent, timestamp):
    print "heartbeat:", agent

  def brokerInfo(self, broker):
    print "brokerInfo:", broker