Author: ritchiem Date: Wed Oct 28 15:42:51 2009 New Revision: 830642 URL: http://svn.apache.org/viewvc?rev=830642&view=rev Log: Removed references to broker and agent bank from API, replaced with the more generic (and forward compatible) "key".
In the Ruby binding, ensured that ruby objects reference their own copies of the wrapped c++ objects to protect from problems when the c++ objects are deleted out from under the wrappers. Added agent discriminator to the console::objects method. Modified: qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/python/qmf.py qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/ruby/qmf.rb qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console.rb qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb qpid/branches/0.5.x-dev/qpid/cpp/include/qmf/engine/Console.h qpid/branches/0.5.x-dev/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp Modified: qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/python/qmf.py URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/python/qmf.py?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/python/qmf.py (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/python/qmf.py Wed Oct 28 15:42:51 2009 @@ -535,7 +535,7 @@ return self.impl.isDeleted() - def index(self): pass + def key(self): pass @@ -545,6 +545,7 @@ self.impl = impl else: self.impl = qmfengine.ObjectId() + self.agent_key = "%d.%d" % (self.impl.getBrokerBank(), self.impl.getAgentBank()) def object_num_high(self): @@ -555,13 +556,8 @@ return self.impl.getObjectNumLo() - def broker_bank(self): - return self.impl.getBrokerBank() - - - def agent_bank(self): - return self.impl.getAgentBank() - + def agent_key(self): + self.agent_key def __eq__(self, other): if not isinstance(other, self.__class__): return False @@ -1076,13 +1072,18 @@ def 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 "_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) @@ -1097,7 +1098,7 @@ self._sync_count = 1 self._sync_result = [] broker = self._broker_list[0] - broker.send_query(query.impl, None) + broker.send_query(query.impl, None, agent) self._cv.wait(timeout) if self._sync_count == 1: raise Exception("Timed out: waiting for query response") @@ -1193,38 +1194,40 @@ valid = self.impl.getEvent(self._event) while valid: count += 1 - 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)) - + 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 @@ -1236,19 +1239,15 @@ def __init__(self, impl, broker): self.impl = impl self.broker = broker + self.key = "%d.%d" % (self.impl.getBrokerBank(), self.impl.getAgentBank()) def label(self): return self.impl.getLabel() - def broker_bank(self): - return self.impl.getBrokerBank() - - - def agent_bank(self): - return self.impl.getAgentBank() - + def key(self): + return self.key class Broker(ConnectionHandler): @@ -1298,8 +1297,11 @@ self._cv.release() - def send_query(self, query, ctx): - self.impl.sendQuery(query, ctx) + 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() Modified: qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/ruby/qmf.rb URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/ruby/qmf.rb?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/ruby/qmf.rb (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/ruby/qmf.rb Wed Oct 28 15:42:51 2009 @@ -448,18 +448,19 @@ @impl.isDeleted end - def index() + def key() end end class ObjectId - attr_reader :impl + attr_reader :impl, :agent_key def initialize(impl=nil) if impl @impl = impl else @impl = Qmfengine::ObjectId.new end + @agent_key = "#...@impl.getbrokerbank}.#{@impl.getAgentBank}" end def object_num_high @@ -470,14 +471,6 @@ @impl.getObjectNumLo end - def broker_bank - @impl.getBrokerBank - end - - def agent_bank - @impl.getAgentBank - end - def ==(other) return (@impl.getObjectNumHi == other.impl.getObjectNumHi) && (@impl.getObjectNumLo == other.impl.getObjectNumLo) @@ -748,7 +741,7 @@ class SchemaClassKey attr_reader :impl def initialize(i) - @impl = i + @impl = Qmfengine::SchemaClassKey.new(i) end def package_name @@ -956,6 +949,7 @@ def objects(query, kwargs = {}) timeout = 30 + agent = nil kwargs.merge!(query) if query.class == Hash if kwargs.include?(:timeout) @@ -963,6 +957,11 @@ kwargs.delete(:timeout) end + if kwargs.include?(:agent) + agent = kwargs[:agent] + kwargs.delete(:agent) + end + query = Query.new(kwargs) if query.class == Hash @select = [] @@ -975,7 +974,7 @@ @sync_result = [] broker = nil synchronize { broker = @broker_list[0] } - broker.send_query(query.impl, nil) + broker.send_query(query.impl, nil, agent) unless @cv.wait(timeout) { @sync_count == 0 } raise "Timed out waiting for response" end @@ -1045,21 +1044,25 @@ valid = @impl.getEvent(@event) while valid count += 1 - case @event.kind - when Qmfengine::ConsoleEvent::AGENT_ADDED - @handler.agent_added(AgentProxy.new(@event.agent, nil)) if @handler - when Qmfengine::ConsoleEvent::AGENT_DELETED - @handler.agent_deleted(AgentProxy.new(@event.agent, nil)) if @handler - when Qmfengine::ConsoleEvent::NEW_PACKAGE - @handler.new_package(@event.name) if @handler - when Qmfengine::ConsoleEvent::NEW_CLASS - @handler.new_class(SchemaClassKey.new(@event.classKey)) if @handler - when Qmfengine::ConsoleEvent::OBJECT_UPDATE - @handler.object_update(ConsoleObject.new(nil, :impl => @event.object), @event.hasProps, @event.hasStats) if @handler - when Qmfengine::ConsoleEvent::EVENT_RECEIVED - when Qmfengine::ConsoleEvent::AGENT_HEARTBEAT - @handler.agent_heartbeat(AgentProxy.new(@event.agent, nil), @event.timestamp) if @handler - when Qmfengine::ConsoleEvent::METHOD_RESPONSE + begin + case @event.kind + when Qmfengine::ConsoleEvent::AGENT_ADDED + @handler.agent_added(AgentProxy.new(@event.agent, nil)) if @handler + when Qmfengine::ConsoleEvent::AGENT_DELETED + @handler.agent_deleted(AgentProxy.new(@event.agent, nil)) if @handler + when Qmfengine::ConsoleEvent::NEW_PACKAGE + @handler.new_package(@event.name) if @handler + when Qmfengine::ConsoleEvent::NEW_CLASS + @handler.new_class(SchemaClassKey.new(@event.classKey)) if @handler + when Qmfengine::ConsoleEvent::OBJECT_UPDATE + @handler.object_update(ConsoleObject.new(nil, :impl => @event.object), @event.hasProps, @event.hasStats) if @handler + when Qmfengine::ConsoleEvent::EVENT_RECEIVED + when Qmfengine::ConsoleEvent::AGENT_HEARTBEAT + @handler.agent_heartbeat(AgentProxy.new(@event.agent, nil), @event.timestamp) if @handler + when Qmfengine::ConsoleEvent::METHOD_RESPONSE + end + rescue + puts "Exception caught in callback thread: #{$!}" end @impl.popEvent valid = @impl.getEvent(@event) @@ -1069,23 +1072,13 @@ end class AgentProxy - attr_reader :broker + attr_reader :impl, :broker, :label, :key def initialize(impl, broker) - @impl = impl + @impl = Qmfengine::AgentProxy.new(impl) @broker = broker - end - - def label - @impl.getLabel - end - - def broker_bank - @impl.getBrokerBank - end - - def agent_bank - @impl.getAgentBank + @label = @impl.getLabel + @key = "#...@impl.getbrokerbank}.#{@impl.getAgentBank}" end end @@ -1130,8 +1123,9 @@ end end - def send_query(query, ctx) - @impl.sendQuery(query, ctx) + def send_query(query, ctx, agent) + agent_impl = agent.impl if agent + @impl.sendQuery(query, ctx, agent_impl) @conn.kick end Modified: qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console.rb URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console.rb?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console.rb (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console.rb Wed Oct 28 15:42:51 2009 @@ -25,7 +25,7 @@ class App < Qmf::ConsoleHandler def agent_added(agent) - puts "AgentAdded: #{agent.label} broker=#{agent.broker_bank} agent=#{agent.agent_bank}" + puts "AgentAdded: label=#{agent.label} key=#{agent.key}" end def agent_deleted(agent) @@ -42,8 +42,7 @@ def object_update(object, hasProps, hasStats) puts "ObjectUpdate: #{object.object_class.class_name} props=#{hasProps} stats=#{hasStats}" - puts " broker-bank=#{object.object_id.broker_bank}" - puts " agent-bank=#{object.object_id.agent_bank}" + puts " agent-key=#{object.object_id.agent_key}" puts " package=#{object.object_class.package_name}" end Modified: qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/bindings/qmf/tests/ruby_console_test.rb Wed Oct 28 15:42:51 2009 @@ -201,6 +201,19 @@ end + def test_D_get_with_agent + agents = @qmfc.agents + agents.each do |agent| + if agent.label == "qmfa" + parent = @qmfc.object(:class => "parent", :agent => agent) + assert(parent, "Number of parent objects") + return + end + end + + fail("Didn't find a non-broker agent") + end + end app = ConsoleTest.new Modified: qpid/branches/0.5.x-dev/qpid/cpp/include/qmf/engine/Console.h URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/include/qmf/engine/Console.h?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/include/qmf/engine/Console.h (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/include/qmf/engine/Console.h Wed Oct 28 15:42:51 2009 @@ -134,6 +134,7 @@ */ class AgentProxy { public: + AgentProxy(const AgentProxy& from); ~AgentProxy(); const char* getLabel() const; uint32_t getBrokerBank() const; Modified: qpid/branches/0.5.x-dev/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp URL: http://svn.apache.org/viewvc/qpid/branches/0.5.x-dev/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp?rev=830642&r1=830641&r2=830642&view=diff ============================================================================== --- qpid/branches/0.5.x-dev/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp (original) +++ qpid/branches/0.5.x-dev/qpid/cpp/src/qmf/engine/BrokerProxyImpl.cpp Wed Oct 28 15:42:51 2009 @@ -741,6 +741,7 @@ //================================================================== AgentProxy::AgentProxy(AgentProxyImpl* i) : impl(i) {} +AgentProxy::AgentProxy(const AgentProxy& from) : impl(new AgentProxyImpl(*(from.impl))) {} AgentProxy::~AgentProxy() { delete impl; } const char* AgentProxy::getLabel() const { return impl->getLabel().c_str(); } uint32_t AgentProxy::getBrokerBank() const { return impl->getBrokerBank(); } --------------------------------------------------------------------- Apache Qpid - AMQP Messaging Implementation Project: http://qpid.apache.org Use/Interact: mailto:commits-subscr...@qpid.apache.org