path: root/qpid/python/qmf/qmfConsole.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.
#
import sys
import os
import logging
import platform
import time
import datetime
import Queue
from threading import Thread
from threading import Lock
from threading import currentThread
from threading import Condition

from qpid.messaging import *

from qmfCommon import (makeSubject, parseSubject, OpCode, QmfQuery, Notifier,
                       QmfQueryPredicate, MsgKey, QmfData, QmfAddress,
                       AMQP_QMF_AGENT_LOCATE, AMQP_QMF_AGENT_INDICATION)  



# global flag that indicates which thread (if any) is
# running the console callback 
_callback_thread=None




##==============================================================================
## Sequence Manager  
##==============================================================================

class _Mailbox(object):
    """
    Virtual base class for all Mailbox-like objects
    """
    def __init__(self):
        self._msgs = []
        self._cv = Condition()
        self._waiting = False

    def deliver(self, obj):
        self._cv.acquire()
        try:
            self._msgs.append(obj)
            # if was empty, notify waiters
            if len(self._msgs) == 1:
                logging.error("Delivering @ %s" % time.time())
                self._cv.notify()
        finally:
            self._cv.release()

    def fetch(self, timeout=None):
        self._cv.acquire()
        try:
            if len(self._msgs) == 0:
                self._cv.wait(timeout)
            if len(self._msgs):
                return self._msgs.pop()
            return None
        finally:
            self._cv.release()



class SequencedWaiter(object):
    """ 
    Manage sequence numbers for asynchronous method calls. 
    Allows the caller to associate a generic piece of data with a unique sequence
    number."""

    def __init__(self):
        self.lock     = Lock()
        self.sequence = 1L
        self.pending  = {}


    def allocate(self):
        """ 
        Reserve a sequence number.
        
        @rtype: long
        @return: a unique nonzero sequence number.
        """
        self.lock.acquire()
        try:
            seq = self.sequence
            self.sequence = self.sequence + 1
            self.pending[seq] = _Mailbox()
        finally:
            self.lock.release()
        logging.debug( "sequence %d allocated" % seq)
        return seq


    def put_data(self, seq, new_data):
        seq = long(seq)
        logging.debug( "putting data [%r] to seq %d..." % (new_data, seq) )
        self.lock.acquire()
        try:
            if seq in self.pending:
                logging.error("Putting seq %d @ %s" % (seq,time.time()))
                self.pending[seq].deliver(new_data)
            else:
                logging.error( "seq %d not found!" % seq )
        finally:
            self.lock.release()



    def get_data(self, seq, timeout=None):
        """ 
        Release a sequence number reserved using the reserve method.  This must
        be called when the sequence is no longer needed.
        
        @type seq: int
        @param seq: a sequence previously allocated by calling reserve().
        @rtype: any
        @return: the data originally associated with the reserved sequence number.
        """
        seq = long(seq)
        logging.debug( "getting data for seq=%d" % seq)
        mbox = None
        self.lock.acquire()
        try:
            if seq in self.pending:
                mbox = self.pending[seq]
        finally:
            self.lock.release()

        # Note well: pending list is unlocked, so we can wait.
        # we reference mbox locally, so it will not be released
        # until we are done.

        if mbox:
            d = mbox.fetch(timeout)
            logging.debug( "seq %d fetched %r!" % (seq, d) )
            return d

        logging.debug( "seq %d not found!" % seq )
        return None


    def release(self, seq):
        """
        Release the sequence, and its mailbox
        """
        seq = long(seq)
        logging.debug( "releasing seq %d" % seq )
        self.lock.acquire()
        try:
            if seq in self.pending:
                del self.pending[seq]
        finally:
            self.lock.release()


    def isValid(self, seq):
        """
        True if seq is in use, else False (seq is unknown)
        """
        seq = long(seq)
        self.lock.acquire()
        try:
            return seq in self.pending
        finally:
            self.lock.release()
        return False


##==============================================================================
## DATA MODEL
##==============================================================================


class QmfConsoleData(QmfData):
    """
    Console's representation of an managed QmfData instance.  
    """
    def __init__(self, map_, agent, _schema=None):
        super(QmfConsoleData, self).__init__(_map=map_,
                                             _schema=_schema,
                                             _const=True) 
        self._agent = agent

    def get_timestamps(self): 
        """
        Returns a list of timestamps describing the lifecycle of
        the object.  All timestamps are represented by the AMQP
        timestamp type.  [0] = time of last update from Agent,
                         [1] = creation timestamp 
                         [2] = deletion timestamp, or zero if not
        deleted.
        """
        return [self._utime, self._ctime, self._dtime]

    def get_create_time(self): 
        """
        returns the creation timestamp
        """
        return self._ctime

    def get_update_time(self): 
        """
        returns the update timestamp
        """
        return self._utime

    def get_delete_time(self): 
        """
        returns the deletion timestamp, or zero if not yet deleted.
        """
        return self._dtime

    def is_deleted(self): 
        """
        True if deletion timestamp not zero.
        """
        return self._dtime != long(0)

    def refresh(self, _reply_handle=None, _timeout=None): 
        """
        request that the Agent update the value of this object's
        contents.
        """
        logging.error(" TBD!!!")
        return None

    def invoke_method(self, name, _in_args=None, _reply_handle=None,
                      _timeout=None):
        """
        invoke the named method.
        """
        logging.error(" TBD!!!")
        return None



class QmfLocalData(QmfData):
    """
    Console's representation of an unmanaged QmfData instance.  There
    is no remote agent associated with this instance. The Console has
    full control over this instance.
    """
    def __init__(self, values, _subtypes={}, _tag=None, _object_id=None,
                 _schema=None):
        # timestamp in millisec since epoch UTC
        ctime = long(time.time() * 1000)
        super(QmfLocalData, self).__init__(_values=values,
                                           _subtypes=_subtypes, _tag=_tag, 
                                           _object_id=_object_id,
                                           _schema=_schema, _ctime=ctime,
                                           _utime=ctime, _const=False)


class Agent(object):
    """
    A local representation of a remote agent managed by this console.
    """
    def __init__(self, name, console):
        """
        @type name: AgentId
        @param name: uniquely identifies this agent in the AMQP domain.
        """

        if not isinstance(console, Console):
            raise TypeError("parameter must be an instance of class Console")

        self._name = name
        self._address = QmfAddress.direct(name, console._domain)
        self._console = console
        self._sender = None
        self._packages = {} # map of {package-name:[list of class-names], } for this agent
        self._subscriptions = [] # list of active standing subscriptions for this agent
        self._announce_timestamp = None # datetime when last announce received
        logging.debug( "Created Agent with address: [%s]" % self._address )


    def get_name(self):
        return self._name

    def isActive(self):
        return self._announce_timestamp != None
    
    def _sendMsg(self, msg, correlation_id=None):
        """
        Low-level routine to asynchronously send a message to this agent.
        """
        msg.reply_to = str(self._console._address)
        # handle = self._console._req_correlation.allocate()
        # if handle == 0:
        #    raise Exception("Can not allocate a correlation id!")
        # msg.correlation_id = str(handle)
        if correlation_id:
            msg.correlation_id = str(correlation_id)
        self._sender.send(msg)
        # return handle

    def get_packages(self):
        """
        Return a list of the names of all packages known to this agent.
        """
        return self._packages.keys()

    def get_classes(self):
        """
        Return a dictionary [key:class] of classes known to this agent.
        """
        return self._packages.copy()

    def get_objects(self, query, kwargs={}):
        """
        Return a list of objects that satisfy the given query.

        @type query: dict, or qmfCommon.Query
        @param query: filter for requested objects
        @type kwargs: dict
        @param kwargs: ??? used to build match selector and query ???
        @rtype: list
        @return: list of matching objects, or None.
        """
        pass

    def get_object(self, query, kwargs={}):
        """
        Get one object - query is expected to match only one object.
        ??? Recommended: explicit timeout param, default None ???

        @type query: dict, or qmfCommon.Query
        @param query: filter for requested objects
        @type kwargs: dict
        @param kwargs: ??? used to build match selector and query ???
        @rtype: qmfConsole.ObjectProxy
        @return: one matching object, or none
        """
        pass


    def create_subscription(self, query):
        """
        Factory for creating standing subscriptions based on a given query.

        @type query: qmfCommon.Query object
        @param query: determines the list of objects for which this subscription applies
        @rtype: qmfConsole.Subscription
        @returns: an object representing the standing subscription.
        """
        pass

    def __repr__(self):
        return str(self._address)
    
    def __str__(self):
        return self.__repr__()

    def _sendQuery(self, query, correlation_id=None):
        """
        """
        msg = Message(subject=makeSubject(OpCode.get_query),
                      properties={"method":"request"},
                      content={MsgKey.query: query.map_encode()})
        self._sendMsg( msg, correlation_id )


  ##==============================================================================
  ## CONSOLE
  ##==============================================================================



class WorkItem(object):
    """
    Describes an event that has arrived at the Console for the
    application to process.  The Notifier is invoked when one or 
    more of these WorkItems become available for processing.
    """
    #
    # Enumeration of the types of WorkItems produced by the Console
    #
    AGENT_ADDED = 1
    AGENT_DELETED = 2
    NEW_PACKAGE = 3
    NEW_CLASS = 4
    OBJECT_UPDATE = 5
    EVENT_RECEIVED = 7
    AGENT_HEARTBEAT = 8

    def __init__(self, kind, kwargs={}):
        """
        Used by the Console to create a work item.
        
        @type kind: int
        @param kind: work item type
        """
        self._kind = kind
        self._param_map = kwargs


    def getType(self):
        return self._kind

    def getParams(self):
        return self._param_map



class Console(Thread):
    """
    A Console manages communications to a collection of agents on behalf of an application.
    """
    def __init__(self, name=None, _domain=None, notifier=None, 
                 reply_timeout = 60,
                 # agent_timeout = 120,
                 agent_timeout = 60,
                 kwargs={}):
        """
        @type name: str
        @param name: identifier for this console.  Must be unique.
        @type notifier: qmfConsole.Notifier
        @param notifier: invoked when events arrive for processing.
        @type kwargs: dict
        @param kwargs: ??? Unused
        """
        Thread.__init__(self)
        if not name:
            self._name = "qmfc-%s.%d" % (platform.node(), os.getpid())
        else:
            self._name = str(name)
        self._domain = _domain
        self._address = QmfAddress.direct(self._name, self._domain)
        self._notifier = notifier
        self._lock = Lock()
        self._conn = None
        self._session = None
        # dict of "agent-direct-address":class Agent entries
        self._agent_map = {}
        self._direct_recvr = None
        self._announce_recvr = None
        self._locate_sender = None
        self._schema_cache = {}
        self._req_correlation = SequencedWaiter()
        self._operational = False
        self._agent_discovery_filter = None
        self._reply_timeout = reply_timeout
        self._agent_timeout = agent_timeout
        self._next_agent_expire = None
        # lock out run() thread
        self._cv = Condition()
        # for passing WorkItems to the application
        self._work_q = Queue.Queue()
        ## Old stuff below???
        #self._broker_list = []
        #self.impl = qmfengine.Console()
        #self._event = qmfengine.ConsoleEvent()
        ##self._cv = Condition()
        ##self._sync_count = 0
        ##self._sync_result = None
        ##self._select = {}
        ##self._cb_cond = Condition()



    def destroy(self, timeout=None):
        """
        Must be called before the Console is deleted.  
        Frees up all resources and shuts down all background threads.

        @type timeout: float
        @param timeout: maximum time in seconds to wait for all background threads to terminate.  Default: forever.
        """
        logging.debug("Destroying Console...")
        if self._conn:
            self.removeConnection(self._conn, timeout)
        logging.debug("Console Destroyed")



    def addConnection(self, conn):
        """
        Add a AMQP connection to the console.  The console will setup a session over the
        connection.  The console will then broadcast an Agent Locate Indication over
        the session in order to discover present agents.

        @type conn: qpid.messaging.Connection
        @param conn: the connection to the AMQP messaging infrastructure.
        """
        if self._conn:
            raise Exception( "Multiple connections per Console not supported." );
        self._conn = conn
        self._session = conn.session(name=self._name)
        self._direct_recvr = self._session.receiver(str(self._address) +
                                                    ";{create:always,"
                                                    " node-properties:"
                                                    " {type:topic,"
                                                    " x-properties:"
                                                    " {type:direct}}}", 
                                                    capacity=1)
        ind_addr = QmfAddress.topic(AMQP_QMF_AGENT_INDICATION, self._domain)
        logging.debug("agent.ind addr=%s" % ind_addr)
        self._announce_recvr = self._session.receiver(str(ind_addr) +
                                                      ";{create:always,"
                                                      " node-properties:{type:topic}}",
                                                      capacity=1)
        locate_addr = QmfAddress.topic(AMQP_QMF_AGENT_LOCATE, self._domain)
        logging.debug("agent.locate addr=%s" % locate_addr)
        self._locate_sender = self._session.sender(str(locate_addr) +
                                                   ";{create:always,"
                                                   " node-properties:{type:topic}}")
        #
        # Now that receivers are created, fire off the receive thread...
        #
        self._operational = True
        self.start()



    def removeConnection(self, conn, timeout=None):
        """
        Remove an AMQP connection from the console.  Un-does the add_connection() operation,
        and releases any agents and sessions associated with the connection.

        @type conn: qpid.messaging.Connection
        @param conn: connection previously added by add_connection()
        """
        if self._conn and conn and conn != self._conn:
            logging.error( "Attempt to delete unknown connection: %s" % str(conn))

        # tell connection thread to shutdown
        self._operational = False
        if self.isAlive():
            # kick my thread to wake it up
            logging.debug("Making temp sender for [%s]" % self._address)
            tmp_sender = self._session.sender(str(self._address))
            try:
                msg = Message(subject=makeSubject(OpCode.noop))
                tmp_sender.send( msg, sync=True )
            except SendError, e:
                logging.error(str(e))
            logging.debug("waiting for console receiver thread to exit")
            self.join(timeout)
            if self.isAlive():
                logging.error( "Console thread '%s' is hung..." % self.getName() )
        self._direct_recvr.close()
        self._announce_recvr.close()
        self._locate_sender.close()
        self._session.close()
        self._session = None
        self._conn = None
        logging.debug("console connection removal complete")


    def getAddress(self):
        """
        The AMQP address this Console is listening to.
        """
        return self._address


    def destroyAgent( self, agent ):
        """
        Undoes create.
        """
        if not isinstance(agent, Agent):
            raise TypeError("agent must be an instance of class Agent")

        self._lock.acquire()
        try:
            if agent._id in self._agent_map:
                del self._agent_map[agent._id]
        finally:
            self._lock.release()

    def findAgent(self, name, timeout=None ):
        """
        Given the id of a particular agent, return an instance of class Agent 
        representing that agent.  Return None if the agent does not exist.
        """

        self._lock.acquire()
        try:
            agent = self._agent_map.get(name)
            if agent:
                return agent
        finally:
            self._lock.release()

        # agent not present yet - ping it with an agent_locate

        handle = self._req_correlation.allocate()
        if handle == 0:
            raise Exception("Can not allocate a correlation id!")
        try:
            tmp_sender = self._session.sender(str(QmfAddress.direct(name,
                                                                    self._domain))
                                              + ";{create:always,"
                                              " node-properties:"
                                              " {type:topic,"
                                              " x-properties:"
                                              " {type:direct}}}")

            query = QmfQuery({QmfQuery._TARGET: {QmfQuery._TARGET_AGENT:None},
                              QmfQuery._PREDICATE:
                                  {QmfQuery._CMP_EQ: ["_name", name]}})
            msg = Message(subject=makeSubject(OpCode.agent_locate),
                          properties={"method":"request"},
                          content={MsgKey.query: query.map_encode()})
            msg.reply_to = str(self._address)
            msg.correlation_id = str(handle)
            logging.debug("Sending Agent Locate (%s)" % time.time())
            tmp_sender.send( msg )
        except SendError, e:
            logging.error(str(e))
            self._req_correlation.release(handle)
            return None

        if not timeout:
            timeout = self._reply_timeout

        new_agent = None
        logging.debug("Waiting for response to Agent Locate (%s)" % timeout)
        self._req_correlation.get_data( handle, timeout )
        self._req_correlation.release(handle)
        logging.debug("Agent Locate wait ended (%s)" % time.time())
        self._lock.acquire()
        try:
            new_agent = self._agent_map.get(name)
        finally:
            self._lock.release()
        return new_agent


    def doQuery(self, agent, query, timeout=None ):
        """
        """

        handle = self._req_correlation.allocate()
        if handle == 0:
            raise Exception("Can not allocate a correlation id!")
        try:
            logging.debug("Sending Query to Agent (%s)" % time.time())
            agent._sendQuery(query, handle)
        except SendError, e:
            logging.error(str(e))
            self._req_correlation.release(handle)
            return None

        if not timeout:
            timeout = self._reply_timeout

        logging.debug("Waiting for response to Query (%s)" % timeout)
        reply = self._req_correlation.get_data(handle, timeout)
        self._req_correlation.release(handle)
        logging.debug("Agent Query wait ended (%s)" % time.time())
        if reply:
            print("Agent Query Reply='%s'" % reply)
            return reply.content
        else:
            print("Agent Query FAILED!!!")
            return None



    def run(self):
        global _callback_thread
        #
        # @todo KAG Rewrite when api supports waiting on multiple receivers
        #
        while self._operational:

            qLen = self._work_q.qsize()

            try:
                msg = self._announce_recvr.fetch(timeout = 0)
                if msg:
                    self._dispatch(msg, _direct=False)
            except Empty:
                pass

            try:
                msg = self._direct_recvr.fetch(timeout = 0)
                if msg:
                    self._dispatch(msg, _direct=True)
            except Empty:
                pass

            self._expireAgents()   # check for expired agents

            if qLen == 0 and self._work_q.qsize() and self._notifier:
                # work queue went non-empty, kick
                # the application...

                _callback_thread = currentThread()
                logging.info("Calling console indication")
                self._notifier.indication()
                _callback_thread = None

            if self._operational:
                # wait for a message to arrive or an agent
                # to expire
                now = datetime.datetime.utcnow()
                if self._next_agent_expire > now:
                    timeout = ((self._next_agent_expire - now) + datetime.timedelta(microseconds=999999)).seconds
                    try:
                        logging.error("waiting for next rcvr (timeout=%s)..." % timeout)
                        self._session.next_receiver(timeout = timeout)
                    except Empty:
                        pass


        logging.debug("Shutting down Console thread")



    # called by run() thread ONLY
    #
    def _dispatch(self, msg, _direct=True):
        """
        PRIVATE: Process a message received from an Agent
        """

        logging.error( "Message received from Agent! [%s]" % msg )

        try:
            version,opcode = parseSubject(msg.subject)
            # @todo: deal with version mismatch!!!
        except:
            logging.error("Ignoring unrecognized broadcast message '%s'" % msg.subject)
            return

        cmap = {}; props = {}
        if msg.content_type == "amqp/map":
            cmap = msg.content
        if msg.properties:
            props = msg.properties

        if opcode == OpCode.agent_ind:
            self._handleAgentIndMsg( msg, cmap, version, _direct )
        elif opcode == OpCode.data_ind:
            self._handleDataIndMsg(msg, cmap, version, _direct)
        elif opcode == OpCode.event_ind:
            logging.warning("!!! event_ind TBD !!!")
        elif opcode == OpCode.managed_object:
            logging.warning("!!! managed_object TBD !!!")
        elif opcode == OpCode.object_ind:
            logging.warning("!!! object_ind TBD !!!")
        elif opcode == OpCode.response:
            logging.warning("!!! response TBD !!!")
        elif opcode == OpCode.schema_ind:
            logging.warning("!!! schema_ind TBD !!!")
        elif opcode == OpCode.noop:
             logging.debug("No-op msg received.")
        else:
            logging.warning("Ignoring message with unrecognized 'opcode' value: '%s'" % opcode)


    def _handleAgentIndMsg(self, msg, cmap, version, direct):
        """
        Process a received agent-ind message.  This message may be a response to a
        agent-locate, or it can be an unsolicited agent announce.
        """
        logging.debug("_handleAgentIndMsg '%s' (%s)" % (msg, time.time()))

        if MsgKey.agent_info in cmap:
            try:
                # TODO: fix
                name = cmap[MsgKey.agent_info]["_name"]
            except:
                logging.warning("Bad agent-ind message received: '%s'" % msg)
                return

        ignore = True
        matched = False
        correlated = False
        if msg.correlation_id:
            correlated = self._req_correlation.isValid(msg.correlation_id)
        if direct and correlated:
            ignore = False
        elif self._agent_discovery_filter:
            logging.error("FIXME: agent discovery filter - new agent name style")
            # matched = self._agent_discovery_filter.evaluate(QmfData(agent_id.mapEncode()))
            # ignore = not matched
            matched = True; ignore = False  # for now

        if not ignore:
            agent = None
            self._lock.acquire()
            try:
                agent = self._agent_map.get(name)
            finally:
                self._lock.release()

            if not agent:
                # need to create and add a new agent
                agent = self._createAgent(name)

            # lock out expiration scanning code
            self._lock.acquire()
            try:
                old_timestamp = agent._announce_timestamp
                agent._announce_timestamp = datetime.datetime.utcnow()
            finally:
                self._lock.release()

            if old_timestamp == None and matched:
                logging.error("AGENT_ADDED for %s (%s)" % (agent, time.time()))
                wi = WorkItem(WorkItem.AGENT_ADDED,
                              {"agent": agent})
                self._work_q.put(wi)

            if correlated:
                # wake up all waiters
                logging.error("waking waiters for correlation id %s" % msg.correlation_id)
                self._req_correlation.put_data(msg.correlation_id, msg)




    def _handleDataIndMsg(self, msg, cmap, version, direct):
        """
        Process a received data-ind message.
        """
        logging.debug("_handleDataIndMsg '%s' (%s)" % (msg, time.time()))

        if not self._req_correlation.isValid(msg.correlation_id):
            logging.error("FIXME: uncorrelated data indicate??? msg='%s'" % str(msg))
            return

        # wake up all waiters
        logging.error("waking waiters for correlation id %s" % msg.correlation_id)
        self._req_correlation.put_data(msg.correlation_id, msg)


    def _expireAgents(self):
        """
        Check for expired agents and issue notifications when they expire.
        """
        now = datetime.datetime.utcnow()
        if self._next_agent_expire and now < self._next_agent_expire:
            return
        lifetime_delta = datetime.timedelta(seconds = self._agent_timeout)
        next_expire_delta = lifetime_delta
        self._lock.acquire()
        try:
            logging.debug("!!! expiring agents '%s'" % now)
            for agent in self._agent_map.itervalues():
                if agent._announce_timestamp:
                    agent_deathtime = agent._announce_timestamp + lifetime_delta
                    if agent_deathtime <= now:
                        logging.debug("AGENT_DELETED for %s" % agent)
                        agent._announce_timestamp = None
                        wi = WorkItem(WorkItem.AGENT_DELETED, {"agent":agent})
                        self._work_q.put(wi)
                    else:
                        if (agent_deathtime - now) < next_expire_delta:
                            next_expire_delta = agent_deathtime - now

            self._next_agent_expire = now + next_expire_delta
            logging.debug("!!! next expire cycle = '%s'" % self._next_agent_expire)
        finally:
            self._lock.release()



    def _createAgent( self, name ):
        """
        Factory to create/retrieve an agent for this console
        """

        self._lock.acquire()
        try:
            agent = self._agent_map.get(name)
            if agent:
                return agent

            agent = Agent(name, self)
            agent._sender = self._session.sender(str(agent._address) + 
                                                    ";{create:always,"
                                                    " node-properties:"
                                                    " {type:topic,"
                                                    " x-properties:"
                                                    " {type:direct}}}") 

            self._agent_map[name] = agent
        finally:
            self._lock.release()

        # new agent - query for its schema database for
        # seeding the schema cache (@todo)
        # query = QmfQuery({QmfQuery._TARGET_SCHEMA_ID:None})
        # agent._sendQuery( query )

        return agent



    def enableAgentDiscovery(self, query=None):
        """
        Called to enable the asynchronous Agent Discovery process.
        Once enabled, AGENT_ADD work items can arrive on the WorkQueue.
        """
        if query:
            if not isinstance(query, QmfQuery):
                raise TypeError("Type QmfQuery expected")
            self._agent_discovery_filter = query
        else:
            # create a match-all agent query (no predicate)
            self._agent_discovery_filter = QmfQuery({QmfQuery._TARGET: 
                                                     {QmfQuery._TARGET_AGENT:None}})

    def disableAgentDiscovery(self):
        """
        Called to disable the async Agent Discovery process enabled by
        calling enableAgentDiscovery()
        """
        self._agent_discovery_filter = None



    def getWorkItemCount(self):
        """
        Returns the count of pending WorkItems that can be retrieved.
        """
        return self._work_q.qsize()



    def getNextWorkItem(self, timeout=None):
        """
        Returns the next pending work item, or None if none available.
        @todo: subclass and return an Empty event instead.
        """
        try:
            wi = self._work_q.get(True, timeout)
        except:
            return None
        return wi


    def releaseWorkItem(self, wi):
        """
        Return a WorkItem to the Console when it is no longer needed.
        @todo: call Queue.task_done() - only 2.5+

        @type wi: class qmfConsole.WorkItem
        @param wi: work item object to return.
        """
        pass


    # def get_packages(self):
    #     plist = []
    #     for i in range(self.impl.packageCount()):
    #         plist.append(self.impl.getPackageName(i))
    #     return plist
    
    
    # def get_classes(self, package, kind=CLASS_OBJECT):
    #     clist = []
    #     for i in range(self.impl.classCount(package)):
    #         key = self.impl.getClass(package, i)
    #         class_kind = self.impl.getClassKind(key)
    #         if class_kind == kind:
    #             if kind == CLASS_OBJECT:
    #                 clist.append(SchemaObjectClass(None, None, {"impl":self.impl.getObjectClass(key)}))
    #             elif kind == CLASS_EVENT:
    #                 clist.append(SchemaEventClass(None, None, {"impl":self.impl.getEventClass(key)}))
    #     return clist
    
    
    # def bind_package(self, package):
    #     return self.impl.bindPackage(package)
    
    
    # def bind_class(self, kwargs = {}):
    #     if "key" in kwargs:
    #         self.impl.bindClass(kwargs["key"])
    #     elif "package" in kwargs:
    #         package = kwargs["package"]
    #         if "class" in kwargs:
    #             self.impl.bindClass(package, kwargs["class"])
    #         else:
    #             self.impl.bindClass(package)
    #     else:
    #         raise Exception("Argument error: invalid arguments, use 'key' or 'package'[,'class']")
    
    
    # def get_agents(self, broker=None):
    #     blist = []
    #     if broker:
    #         blist.append(broker)
    #     else:
    #         self._cv.acquire()
    #         try:
    #             # copy while holding lock
    #             blist = self._broker_list[:]
    #         finally:
    #             self._cv.release()

    #     agents = []
    #     for b in blist:
    #         for idx in range(b.impl.agentCount()):
    #             agents.append(AgentProxy(b.impl.getAgent(idx), b))

    #     return agents
    
    
    # def get_objects(self, query, kwargs = {}):
    #     timeout = 30
    #     agent = None
    #     temp_args = kwargs.copy()
    #     if type(query) == type({}):
    #         temp_args.update(query)

    #     if "_timeout" in temp_args:
    #         timeout = temp_args["_timeout"]
    #         temp_args.pop("_timeout")

    #     if "_agent" in temp_args:
    #         agent = temp_args["_agent"]
    #         temp_args.pop("_agent")

    #     if type(query) == type({}):
    #         query = Query(temp_args)

    #     self._select = {}
    #     for k in temp_args.iterkeys():
    #         if type(k) == str:
    #             self._select[k] = temp_args[k]

    #     self._cv.acquire()
    #     try:
    #         self._sync_count = 1
    #         self._sync_result = []
    #         broker = self._broker_list[0]
    #         broker.send_query(query.impl, None, agent)
    #         self._cv.wait(timeout)
    #         if self._sync_count == 1:
    #             raise Exception("Timed out: waiting for query response")
    #     finally:
    #         self._cv.release()

    #     return self._sync_result
    
    
    # def get_object(self, query, kwargs = {}):
    #     '''
    #     Return one and only one object or None.
    #     '''
    #     objs = objects(query, kwargs)
    #     if len(objs) == 1:
    #         return objs[0]
    #     else:
    #         return None


    # def first_object(self, query, kwargs = {}):
    #     '''
    #     Return the first of potentially many objects.
    #     '''
    #     objs = objects(query, kwargs)
    #     if objs:
    #         return objs[0]
    #     else:
    #         return None


    # # Check the object against select to check for a match
    # def _select_match(self, object):
    #     schema_props = object.properties()
    #     for key in self._select.iterkeys():
    #         for prop in schema_props:
    #             if key == p[0].name() and self._select[key] != p[1]:
    #                 return False
    #     return True


    # def _get_result(self, list, context):
    #     '''
    #     Called by Broker proxy to return the result of a query.
    #     '''
    #     self._cv.acquire()
    #     try:
    #         for item in list:
    #             if self._select_match(item):
    #                 self._sync_result.append(item)
    #         self._sync_count -= 1
    #         self._cv.notify()
    #     finally:
    #         self._cv.release()


    # def start_sync(self, query): pass
    
    
    # def touch_sync(self, sync): pass
    
    
    # def end_sync(self, sync): pass
    
    


#     def start_console_events(self):
#         self._cb_cond.acquire()
#         try:
#             self._cb_cond.notify()
#         finally:
#             self._cb_cond.release()


#     def _do_console_events(self):
#         '''
#         Called by the Console thread to poll for events.  Passes the events
#         onto the ConsoleHandler associated with this Console.  Is called
#         periodically, but can also be kicked by Console.start_console_events().
#         '''
#         count = 0
#         valid = self.impl.getEvent(self._event)
#         while valid:
#             count += 1
#             try:
#                 if self._event.kind == qmfengine.ConsoleEvent.AGENT_ADDED:
#                     logging.debug("Console Event AGENT_ADDED received")
#                     if self._handler:
#                         self._handler.agent_added(AgentProxy(self._event.agent, None))
#                 elif self._event.kind == qmfengine.ConsoleEvent.AGENT_DELETED:
#                     logging.debug("Console Event AGENT_DELETED received")
#                     if self._handler:
#                         self._handler.agent_deleted(AgentProxy(self._event.agent, None))
#                 elif self._event.kind == qmfengine.ConsoleEvent.NEW_PACKAGE:
#                     logging.debug("Console Event NEW_PACKAGE received")
#                     if self._handler:
#                         self._handler.new_package(self._event.name)
#                 elif self._event.kind == qmfengine.ConsoleEvent.NEW_CLASS:
#                     logging.debug("Console Event NEW_CLASS received")
#                     if self._handler:
#                         self._handler.new_class(SchemaClassKey(self._event.classKey))
#                 elif self._event.kind == qmfengine.ConsoleEvent.OBJECT_UPDATE:
#                     logging.debug("Console Event OBJECT_UPDATE received")
#                     if self._handler:
#                         self._handler.object_update(ConsoleObject(None, {"impl":self._event.object}),
#                                                     self._event.hasProps, self._event.hasStats)
#                 elif self._event.kind == qmfengine.ConsoleEvent.EVENT_RECEIVED:
#                     logging.debug("Console Event EVENT_RECEIVED received")
#                 elif self._event.kind == qmfengine.ConsoleEvent.AGENT_HEARTBEAT:
#                     logging.debug("Console Event AGENT_HEARTBEAT received")
#                     if self._handler:
#                         self._handler.agent_heartbeat(AgentProxy(self._event.agent, None), self._event.timestamp)
#                 elif self._event.kind == qmfengine.ConsoleEvent.METHOD_RESPONSE:
#                     logging.debug("Console Event METHOD_RESPONSE received")
#                 else:
#                     logging.debug("Console thread received unknown event: '%s'" % str(self._event.kind))
#             except e:
#                 print "Exception caught in callback thread:", e
#             self.impl.popEvent()
#             valid = self.impl.getEvent(self._event)
#         return count





# class Broker(ConnectionHandler):
#     #   attr_reader :impl :conn, :console, :broker_bank
#     def __init__(self, console, conn):
#         self.broker_bank = 1
#         self.console = console
#         self.conn = conn
#         self._session = None
#         self._cv = Condition()
#         self._stable = None
#         self._event = qmfengine.BrokerEvent()
#         self._xmtMessage = qmfengine.Message()
#         self.impl = qmfengine.BrokerProxy(self.console.impl)
#         self.console.impl.addConnection(self.impl, self)
#         self.conn.add_conn_handler(self)
#         self._operational = True
    
    
#     def shutdown(self):
#         logging.debug("broker.shutdown() called.")
#         self.console.impl.delConnection(self.impl)
#         self.conn.del_conn_handler(self)
#         if self._session:
#             self.impl.sessionClosed()
#             logging.debug("broker.shutdown() sessionClosed done.")
#             self._session.destroy()
#             logging.debug("broker.shutdown() session destroy done.")
#             self._session = None
#         self._operational = False
#         logging.debug("broker.shutdown() done.")


#     def wait_for_stable(self, timeout = None):
#         self._cv.acquire()
#         try:
#             if self._stable:
#                 return
#             if timeout:
#                 self._cv.wait(timeout)
#                 if not self._stable:
#                     raise Exception("Timed out: waiting for broker connection to become stable")
#             else:
#                 while not self._stable:
#                     self._cv.wait()
#         finally:
#             self._cv.release()


#     def send_query(self, query, ctx, agent):
#         agent_impl = None
#         if agent:
#             agent_impl = agent.impl
#         self.impl.sendQuery(query, ctx, agent_impl)
#         self.conn.kick()


#     def _do_broker_events(self):
#         count = 0
#         valid = self.impl.getEvent(self._event)
#         while valid:
#             count += 1
#             if self._event.kind == qmfengine.BrokerEvent.BROKER_INFO:
#                 logging.debug("Broker Event BROKER_INFO received");
#             elif self._event.kind == qmfengine.BrokerEvent.DECLARE_QUEUE:
#                 logging.debug("Broker Event DECLARE_QUEUE received");
#                 self.conn.impl.declareQueue(self._session.handle, self._event.name)
#             elif self._event.kind == qmfengine.BrokerEvent.DELETE_QUEUE:
#                 logging.debug("Broker Event DELETE_QUEUE received");
#                 self.conn.impl.deleteQueue(self._session.handle, self._event.name)
#             elif self._event.kind == qmfengine.BrokerEvent.BIND:
#                 logging.debug("Broker Event BIND received");
#                 self.conn.impl.bind(self._session.handle, self._event.exchange, self._event.name, self._event.bindingKey)
#             elif self._event.kind == qmfengine.BrokerEvent.UNBIND:
#                 logging.debug("Broker Event UNBIND received");
#                 self.conn.impl.unbind(self._session.handle, self._event.exchange, self._event.name, self._event.bindingKey)
#             elif self._event.kind == qmfengine.BrokerEvent.SETUP_COMPLETE:
#                 logging.debug("Broker Event SETUP_COMPLETE received");
#                 self.impl.startProtocol()
#             elif self._event.kind == qmfengine.BrokerEvent.STABLE:
#                 logging.debug("Broker Event STABLE received");
#                 self._cv.acquire()
#                 try:
#                     self._stable = True
#                     self._cv.notify()
#                 finally:
#                     self._cv.release()
#             elif self._event.kind == qmfengine.BrokerEvent.QUERY_COMPLETE:
#                 result = []
#                 for idx in range(self._event.queryResponse.getObjectCount()):
#                     result.append(ConsoleObject(None, {"impl":self._event.queryResponse.getObject(idx), "broker":self}))
#                 self.console._get_result(result, self._event.context)
#             elif self._event.kind == qmfengine.BrokerEvent.METHOD_RESPONSE:
#                 obj = self._event.context
#                 obj._method_result(MethodResponse(self._event.methodResponse()))
            
#             self.impl.popEvent()
#             valid = self.impl.getEvent(self._event)
        
#         return count
    
    
#     def _do_broker_messages(self):
#         count = 0
#         valid = self.impl.getXmtMessage(self._xmtMessage)
#         while valid:
#             count += 1
#             logging.debug("Broker: sending msg on connection")
#             self.conn.impl.sendMessage(self._session.handle, self._xmtMessage)
#             self.impl.popXmt()
#             valid = self.impl.getXmtMessage(self._xmtMessage)
        
#         return count
    
    
#     def _do_events(self):
#         while True:
#             self.console.start_console_events()
#             bcnt = self._do_broker_events()
#             mcnt = self._do_broker_messages()
#             if bcnt == 0 and mcnt == 0:
#                 break;
    
    
#     def conn_event_connected(self):
#         logging.debug("Broker: Connection event CONNECTED")
#         self._session = Session(self.conn, "qmfc-%s.%d" % (socket.gethostname(), os.getpid()), self)
#         self.impl.sessionOpened(self._session.handle)
#         self._do_events()
    
    
#     def conn_event_disconnected(self, error):
#         logging.debug("Broker: Connection event DISCONNECTED")
#         pass
    
    
#     def conn_event_visit(self):
#         self._do_events()


#     def sess_event_session_closed(self, context, error):
#         logging.debug("Broker: Session event CLOSED")
#         self.impl.sessionClosed()
    
    
#     def sess_event_recv(self, context, message):
#         logging.debug("Broker: Session event MSG_RECV")
#         if not self._operational:
#             logging.warning("Unexpected session event message received by Broker proxy: context='%s'" % str(context))
#         self.impl.handleRcvMessage(message)
#         self._do_events()



################################################################################
################################################################################
################################################################################
################################################################################
#                 TEMPORARY TEST CODE - TO BE DELETED
################################################################################
################################################################################
################################################################################
################################################################################

if __name__ == '__main__':
    # temp test code
    from qmfCommon import (qmfTypes, QmfData,
                           QmfEvent, SchemaClassId, SchemaEventClass,
                           SchemaProperty, SchemaObjectClass)

    logging.getLogger().setLevel(logging.INFO)

    logging.info( "************* Creating Async Console **************" )

    class MyNotifier(Notifier):
        def __init__(self, context):
            self._myContext = context
            self.WorkAvailable = False

        def indication(self):
            print("Indication received! context=%d" % self._myContext)
            self.WorkAvailable = True

    _noteMe = MyNotifier( 666 )

    _myConsole = Console(notifier=_noteMe)

    _myConsole.enableAgentDiscovery()
    logging.info("Waiting...")


    logging.info( "Destroying console:" )
    _myConsole.destroy( 10 )

    logging.info( "******** Messing around with Schema ********" )

    _sec = SchemaEventClass( _classId=SchemaClassId("myPackage", "myClass",
                                                    stype=SchemaClassId.TYPE_EVENT), 
                             _desc="A typical event schema",
                             _props={"Argument-1": SchemaProperty(_type_code=qmfTypes.TYPE_UINT8,
                                                                  kwargs = {"min":0,
                                                                            "max":100,
                                                                            "unit":"seconds",
                                                                            "desc":"sleep value"}),
                                     "Argument-2": SchemaProperty(_type_code=qmfTypes.TYPE_LSTR,
                                                                  kwargs={"maxlen":100,
                                                                          "desc":"a string argument"})})
    print("_sec=%s" % _sec.get_class_id())
    print("_sec.gePropertyCount()=%d" % _sec.get_property_count() )
    print("_sec.getProperty('Argument-1`)=%s" % _sec.get_property('Argument-1') )
    print("_sec.getProperty('Argument-2`)=%s" % _sec.get_property('Argument-2') )
    try:
        print("_sec.getProperty('not-found')=%s" % _sec.get_property('not-found') )
    except:
        pass
    print("_sec.getProperties()='%s'" % _sec.get_properties())

    print("Adding another argument")
    _arg3 = SchemaProperty( _type_code=qmfTypes.TYPE_BOOL,
                            kwargs={"dir":"IO",
                                    "desc":"a boolean argument"})
    _sec.add_property('Argument-3', _arg3)
    print("_sec=%s" % _sec.get_class_id())
    print("_sec.getPropertyCount()=%d" % _sec.get_property_count() )
    print("_sec.getProperty('Argument-1')=%s" % _sec.get_property('Argument-1') )
    print("_sec.getProperty('Argument-2')=%s" % _sec.get_property('Argument-2') )
    print("_sec.getProperty('Argument-3')=%s" % _sec.get_property('Argument-3') )

    print("_arg3.mapEncode()='%s'" % _arg3.map_encode() )

    _secmap = _sec.map_encode()
    print("_sec.mapEncode()='%s'" % _secmap )

    _sec2 = SchemaEventClass( _map=_secmap )

    print("_sec=%s" % _sec.get_class_id())
    print("_sec2=%s" % _sec2.get_class_id())

    _soc = SchemaObjectClass( _map = {"_schema_id": {"_package_name": "myOtherPackage",
                                                     "_class_name":   "myOtherClass",
                                                     "_type":         "_data"},
                                      "_desc": "A test data object",
                                      "_values":
                                          {"prop1": {"amqp_type": qmfTypes.TYPE_UINT8,
                                                     "access": "RO",
                                                     "index": True,
                                                     "unit": "degrees"},
                                           "prop2": {"amqp_type": qmfTypes.TYPE_UINT8,
                                                     "access": "RW",
                                                     "index": True,
                                                     "desc": "The Second Property(tm)",
                                                     "unit": "radians"},
                                           "statistics": { "amqp_type": qmfTypes.TYPE_DELTATIME,
                                                           "unit": "seconds",
                                                           "desc": "time until I retire"},
                                           "meth1": {"desc": "A test method",
                                                     "arguments":
                                                         {"arg1": {"amqp_type": qmfTypes.TYPE_UINT32,
                                                                   "desc": "an argument 1",
                                                                   "dir":  "I"},
                                                          "arg2": {"amqp_type": qmfTypes.TYPE_BOOL,
                                                                   "dir":  "IO",
                                                                   "desc": "some weird boolean"}}},
                                           "meth2": {"desc": "A test method",
                                                     "arguments":
                                                         {"m2arg1": {"amqp_type": qmfTypes.TYPE_UINT32,
                                                                     "desc": "an 'nuther argument",
                                                                     "dir":
                                                                         "I"}}}},
                                      "_subtypes":
                                          {"prop1":"qmfProperty",
                                           "prop2":"qmfProperty",
                                           "statistics":"qmfProperty",
                                           "meth1":"qmfMethod",
                                           "meth2":"qmfMethod"},
                                      "_primary_key_names": ["prop2", "prop1"]})

    print("_soc='%s'" % _soc)

    print("_soc.getPrimaryKeyList='%s'" % _soc.get_id_names())

    print("_soc.getPropertyCount='%d'" % _soc.get_property_count())
    print("_soc.getProperties='%s'" % _soc.get_properties())
    print("_soc.getProperty('prop2')='%s'" % _soc.get_property('prop2'))

    print("_soc.getMethodCount='%d'" % _soc.get_method_count())
    print("_soc.getMethods='%s'" % _soc.get_methods())
    print("_soc.getMethod('meth2')='%s'" % _soc.get_method('meth2'))

    _socmap = _soc.map_encode()
    print("_socmap='%s'" % _socmap)
    _soc2 = SchemaObjectClass( _map=_socmap )
    print("_soc='%s'" % _soc)
    print("_soc2='%s'" % _soc2)

    if _soc2.get_class_id() == _soc.get_class_id():
        print("soc and soc2 are the same schema")


    logging.info( "******** Messing around with ObjectIds ********" )


    qd = QmfData( _values={"prop1":1, "prop2":True, "prop3": {"a":"map"}, "prop4": "astring"} )
    print("qd='%s':" % qd)

    print("prop1=%d prop2=%s prop3=%s prop4=%s" % (qd.prop1, qd.prop2, qd.prop3, qd.prop4))

    print("qd map='%s'" % qd.map_encode())
    print("qd getProperty('prop4')='%s'" % qd.get_value("prop4"))
    qd.set_value("prop4", 4, "A test property called 4")
    print("qd setProperty('prop4', 4)='%s'" % qd.get_value("prop4"))
    qd.prop4 = 9
    print("qd.prop4 = 9 ='%s'" % qd.prop4)
    qd["prop4"] = 11
    print("qd[prop4] = 11 ='%s'" % qd["prop4"])

    print("qd.mapEncode()='%s'" % qd.map_encode())
    _qd2 = QmfData( _map = qd.map_encode() )
    print("_qd2.mapEncode()='%s'" % _qd2.map_encode())

    _qmfDesc1 = QmfConsoleData( {"_values" : {"prop1": 1, "statistics": 666,
                                              "prop2": 0}},
                                agent="some agent name?",
                                _schema = _soc)

    print("_qmfDesc1 map='%s'" % _qmfDesc1.map_encode())

    _qmfDesc1._set_schema( _soc )

    print("_qmfDesc1 prop2 = '%s'" % _qmfDesc1.get_value("prop2"))
    print("_qmfDesc1 primarykey = '%s'" % _qmfDesc1.get_object_id())
    print("_qmfDesc1 classid = '%s'" % _qmfDesc1.get_schema_class_id())


    _qmfDescMap = _qmfDesc1.map_encode()
    print("_qmfDescMap='%s'" % _qmfDescMap)

    _qmfDesc2 = QmfData( _map=_qmfDescMap, _schema=_soc )

    print("_qmfDesc2 map='%s'" % _qmfDesc2.map_encode())
    print("_qmfDesc2 prop2 = '%s'" % _qmfDesc2.get_value("prop2"))
    print("_qmfDesc2 primary key = '%s'" % _qmfDesc2.get_object_id())


    logging.info( "******** Messing around with QmfEvents ********" )


    _qmfevent1 = QmfEvent( _timestamp = 1111,
                           _schema = _sec,
                           _values = {"Argument-1": 77, 
                                      "Argument-3": True,
                                      "Argument-2": "a string"})
    print("_qmfevent1.mapEncode()='%s'" % _qmfevent1.map_encode())
    print("_qmfevent1.getTimestamp()='%s'" % _qmfevent1.get_timestamp())

    _qmfevent1Map = _qmfevent1.map_encode()

    _qmfevent2 = QmfEvent(_map=_qmfevent1Map, _schema=_sec)
    print("_qmfevent2.mapEncode()='%s'" % _qmfevent2.map_encode())


    logging.info( "******** Messing around with Queries ********" )

    _q1 = QmfQuery({QmfQuery._TARGET: {QmfQuery._TARGET_AGENT:None},
                    QmfQuery._PREDICATE:
                        {QmfQuery._LOGIC_AND: 
                         [{QmfQuery._CMP_EQ: ["vendor",  "AVendor"]},
                          {QmfQuery._CMP_EQ: ["product", "SomeProduct"]},
                          {QmfQuery._CMP_EQ: ["name", "Thingy"]},
                          {QmfQuery._LOGIC_OR:
                               [{QmfQuery._CMP_LE: ["temperature", -10]},
                                {QmfQuery._CMP_FALSE: None},
                                {QmfQuery._CMP_EXISTS: ["namey"]}]}]}})

    print("_q1.mapEncode() = [%s]" % _q1)