[rhmessaging-commits] rhmessaging commits: r4028 - in mgmt/newdata/cumin: python/cumin and 1 other directory.

Author: justi9 Date: 2010-06-16 14:20:33 -0400 (Wed, 16 Jun 2010) New Revision: 4028 Modified: mgmt/newdata/cumin/bin/cumin-smoke-test mgmt/newdata/cumin/python/cumin/session.py Log: A safer means of calling a qmf method; improve cumin-smoke-test Modified: mgmt/newdata/cumin/bin/cumin-smoke-test =================================================================== --- mgmt/newdata/cumin/bin/cumin-smoke-test 2010-06-15 18:47:09 UTC (rev 4027) +++ mgmt/newdata/cumin/bin/cumin-smoke-test 2010-06-16 18:20:33 UTC (rev 4028) @@ -3,6 +3,8 @@ import os import sys +from threading import Event + home = os.environ.get("CUMIN_HOME", os.path.normpath("/usr/share/cumin")) sys.path.append(os.path.join(home, "python")) @@ -36,24 +38,25 @@ cumin.start() - sleep(5) - conn = cumin.database.get_connection() cursor = conn.cursor() - cls = cumin.model.org_apache_qpid_broker.Broker - broker = cls.get_object(cursor) + # cls = cumin.model.org_apache_qpid_broker.Broker XXX fails + cls = cumin.model.com_redhat_grid.Scheduler + obj = cls.get_object(cursor) + print "Calling echo on", obj + + completed = Event() + def completion(x, y): - print "XXX", x, y + print x, y + completed.set() - cumin.session.call_method(completion, broker, "echo", (1, "yeah")) + cumin.session.call_method(completion, obj, "echo", (1, "Hello!")) try: - while True: - # print_threads() - - sleep(5) + completed.wait() finally: cumin.stop() Modified: mgmt/newdata/cumin/python/cumin/session.py =================================================================== --- mgmt/newdata/cumin/python/cumin/session.py 2010-06-15 18:47:09 UTC (rev 4027) +++ mgmt/newdata/cumin/python/cumin/session.py 2010-06-16 18:20:33 UTC (rev 4028) @@ -12,6 +12,7 @@ self.qmf_session = None self.qmf_brokers = list() + self.qmf_agents = dict() # int seq => callable self.outstanding_method_calls = dict() @@ -25,9 +26,6 @@ qmf_broker = self.qmf_session.addBroker(uri) - name = qmf_broker.thread.__class__.__name__ - qmf_broker.thread.name = "%s(%s)" % (name, uri) - self.qmf_brokers.append(qmf_broker) def check(self): @@ -55,11 +53,24 @@ for qmf_broker in self.qmf_brokers: self.qmf_session.delBroker(qmf_broker) + def get_agent(self, agent_id): + self.lock.acquire() + try: + return self.qmf_agents.get(agent_id) + finally: + self.lock.release() + def call_method(self, callback, obj, name, args): assert isinstance(obj, RosemaryObject) - agent = self.qmf_session._getAgentForAgentAddr(obj._qmf_agent_id) + for i in range(10): + agent = self.get_agent(obj._qmf_agent_id) + if agent: + break + + sleep(1) + if not agent: raise Exception("Agent '%s' is unknown" % obj._qmf_agent_id) @@ -92,18 +103,30 @@ def __init__(self, session): self.session = session + def newPackage(self, name): + log.info("New package %s", name) + + def newClass(self, kind, classKey): + log.info("New class %s", classKey) + def newAgent(self, qmf_agent): log.info("New agent %s", qmf_agent) + self.session.lock.acquire() + try: + self.session.qmf_agents[qmf_agent.getAgentBank()] = qmf_agent + finally: + self.session.lock.release() + def delAgent(self, qmf_agent): log.info("Deleting agent %s", qmf_agent) - def newPackage(self, name): - log.info("New package %s", name) + self.session.lock.acquire() + try: + del self.session.qmf_agents[qmf_agent.getAgentBank()] + finally: + self.session.lock.release() - def newClass(self, kind, classKey): - log.info("New class %s", classKey) - def methodResponse(self, broker, seq, response): log.info("Method response for request %i received from %s", seq, broker)