From 7753a6839ac0abb282e2d4372f7a4199243cf4ff Mon Sep 17 00:00:00 2001 From: Ted Ross Date: Wed, 19 Aug 2009 18:31:31 +0000 Subject: Introduce the public includes for the QMF interfaces. Rename Agent to AgentEngine to differentiate the API from the underlying engine. Note that some of these public headers will overlap with the emerging "messaging" API (notably Connection.h and ConnectionSettings.h). It is desirable that these components of the API become common between "messaging" and "qmf". As such, once the differences are reconciled, they will most likely be removed from the qmf space and placed in the messaging space. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@805916 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/bindings/qmf/qmfengine.i | 4 +- qpid/cpp/bindings/qmf/ruby/Makefile.am | 2 +- qpid/cpp/bindings/qmf/ruby/qmf.rb | 22 +- qpid/cpp/include/qmf/Agent.h | 283 +++++++++ qpid/cpp/include/qmf/AgentObject.h | 93 +++ qpid/cpp/include/qmf/Connection.h | 118 ++++ qpid/cpp/include/qmf/ConnectionSettings.h | 135 +++++ qpid/cpp/include/qmf/QmfImportExport.h | 33 + qpid/cpp/src/qmf.mk | 22 +- qpid/cpp/src/qmf/Agent.cpp | 958 ------------------------------ qpid/cpp/src/qmf/Agent.h | 206 ------- qpid/cpp/src/qmf/AgentEngine.cpp | 958 ++++++++++++++++++++++++++++++ qpid/cpp/src/qmf/AgentEngine.h | 207 +++++++ qpid/cpp/src/qmf/Console.h | 82 --- qpid/cpp/src/qmf/ConsoleEngine.h | 83 +++ qpid/cpp/src/qmf/Object.h | 2 +- 16 files changed, 1937 insertions(+), 1271 deletions(-) create mode 100644 qpid/cpp/include/qmf/Agent.h create mode 100644 qpid/cpp/include/qmf/AgentObject.h create mode 100644 qpid/cpp/include/qmf/Connection.h create mode 100644 qpid/cpp/include/qmf/ConnectionSettings.h create mode 100644 qpid/cpp/include/qmf/QmfImportExport.h delete mode 100644 qpid/cpp/src/qmf/Agent.cpp delete mode 100644 qpid/cpp/src/qmf/Agent.h create mode 100644 qpid/cpp/src/qmf/AgentEngine.cpp create mode 100644 qpid/cpp/src/qmf/AgentEngine.h delete mode 100644 qpid/cpp/src/qmf/Console.h create mode 100644 qpid/cpp/src/qmf/ConsoleEngine.h (limited to 'qpid/cpp') diff --git a/qpid/cpp/bindings/qmf/qmfengine.i b/qpid/cpp/bindings/qmf/qmfengine.i index 3c67d92031..e1a4c53bec 100644 --- a/qpid/cpp/bindings/qmf/qmfengine.i +++ b/qpid/cpp/bindings/qmf/qmfengine.i @@ -19,7 +19,7 @@ %{ -#include "Agent.h" +#include "AgentEngine.h" #include %} @@ -27,7 +27,7 @@ %include %include -%include +%include %include %include %include diff --git a/qpid/cpp/bindings/qmf/ruby/Makefile.am b/qpid/cpp/bindings/qmf/ruby/Makefile.am index 532fdb6875..1dc08f646a 100644 --- a/qpid/cpp/bindings/qmf/ruby/Makefile.am +++ b/qpid/cpp/bindings/qmf/ruby/Makefile.am @@ -36,7 +36,7 @@ rubylibarchdir = $(RUBY_LIB_ARCH) rubylibarch_LTLIBRARIES = qmfengine.la qmfengine_la_LDFLAGS = -avoid-version -module -shrext ".$(RUBY_DLEXT)" -qmfengine_la_LIBADD = $(RUBY_LIBS) -L$(top_builddir)/src/.libs -lqpidclient $(top_builddir)/src/libqmfcommon.la +qmfengine_la_LIBADD = $(RUBY_LIBS) -L$(top_builddir)/src/.libs -lqpidclient $(top_builddir)/src/libqmfagent.la qmfengine_la_CXXFLAGS = $(INCLUDES) -I$(RUBY_INC) -I$(RUBY_INC_ARCH) nodist_qmfengine_la_SOURCES = qmfengine.cpp diff --git a/qpid/cpp/bindings/qmf/ruby/qmf.rb b/qpid/cpp/bindings/qmf/ruby/qmf.rb index 7ee447c675..35a76c490a 100644 --- a/qpid/cpp/bindings/qmf/ruby/qmf.rb +++ b/qpid/cpp/bindings/qmf/ruby/qmf.rb @@ -33,11 +33,11 @@ module Qmf class ConnectionSettings < Qmfengine::ConnectionSettings end - class ConnectionEvent + class ConnectionHandler def conn_event_connected(); end def conn_event_disconnected(error); end - def conn_event_session_closed(context, error); end - def conn_event_recv(context, message); end + def sess_event_session_closed(context, error); end + def sess_event_recv(context, message); end end class Query @@ -63,15 +63,10 @@ module Qmf end end - class AgentHandler - def get_query(context, query, userId); end - def method_call(context, name, object_id, args, userId); end - end - class Connection attr_reader :impl - def initialize(settings, event_handler = nil, delay_min = 1, delay_max = 128, delay_factor = 2) + def initialize(settings, delay_min = 1, delay_max = 128, delay_factor = 2) @impl = Qmfengine::ResilientConnection.new(settings, delay_min, delay_max, delay_factor) @sockEngine, @sock = Socket::socketpair(Socket::PF_UNIX, Socket::SOCK_STREAM, 0) @impl.setNotifyFd(@sockEngine.fileno) @@ -232,7 +227,12 @@ module Qmf end end - class Agent + class AgentHandler + def get_query(context, query, userId); end + def method_call(context, name, object_id, args, userId); end + end + + class Agent < ConnectionHandler def initialize(handler, label="") if label == "" @agentLabel = "rb-%s.%d" % [Socket.gethostname, Process::pid] @@ -241,7 +241,7 @@ module Qmf end @conn = nil @handler = handler - @impl = Qmfengine::Agent.new(@agentLabel) + @impl = Qmfengine::AgentEngine.new(@agentLabel) @event = Qmfengine::AgentEvent.new @xmtMessage = Qmfengine::Message.new end diff --git a/qpid/cpp/include/qmf/Agent.h b/qpid/cpp/include/qmf/Agent.h new file mode 100644 index 0000000000..2ba639ca37 --- /dev/null +++ b/qpid/cpp/include/qmf/Agent.h @@ -0,0 +1,283 @@ +#ifndef _QmfAgent_ +#define _QmfAgent_ + +/* + * 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. + */ + +#include "qmf/QmfImportExport.h" + +namespace qmf { + + class AgentImpl; + class Connection; + class ObjectId; + class AgentObject; + class Value; + class Event; + class SchemaObjectClass; + + /** + * AgentListener is used by agents that select the internalStore=false option (see Agent + * constructor) or by agents that wish to provide access control for queries and methods. + */ + class AgentListener { + QMF_EXTERN virtual ~AgentListener(); + + /** + * allowQuery is called before a query operation is executed. If true is returned + * by this function, the query will proceed. If false is returned, the query will + * be forbidden. + * + * @param q The query being requested. + * @param userId The authenticated identity of the user requesting the query. + */ + virtual bool allowQuery(const Query& q, const char* userId); + + /** + * allowMethod is called before a method call is executed. If true is returned + * by this function, the method call will proceed. If false is returned, the method + * call will be forbidden. + * + * @param name The name of the method being called. + * @param args A value object (of type "map") that contains both input and output arguments. + * @param oid The objectId that identifies the instance of the object being called. + * @param cls The Schema describing the object being called. + * @param userId The authenticated identity of the requesting user. + */ + virtual bool allowMethod(const char* name, const Value& args, const ObjectId& oid, + const SchemaObjectClass& cls, const char* userId); + + /** + * query is called when the agent receives a query request. The handler must invoke + * Agent::queryResponse zero or more times (using the supplied context) followed by + * a single invocation of Agent::queryComplete. These calls do not need to be made + * synchronously in the context of this function. They may occur before or after this + * function returns. + * + * This function will only be invoked if internalStore=false in the Agent's constructor. + * + * @param context A context value to use in resulting calls to queryResponse and quertComplete. + * @param q The query requested by the console. + * @param userId the authenticated identity of the user requesting the query. + */ + virtual void query(uint32_t context, const Query& q, const char* userId); + + /** + * syncStart is called when a console requests a standing query. This function must + * behave exactly like AgentListener::query (i.e. send zero or more responses followed + * by a queryComplete) except it then remembers the context and the query and makes + * subsequent queryResponse calls whenever appropriate according the the query. + * + * The standing query shall stay in effect until syncStop is called with the same context + * value or until a specified period of time elapses without receiving a syncTouch for the + * context. + * + * This function will only be invoked if internalStore=false in the Agent's constructor. + * + * @param context A context value to use in resulting calls to queryResponse and queryComplete. + * @param q The query requested by the console. + * @param userId the authenticated identity of the user requesting the query. + */ + virtual void syncStart(uint32_t context, const Query& q, const char* userId); + + /** + * syncTouch is called when the console that requested a standing query refreshes its + * interest in the query. The console must periodically "touch" a standing query to keep + * it alive. This prevents standing queries from accumulating when the console disconnects + * before it can stop the query. + * + * This function will only be invoked if internalStore=false in the Agent's constructor. + * + * @param context The context supplied in a previous call to syncStart. + * @param userId The authenticated identity of the requesting user. + */ + virtual void syncTouch(uint32_t context, const char* userId); + + /** + * syncStop is called when the console that requested a standing query no longer wishes to + * receive data associated with that query. The application shall stop processing this + * query and shall remove its record of the context value. + * + * This function will only be invoked if internalStore=false in the Agent's constructor. + * + * @param context The context supplied in a previous call to syncStart. + * @param userId The authenticated identity of the requesting user. + */ + virtual void syncStop(uint32_t context, const char* userId); + + /** + * methodCall is called when a console invokes a method on a QMF object. The application + * must call Agent::methodResponse once in response to this function. The response does + * not need to be called synchronously in the context of this function. It may be called + * before or after this function returns. + * + * This function will only be invoked if internalStore=false in the Agent's constructor. + * + * @param context A context value to use in resulting call to methodResponse. + * @param name The name of the method being called. + * @param args A value object (of type "map") that contains both input and output arguments. + * @param oid The objectId that identifies the instance of the object being called. + * @param cls The Schema describing the object being called. + * @param userId The authenticated identity of the requesting user. + */ + virtual void methodCall(uint32_t context, const char* name, Value& args, + const ObjectId& oid, const SchemaObjectClass& cls, const char* userId); + }; + + /** + * The Agent class is the QMF Agent portal. It should be instantiated once and associated with a + * Connection (setConnection) to connect an agent to the QMF infrastructure. + */ + class Agent { + public: + /** + * Create an instance of the Agent class. + * + * @param label An optional string label that can be used to identify the agent. + * + * @param internalStore If true, objects shall be tracked internally by the agent. + * If false, the user of the agent must track the objects. + * If the agent is tracking the objects, queries and syncs are handled by + * the agent. The only involvement the user has is to optionally authorize + * individual operations. If the user is tracking the objects, the user code + * must implement queries and syncs (standing queries). + * + * @param listener A pointer to a class that implements the AgentListener interface. + * This must be supplied if any of the following conditions are true: + * - The agent model contains methods + * - The user wishes to individually authorize query and sync operations. + * - internalStore = false + */ + QMF_EXTERN Agent(char* label="qmfa", bool internalStore=true, AgentListener* listener=0); + + /** + * Destroy an instance of the Agent class. + */ + QMF_EXTERN ~Agent(); + + /** + * Set the persistent store file. This file, if specified, is used to store state information + * about the Agent. For example, if object-ids must be persistent across restarts of the Agent + * program, this file path must be supplied. + * + * @param path Full path to a file that is both writable and readable by the Agent program. + */ + QMF_EXTERN void setStoreDir(const char* path); + + /** + * Provide a connection (to a Qpid broker) over which the agent can communicate. + * + * @param conn Pointer to a Connection object. + */ + QMF_EXTERN void setConnection(Connection* conn); + + /** + * Register a class schema (object or event) with the agent. The agent must have a registered + * schema for an object class or an event class before it can handle objects or events of that + * class. + * + * @param cls Pointer to the schema structure describing the class. + */ + QMF_EXTERN void registerClass(SchemaObjectClass* cls); + QMF_EXTERN void registerClass(SchemaEventClass* cls); + + /** + * Add an object to the agent (for internal storage mode only). + * + * @param obj Reference to the object to be managed by the agent. + * + * @param persistent Iff true, the object ID assigned to the object shall indicate persistence + * (i.e. the object ID shall be the same across restarts of the agent program). + * + * @param oid 64-bit value for the oid (if zero, the agent will assign the value). + * + * @param oidLo 32-bit value for the lower 32-bits of the oid. + * + * @param oidHi 32-bit value for the upper 32-bits of the oid. + */ + QMF_EXTERN const ObjectId* addObject(AgentObject& obj, bool persistent=false, uint64_t oid=0); + QMF_EXTERN const ObjectId* addObject(AgentObject& obj, bool persistent, uint32_t oidLo, uint32_t oidHi); + + /** + * Allocate an object ID for an object (for external storage mode only). + * + * @param persistent Iff true, the object ID allocated shall indicate persistence + * (i.e. the object ID shall be the same across restarts of the agent program). + * + * @param oid 64-bit value for the oid (if zero, the agent will assign the value). + * + * @param oidLo 32-bit value for the lower 32-bits of the oid. + * + * @param oidHi 32-bit value for the upper 32-bits of the oid. + */ + QMF_EXTERN const ObjectId* allocObjectId(bool persistent=false, uint64_t oid=0); + QMF_EXTERN const ObjectId* allocObjectId(bool persistent, uint32_t oidLo, uint32_t oidHi); + + /** + * Raise a QMF event. + * + * @param event Reference to an event object to be raised to the QMF infrastructure. + */ + QMF_EXTERN void raiseEvent(Event& event); + + /** + * Provide a response to a query (for external storage mode only). + * + * @param context The context value supplied in the query (via the AgentListener interface). + * + * @param object A reference to the agent that matched the query criteria. + * + * @param prop If true, transmit the property attributes of this object. + * + * @param stat If true, transmit the statistic attributes of this object. + */ + QMF_EXTERN void queryResponse(uint32_t context, AgentObject& object, bool prop = true, bool stat = true); + + /** + * Indicate that a query (or the initial dump of a sync) is complete (for external storage mode only). + * + * @param context The context value supplied in the query/sync (via the AgentListener interface). + */ + QMF_EXTERN void queryComplete(uint32_t context); + + /** + * Provide the response to a method call. + * + * @param context The context value supplied in the method request (via the AgentListener interface). + * + * @param args The argument list from the method call. Must include the output arguments (may include + * the input arguments). + * + * @param status Numerical return status: zero indicates no error, non-zero indicates error. + * + * @param exception Pointer to an exception value. If status is non-zero, the exception value is + * sent to the caller. It is optional (i.e. leave the pointer as 0), or may be + * set to any legal value. A string may be supplied, but an unmanaged object of + * any schema may also be passed. + */ + QMF_EXTERN void methodResponse(uint32_t context, const Value& args, uint32_t status=0, + const Value* exception=0); + + private: + AgentImpl* impl; + }; + +} + +#endif diff --git a/qpid/cpp/include/qmf/AgentObject.h b/qpid/cpp/include/qmf/AgentObject.h new file mode 100644 index 0000000000..d6073bca3c --- /dev/null +++ b/qpid/cpp/include/qmf/AgentObject.h @@ -0,0 +1,93 @@ +#ifndef _QmfAgentObject_ +#define _QmfAgentObject_ + +/* + * 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. + */ + +#include "qmf/QmfImportExport.h" + +namespace qmf { + + class AgentObjectImpl; + class SchemaObjectClass; + class ObjectId; + class Value; + class Agent; + + /** + * AgentObject is an extension of Object with agent-specific methods added. + */ + class AgentObject : public Object { + public: + /** + * Create a new Object of a specific type. + * + * @param type Pointer to the schema class to use as a type for this object. + */ + QMF_EXTERN AgentObject(const SchemaObjectClass* type); + + /** + * Schedule this object for deletion. Agent objects should never be directly + * destroyed, rather this method should be called and all pointers to this + * object dropped. The agent will clean up and properly delete the object at + * the appropraite time. + */ + QMF_EXTERN void destroy(); + + /** + * Set the object ID for this object if it is to be managed by the agent. + * + * @param oid The new object ID for the managed object. + */ + QMF_EXTERN void setObjectId(ObjectId& oid); + + /** + * Handler for invoked method calls. This will only be called for objects that + * are being managed and stored by an agent (see internalStore argument in Agent::Agent). + * If this function is not overridden in a child class, the default implementation will + * cause AgentListener::methodCall to be invoked in the application program. + * + * If this function is overridden in a sub-class, the implementation must perform + * the actions associated with the method call (i.e. implement the method). Once the + * method execution is complete, it must call Agent::methodResponse with the result + * of the method execution. Agent::methodResponse does not need to be called + * synchronously in the context of this function call. It may be called at a later + * time from a different thread. + * + * @param context Context supplied by the agent and required to be passed in the + * call to Agent::methodResponse + * + * @param name The name of the method. + * + * @param args A Value (of type map) that contains the input and output arguments. + * + * @param userId The authenticated identity of the user who invoked the method. + */ + QMF_EXTERN virtual void methodInvoked(uint32_t context, const char* name, Value& args, + const char* userId); + private: + friend class Agent; + virtual ~AgentObject(); + void setAgent(Agent* agent); + AgentObjectImpl* impl; + }; + +} + +#endif diff --git a/qpid/cpp/include/qmf/Connection.h b/qpid/cpp/include/qmf/Connection.h new file mode 100644 index 0000000000..dbe8e97cfe --- /dev/null +++ b/qpid/cpp/include/qmf/Connection.h @@ -0,0 +1,118 @@ +#ifndef _QmfConnection_ +#define _QmfConnection_ + +/* + * 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. + */ + +#include "qmf/QmfImportExport.h" +#include "qmf/ConnectionSettings.h" + +namespace qmf { + + enum ConnectionState { + CONNECTION_UP = 1, + CONNECTION_DOWN = 2 + }; + + /** + * Implement a subclass of ConnectionListener and provide it with the + * Connection constructor to receive notification of changes in the + * connection state. + * + * \ingroup qmfapi + */ + class ConnectionListener { + QMF_EXTERN virtual ~ConnectionListener(); + + /** + * Called each time the state of the connection changes. + */ + virtual void newState(ConnectionState state); + + /** + * Called if the connection requires input from an interactive client. + * + * @param prompt Text of the prompt - describes what information is required. + * @param answer The interactive user input. + * @param answerLen on Input - the maximum number of bytes that can be copied to answer. + * on Output - the number of bytes copied to answer. + */ + virtual void interactivePrompt(const char* prompt, char* answer, uint32_t answerLen); + }; + + class ConnectionImpl; + + /** + * The Connection class represents a connection to a QPID broker that can + * be used by agents and consoles, possibly multiple at the same time. + * + * \ingroup qmfapi + */ + class Connection { + public: + + /** + * Creates a connection object and begins the process of attempting to + * connect to the QPID broker. + * + * @param settings The settings that control how the connection is set + * up. + * + * @param listener An optional pointer to a subclass of + * ConnectionListener to receive notifications of events related to + * this connection. + */ + QMF_EXTERN Connection(const ConnectionSettings& settings, + const ConnectionListener* listener = 0); + + /** + * Destroys a connection, causing the connection to be closed. + */ + QMF_EXTERN ~Connection(); + + /** + * Set the administrative state of the connection (enabled or disabled). + * + * @param enabled True => enable connection, False => disable connection + */ + QMF_EXTERN void setAdminState(bool enabled); + + /** + * Return the current operational state of the connection (up or down). + * + * @return the current connection state. + */ + QMF_EXTERN ConnectionState getOperState() const; + + /** + * Get the error message from the last failure to connect. + * + * @return Null-terminated string containing the error message. + */ + QMF_EXTERN const char* getLastError() const; + + private: + friend class AgentImpl; + friend class ConsoleImpl; + ConnectionImpl* impl; + }; + +} + +#endif diff --git a/qpid/cpp/include/qmf/ConnectionSettings.h b/qpid/cpp/include/qmf/ConnectionSettings.h new file mode 100644 index 0000000000..b5f3be7eef --- /dev/null +++ b/qpid/cpp/include/qmf/ConnectionSettings.h @@ -0,0 +1,135 @@ +#ifndef _QmfConnectionSettings_ +#define _QmfConnectionSettings_ + +/* + * 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. + */ + +#include "qmf/QmfImportExport.h" + +namespace qmf { + + class ConnectionSettingsImpl; + class Value; + + /** + * Settings for AMQP connections to the broker. + */ + class ConnectionSettings { + public: + + /** + * Create a set of default connection settings. + * + * If no further attributes are set, the settings will cause a connection to be made to + * the default broker (on localhost or at a host/port supplied by service discovery) and + * authentication will be the best-available (GSSAPI/Kerberos, Anonymous, Plain with prompts + * for username and password). + */ + QMF_EXTERN ConnectionSettings(); + + /** + * Create a set of connection settings by URL. + * + * @param url Universal resource locator describing the broker address and additional attributes. + * + * The URL is of the form: + * amqp[s]://host[:port][?key=value[&key=value]*] + * + * For example: + * amqp://localhost + * amqp://broker?transport=rdma&authmech=GSSAPI&authservice=qpidd + * amqps://broker?authmech=PLAIN&authuser=guest&authpass=guest + */ + QMF_EXTERN ConnectionSettings(const char* url); + + /** + * Destroy the connection settings object. + */ + QMF_EXTERN ~ConnectionSettings(); + + /** + * Set an attribute to control connection setup. + * + * @param key A null-terminated string that is an attribute name. + * + * @param value Reference to a value to be stored as the attribute. The type of the value + * is specific to the key. + */ + QMF_EXTERN void setAttr(const char* key, const Value& value); + + /** + * Get the value of an attribute. + * + * @param key A null-terminated attribute name. + */ + QMF_EXTERN const Value& getAttr(const char* key) const; + + /** + * Get the attribute string (the portion of the URL following the '?') for the settings. + * + * @return A pointer to the attribute string. If the content of this string needs to be + * available beyond the scope of the calling function, it should be copied. The + * returned pointer may become invalid if the set of attributes is changed. + */ + QMF_EXTERN const char* getAttrString() const; + + /** + * Shortcuts for setting the transport for the connection. + * + * @param port The port value for the connection address. + */ + QMF_EXTERN void transportTcp(uint16_t port = 5672); + QMF_EXTERN void transportSsl(uint16_t port = 5671); + QMF_EXTERN void transportRdma(uint16_t port = 5672); + + /** + * Shortcuts for setting authentication mechanisms. + * + * @param username Null-terminated authentication user name. + * + * @param password Null-terminated authentication password. + * + * @param serviceName Null-terminated GSSAPI service name (Kerberos service principal) + * + * @param minSsf Minimum security factor for connections. 0 = encryption not required. + * + * @param maxSsf Maximum security factor for connections. 0 = encryption not permitted. + */ + QMF_EXTERN void authAnonymous(const char* username = 0); + QMF_EXTERN void authPlain(const char* username = 0, const char* password = 0); + QMF_EXTERN void authGssapi(const char* serviceName, uint32_t minSsf = 0, uint32_t maxSsf = 256); + + /** + * Shortcut for setting connection retry attributes. + * + * @param delayMin Minimum delay (in seconds) between connection attempts. + * + * @param delaxMax Maximum delay (in seconds) between connection attempts. + * + * @param delayFactor Factor to multiply the delay by between failed connection attempts. + */ + QMF_EXTERN void setRetry(int delayMin = 1, int delayMax = 128, int delayFactor = 2); + + private: + ConnectionSettingsImpl* impl; + }; + +} + +#endif diff --git a/qpid/cpp/include/qmf/QmfImportExport.h b/qpid/cpp/include/qmf/QmfImportExport.h new file mode 100644 index 0000000000..8353a3cc16 --- /dev/null +++ b/qpid/cpp/include/qmf/QmfImportExport.h @@ -0,0 +1,33 @@ +#ifndef QMF_IMPORT_EXPORT_H +#define QMF_IMPORT_EXPORT_H + +/* + * 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. + */ + +#if defined(WIN32) && !defined(QPID_DECLARE_STATIC) +# if defined(QMF_EXPORT) || defined (qmf_EXPORTS) +# define QMF_EXTERN __declspec(dllexport) +# else +# define QMF_EXTERN __declspec(dllimport) +# endif +#else +# define QMF_EXTERN +#endif + +#endif diff --git a/qpid/cpp/src/qmf.mk b/qpid/cpp/src/qmf.mk index 62393cdcfb..8ccc1f4a95 100644 --- a/qpid/cpp/src/qmf.mk +++ b/qpid/cpp/src/qmf.mk @@ -18,7 +18,7 @@ # # -# qmf agent library makefile fragment, to be included in Makefile.am +# qmf library makefile fragment, to be included in Makefile.am # lib_LTLIBRARIES += \ libqmfcommon.la \ @@ -27,13 +27,15 @@ lib_LTLIBRARIES += \ # Public header files nobase_include_HEADERS += \ ../include/qpid/agent/ManagementAgent.h \ - ../include/qpid/agent/QmfAgentImportExport.h - + ../include/qpid/agent/QmfAgentImportExport.h \ + ../include/qmf/Agent.h \ + ../include/qmf/Connection.h \ + ../include/qmf/QmfImportExport.h \ + ../include/qmf/ConnectionSettings.h \ + ../include/qmf/AgentObject.h libqmfcommon_la_SOURCES = \ - qmf/Agent.cpp \ - qmf/Agent.h \ - qmf/Console.h \ + qmf/ConsoleEngine.h \ qmf/Event.h \ qmf/Message.h \ qmf/MessageImpl.cpp \ @@ -58,9 +60,9 @@ libqmfcommon_la_SOURCES = \ qmf/ValueImpl.h libqmfagent_la_SOURCES = \ - ../include/qpid/agent/ManagementAgent.h \ + qmf/AgentEngine.cpp \ + qmf/AgentEngine.h \ qpid/agent/ManagementAgentImpl.cpp \ - qpid/agent/ManagementAgentImpl.h \ - qmf/Agent.cpp + qpid/agent/ManagementAgentImpl.h -libqmfagent_la_LIBADD = libqpidclient.la +libqmfagent_la_LIBADD = libqpidclient.la libqmfcommon.la diff --git a/qpid/cpp/src/qmf/Agent.cpp b/qpid/cpp/src/qmf/Agent.cpp deleted file mode 100644 index 6d59ae2750..0000000000 --- a/qpid/cpp/src/qmf/Agent.cpp +++ /dev/null @@ -1,958 +0,0 @@ -/* - * 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. - */ - -#include "qmf/Agent.h" -#include "qmf/MessageImpl.h" -#include "qmf/SchemaImpl.h" -#include "qmf/Typecode.h" -#include "qmf/ObjectImpl.h" -#include "qmf/ObjectIdImpl.h" -#include "qmf/QueryImpl.h" -#include "qmf/ValueImpl.h" -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -using namespace std; -using namespace qmf; -using namespace qpid::framing; -using namespace qpid::sys; - -namespace qmf { - - struct AgentEventImpl { - typedef boost::shared_ptr Ptr; - AgentEvent::EventKind kind; - uint32_t sequence; - string authUserId; - string authToken; - string name; - Object* object; - boost::shared_ptr objectId; - Query query; - boost::shared_ptr arguments; - string exchange; - string bindingKey; - SchemaObjectClass* objectClass; - - AgentEventImpl(AgentEvent::EventKind k) : - kind(k), sequence(0), object(0), objectClass(0) {} - ~AgentEventImpl() {} - AgentEvent copy(); - }; - - struct AgentQueryContext { - typedef boost::shared_ptr Ptr; - uint32_t sequence; - string exchange; - string key; - SchemaMethodImpl* schemaMethod; - AgentQueryContext() : schemaMethod(0) {} - }; - - class AgentImpl { - public: - AgentImpl(char* label, bool internalStore); - ~AgentImpl(); - - void setStoreDir(char* path); - void setTransferDir(char* path); - void handleRcvMessage(Message& message); - bool getXmtMessage(Message& item); - void popXmt(); - bool getEvent(AgentEvent& event); - void popEvent(); - void newSession(); - void startProtocol(); - void heartbeat(); - void methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments); - void queryResponse(uint32_t sequence, Object& object, bool prop, bool stat); - void queryComplete(uint32_t sequence); - void registerClass(SchemaObjectClass* cls); - void registerClass(SchemaEventClass* cls); - const ObjectId* addObject(Object& obj, uint64_t persistId); - const ObjectId* allocObjectId(uint64_t persistId); - const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); - void raiseEvent(Event& event); - - private: - Mutex lock; - Mutex addLock; - string label; - string queueName; - string storeDir; - string transferDir; - bool internalStore; - uint64_t nextTransientId; - Uuid systemId; - uint32_t requestedBrokerBank; - uint32_t requestedAgentBank; - uint32_t assignedBrokerBank; - uint32_t assignedAgentBank; - AgentAttachment attachment; - uint16_t bootSequence; - uint64_t nextObjectId; - uint32_t nextContextNum; - deque eventQueue; - deque xmtQueue; - map contextMap; - - static const char* QMF_EXCHANGE; - static const char* DIR_EXCHANGE; - static const char* BROKER_KEY; - static const uint32_t MERR_UNKNOWN_METHOD = 2; - static const uint32_t MERR_UNKNOWN_PACKAGE = 8; - static const uint32_t MERR_UNKNOWN_CLASS = 9; - static const uint32_t MERR_INTERNAL_ERROR = 10; -# define MA_BUFFER_SIZE 65536 - char outputBuffer[MA_BUFFER_SIZE]; - - struct SchemaClassKey { - string name; - uint8_t hash[16]; - SchemaClassKey(const string& n, const uint8_t* h) : name(n) { - memcpy(hash, h, 16); - } - SchemaClassKey(Buffer& buffer) { - buffer.getShortString(name); - buffer.getBin128(hash); - } - string repr() { - return name; - } - }; - - struct SchemaClassKeyComp { - bool operator() (const SchemaClassKey& lhs, const SchemaClassKey& rhs) const - { - if (lhs.name != rhs.name) - return lhs.name < rhs.name; - else - for (int i = 0; i < 16; i++) - if (lhs.hash[i] != rhs.hash[i]) - return lhs.hash[i] < rhs.hash[i]; - return false; - } - }; - - typedef map ObjectClassMap; - typedef map EventClassMap; - - struct ClassMaps { - ObjectClassMap objectClasses; - EventClassMap eventClasses; - }; - - map packages; - - bool checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq); - void encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq = 0); - AgentEventImpl::Ptr eventDeclareQueue(const string& queueName); - AgentEventImpl::Ptr eventBind(const string& exchange, const string& queue, const string& key); - AgentEventImpl::Ptr eventSetupComplete(); - AgentEventImpl::Ptr eventQuery(uint32_t num, const string& userId, const string& package, const string& cls, - boost::shared_ptr oid); - AgentEventImpl::Ptr eventMethod(uint32_t num, const string& userId, const string& method, - boost::shared_ptr oid, boost::shared_ptr argMap, - SchemaObjectClass* objectClass); - void sendBufferLH(Buffer& buf, const string& destination, const string& routingKey); - - void sendPackageIndicationLH(const string& packageName); - void sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key); - void sendCommandCompleteLH(const string& exchange, const string& key, uint32_t seq, - uint32_t code = 0, const string& text = "OK"); - void sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text=""); - void handleAttachResponse(Buffer& inBuffer); - void handlePackageRequest(Buffer& inBuffer); - void handleClassQuery(Buffer& inBuffer); - void handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, - const string& replyToExchange, const string& replyToKey); - void handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); - void handleMethodRequest(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); - void handleConsoleAddedIndication(); - }; -} - -const char* AgentImpl::QMF_EXCHANGE = "qpid.management"; -const char* AgentImpl::DIR_EXCHANGE = "amq.direct"; -const char* AgentImpl::BROKER_KEY = "broker"; - -#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} - -AgentEvent AgentEventImpl::copy() -{ - AgentEvent item; - - ::memset(&item, 0, sizeof(AgentEvent)); - item.kind = kind; - item.sequence = sequence; - item.object = object; - item.objectId = objectId.get(); - item.query = &query; - item.arguments = arguments.get(); - item.objectClass = objectClass; - - STRING_REF(authUserId); - STRING_REF(authToken); - STRING_REF(name); - STRING_REF(exchange); - STRING_REF(bindingKey); - - return item; -} - -AgentImpl::AgentImpl(char* _label, bool i) : - label(_label), queueName("qmfa-"), internalStore(i), nextTransientId(1), - requestedBrokerBank(0), requestedAgentBank(0), - assignedBrokerBank(0), assignedAgentBank(0), - bootSequence(1), nextObjectId(1), nextContextNum(1) -{ - queueName += label; -} - -AgentImpl::~AgentImpl() -{ -} - -void AgentImpl::setStoreDir(char* path) -{ - Mutex::ScopedLock _lock(lock); - if (path) - storeDir = path; - else - storeDir.clear(); -} - -void AgentImpl::setTransferDir(char* path) -{ - Mutex::ScopedLock _lock(lock); - if (path) - transferDir = path; - else - transferDir.clear(); -} - -void AgentImpl::handleRcvMessage(Message& message) -{ - Buffer inBuffer(message.body, message.length); - uint8_t opcode; - uint32_t sequence; - string replyToExchange(message.replyExchange ? message.replyExchange : ""); - string replyToKey(message.replyKey ? message.replyKey : ""); - string userId(message.userId ? message.userId : ""); - - if (checkHeader(inBuffer, &opcode, &sequence)) { - if (opcode == 'a') handleAttachResponse(inBuffer); - else if (opcode == 'S') handleSchemaRequest(inBuffer, sequence, replyToExchange, replyToKey); - else if (opcode == 'x') handleConsoleAddedIndication(); - else if (opcode == 'G') handleGetQuery(inBuffer, sequence, replyToKey, userId); - else if (opcode == 'M') handleMethodRequest(inBuffer, sequence, replyToKey, userId); - } -} - -bool AgentImpl::getXmtMessage(Message& item) -{ - Mutex::ScopedLock _lock(lock); - if (xmtQueue.empty()) - return false; - item = xmtQueue.front()->copy(); - return true; -} - -void AgentImpl::popXmt() -{ - Mutex::ScopedLock _lock(lock); - if (!xmtQueue.empty()) - xmtQueue.pop_front(); -} - -bool AgentImpl::getEvent(AgentEvent& event) -{ - Mutex::ScopedLock _lock(lock); - if (eventQueue.empty()) - return false; - event = eventQueue.front()->copy(); - return true; -} - -void AgentImpl::popEvent() -{ - Mutex::ScopedLock _lock(lock); - if (!eventQueue.empty()) - eventQueue.pop_front(); -} - -void AgentImpl::newSession() -{ - Mutex::ScopedLock _lock(lock); - eventQueue.clear(); - xmtQueue.clear(); - eventQueue.push_back(eventDeclareQueue(queueName)); - eventQueue.push_back(eventBind("amq.direct", queueName, queueName)); - eventQueue.push_back(eventSetupComplete()); -} - -void AgentImpl::startProtocol() -{ - Mutex::ScopedLock _lock(lock); - char rawbuffer[512]; - Buffer buffer(rawbuffer, 512); - - encodeHeader(buffer, 'A'); - buffer.putShortString("qmfa"); - systemId.encode(buffer); - buffer.putLong(requestedBrokerBank); - buffer.putLong(requestedAgentBank); - sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT AttachRequest: reqBroker=" << requestedBrokerBank << - " reqAgent=" << requestedAgentBank); -} - -void AgentImpl::heartbeat() -{ - Mutex::ScopedLock _lock(lock); - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - - encodeHeader(buffer, 'h'); - buffer.putLongLong(uint64_t(Duration(now()))); - stringstream key; - key << "console.heartbeat." << assignedBrokerBank << "." << assignedAgentBank; - sendBufferLH(buffer, QMF_EXCHANGE, key.str()); - QPID_LOG(trace, "SENT HeartbeatIndication"); -} - -void AgentImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, - const Value& argMap) -{ - Mutex::ScopedLock _lock(lock); - map::iterator iter = contextMap.find(sequence); - if (iter == contextMap.end()) - return; - AgentQueryContext::Ptr context = iter->second; - contextMap.erase(iter); - - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'm', context->sequence); - buffer.putLong(status); - buffer.putMediumString(text); - if (status == 0) { - for (vector::const_iterator aIter = context->schemaMethod->arguments.begin(); - aIter != context->schemaMethod->arguments.end(); aIter++) { - const SchemaArgumentImpl* schemaArg = *aIter; - if (schemaArg->dir == DIR_OUT || schemaArg->dir == DIR_IN_OUT) { - if (argMap.keyInMap(schemaArg->name.c_str())) { - const Value* val = argMap.byKey(schemaArg->name.c_str()); - val->impl->encode(buffer); - } else { - Value val(schemaArg->typecode); - val.impl->encode(buffer); - } - } - } - } - sendBufferLH(buffer, context->exchange, context->key); - QPID_LOG(trace, "SENT MethodResponse"); -} - -void AgentImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) -{ - Mutex::ScopedLock _lock(lock); - map::iterator iter = contextMap.find(sequence); - if (iter == contextMap.end()) - return; - AgentQueryContext::Ptr context = iter->second; - - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'g', context->sequence); - - object.impl->encodeSchemaKey(buffer); - object.impl->encodeManagedObjectData(buffer); - if (prop) - object.impl->encodeProperties(buffer); - if (stat) - object.impl->encodeStatistics(buffer); - - sendBufferLH(buffer, context->exchange, context->key); - QPID_LOG(trace, "SENT ContentIndication"); -} - -void AgentImpl::queryComplete(uint32_t sequence) -{ - Mutex::ScopedLock _lock(lock); - map::iterator iter = contextMap.find(sequence); - if (iter == contextMap.end()) - return; - - AgentQueryContext::Ptr context = iter->second; - contextMap.erase(iter); - sendCommandCompleteLH(context->exchange, context->key, context->sequence, 0, "OK"); -} - -void AgentImpl::registerClass(SchemaObjectClass* cls) -{ - Mutex::ScopedLock _lock(lock); - SchemaObjectClassImpl* impl = cls->impl; - - map::iterator iter = packages.find(impl->package); - if (iter == packages.end()) { - packages[impl->package] = ClassMaps(); - iter = packages.find(impl->package); - // TODO: Indicate this package if connected - } - - SchemaClassKey key(impl->name, impl->getHash()); - iter->second.objectClasses[key] = impl; - - // TODO: Indicate this schema if connected. -} - -void AgentImpl::registerClass(SchemaEventClass* cls) -{ - Mutex::ScopedLock _lock(lock); - SchemaEventClassImpl* impl = cls->impl; - - map::iterator iter = packages.find(impl->package); - if (iter == packages.end()) { - packages[impl->package] = ClassMaps(); - iter = packages.find(impl->package); - // TODO: Indicate this package if connected - } - - SchemaClassKey key(impl->name, impl->getHash()); - iter->second.eventClasses[key] = impl; - - // TODO: Indicate this schema if connected. -} - -const ObjectId* AgentImpl::addObject(Object&, uint64_t) -{ - Mutex::ScopedLock _lock(lock); - return 0; -} - -const ObjectId* AgentImpl::allocObjectId(uint64_t persistId) -{ - Mutex::ScopedLock _lock(lock); - uint16_t sequence = persistId ? 0 : bootSequence; - uint64_t objectNum = persistId ? persistId : nextObjectId++; - - ObjectIdImpl* oid = new ObjectIdImpl(&attachment, 0, sequence, objectNum); - return oid->envelope; -} - -const ObjectId* AgentImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) -{ - return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo); -} - -void AgentImpl::raiseEvent(Event&) -{ - Mutex::ScopedLock _lock(lock); -} - -void AgentImpl::encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq) -{ - buf.putOctet('A'); - buf.putOctet('M'); - buf.putOctet('3'); - buf.putOctet(opcode); - buf.putLong (seq); -} - -bool AgentImpl::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq) -{ - if (buf.getSize() < 8) - return false; - - uint8_t h1 = buf.getOctet(); - uint8_t h2 = buf.getOctet(); - uint8_t h3 = buf.getOctet(); - - *opcode = buf.getOctet(); - *seq = buf.getLong(); - - return h1 == 'A' && h2 == 'M' && h3 == '3'; -} - -AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name) -{ - AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::DECLARE_QUEUE)); - event->name = name; - - return event; -} - -AgentEventImpl::Ptr AgentImpl::eventBind(const string& exchange, const string& queue, - const string& key) -{ - AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::BIND)); - event->name = queue; - event->exchange = exchange; - event->bindingKey = key; - - return event; -} - -AgentEventImpl::Ptr AgentImpl::eventSetupComplete() -{ - AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::SETUP_COMPLETE)); - return event; -} - -AgentEventImpl::Ptr AgentImpl::eventQuery(uint32_t num, const string& userId, const string& package, - const string& cls, boost::shared_ptr oid) -{ - AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::GET_QUERY)); - event->sequence = num; - event->authUserId = userId; - event->query.impl->packageName = package; - event->query.impl->className = cls; - event->query.impl->oid = oid; - return event; -} - -AgentEventImpl::Ptr AgentImpl::eventMethod(uint32_t num, const string& userId, const string& method, - boost::shared_ptr oid, boost::shared_ptr argMap, - SchemaObjectClass* objectClass) -{ - AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::METHOD_CALL)); - event->sequence = num; - event->authUserId = userId; - event->name = method; - event->objectId = oid; - event->arguments = argMap; - event->objectClass = objectClass; - return event; -} - -void AgentImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) -{ - uint32_t length = buf.getPosition(); - MessageImpl::Ptr message(new MessageImpl); - - buf.reset(); - buf.getRawData(message->body, length); - message->destination = destination; - message->routingKey = routingKey; - message->replyExchange = "amq.direct"; - message->replyKey = queueName; - - xmtQueue.push_back(message); -} - -void AgentImpl::sendPackageIndicationLH(const string& packageName) -{ - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'p'); - buffer.putShortString(packageName); - sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT PackageIndication: package_name=" << packageName); -} - -void AgentImpl::sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key) -{ - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'q'); - buffer.putOctet((int) kind); - buffer.putShortString(packageName); - buffer.putShortString(key.name); - buffer.putBin128(const_cast(key.hash)); // const_cast needed for older Qpid libraries - sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); - QPID_LOG(trace, "SENT ClassIndication: package_name=" << packageName << " class_name=" << key.name); -} - -void AgentImpl::sendCommandCompleteLH(const string& exchange, const string& replyToKey, - uint32_t sequence, uint32_t code, const string& text) -{ - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'z', sequence); - buffer.putLong(code); - buffer.putShortString(text); - sendBufferLH(buffer, exchange, replyToKey); - QPID_LOG(trace, "SENT CommandComplete: seq=" << sequence << " code=" << code << " text=" << text); -} - -void AgentImpl::sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text) -{ - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 'm', sequence); - buffer.putLong(code); - - string fulltext; - switch (code) { - case MERR_UNKNOWN_PACKAGE: fulltext = "Unknown Package"; break; - case MERR_UNKNOWN_CLASS: fulltext = "Unknown Class"; break; - case MERR_UNKNOWN_METHOD: fulltext = "Unknown Method"; break; - case MERR_INTERNAL_ERROR: fulltext = "Internal Error"; break; - default: fulltext = "Unspecified Error"; break; - } - - if (!text.empty()) { - fulltext += " ("; - fulltext += text; - fulltext += ")"; - } - - buffer.putMediumString(fulltext); - sendBufferLH(buffer, DIR_EXCHANGE, key); - QPID_LOG(trace, "SENT MethodResponse: errorCode=" << code << " text=" << fulltext); -} - -void AgentImpl::handleAttachResponse(Buffer& inBuffer) -{ - Mutex::ScopedLock _lock(lock); - - assignedBrokerBank = inBuffer.getLong(); - assignedAgentBank = inBuffer.getLong(); - - QPID_LOG(trace, "RCVD AttachResponse: broker=" << assignedBrokerBank << " agent=" << assignedAgentBank); - - if ((assignedBrokerBank != requestedBrokerBank) || - (assignedAgentBank != requestedAgentBank)) { - if (requestedAgentBank == 0) { - QPID_LOG(notice, "Initial object-id bank assigned: " << assignedBrokerBank << "." << - assignedAgentBank); - } else { - QPID_LOG(warning, "Collision in object-id! New bank assigned: " << assignedBrokerBank << - "." << assignedAgentBank); - } - //storeData(); // TODO - requestedBrokerBank = assignedBrokerBank; - requestedAgentBank = assignedAgentBank; - } - - attachment.setBanks(assignedBrokerBank, assignedAgentBank); - - // Bind to qpid.management to receive commands - stringstream key; - key << "agent." << assignedBrokerBank << "." << assignedAgentBank; - eventQueue.push_back(eventBind(QMF_EXCHANGE, queueName, key.str())); - - // Send package indications for all local packages - for (map::iterator pIter = packages.begin(); - pIter != packages.end(); - pIter++) { - sendPackageIndicationLH(pIter->first); - - // Send class indications for all local classes - ClassMaps cMap = pIter->second; - for (ObjectClassMap::iterator cIter = cMap.objectClasses.begin(); - cIter != cMap.objectClasses.end(); cIter++) - sendClassIndicationLH(CLASS_OBJECT, pIter->first, cIter->first); - for (EventClassMap::iterator cIter = cMap.eventClasses.begin(); - cIter != cMap.eventClasses.end(); cIter++) - sendClassIndicationLH(CLASS_EVENT, pIter->first, cIter->first); - } -} - -void AgentImpl::handlePackageRequest(Buffer&) -{ - Mutex::ScopedLock _lock(lock); -} - -void AgentImpl::handleClassQuery(Buffer&) -{ - Mutex::ScopedLock _lock(lock); -} - -void AgentImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, - const string& replyExchange, const string& replyKey) -{ - Mutex::ScopedLock _lock(lock); - string rExchange(replyExchange); - string rKey(replyKey); - string packageName; - inBuffer.getShortString(packageName); - SchemaClassKey key(inBuffer); - - if (rExchange.empty()) - rExchange = QMF_EXCHANGE; - if (rKey.empty()) - rKey = BROKER_KEY; - - QPID_LOG(trace, "RCVD SchemaRequest: package=" << packageName << " class=" << key.name); - - map::iterator pIter = packages.find(packageName); - if (pIter == packages.end()) { - sendCommandCompleteLH(rExchange, rKey, sequence, 1, "package not found"); - return; - } - - ClassMaps cMap = pIter->second; - ObjectClassMap::iterator ocIter = cMap.objectClasses.find(key); - if (ocIter != cMap.objectClasses.end()) { - SchemaObjectClassImpl* oImpl = ocIter->second; - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 's', sequence); - oImpl->encode(buffer); - sendBufferLH(buffer, rExchange, rKey); - QPID_LOG(trace, "SENT SchemaResponse: (object) package=" << packageName << " class=" << key.name); - return; - } - - EventClassMap::iterator ecIter = cMap.eventClasses.find(key); - if (ecIter != cMap.eventClasses.end()) { - SchemaEventClassImpl* eImpl = ecIter->second; - Buffer buffer(outputBuffer, MA_BUFFER_SIZE); - encodeHeader(buffer, 's', sequence); - eImpl->encode(buffer); - sendBufferLH(buffer, rExchange, rKey); - QPID_LOG(trace, "SENT SchemaResponse: (event) package=" << packageName << " class=" << key.name); - return; - } - - sendCommandCompleteLH(rExchange, rKey, sequence, 1, "class not found"); -} - -void AgentImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId) -{ - Mutex::ScopedLock _lock(lock); - FieldTable ft; - FieldTable::ValuePtr value; - map::const_iterator pIter = packages.end(); - string pname; - string cname; - string oidRepr; - boost::shared_ptr oid; - - ft.decode(inBuffer); - - QPID_LOG(trace, "RCVD GetQuery: map=" << ft); - - value = ft.get("_package"); - if (value.get() && value->convertsTo()) { - pname = value->get(); - pIter = packages.find(pname); - if (pIter == packages.end()) { - sendCommandCompleteLH(DIR_EXCHANGE, replyTo, sequence); - return; - } - } - - value = ft.get("_class"); - if (value.get() && value->convertsTo()) { - cname = value->get(); - // TODO - check for validity of class (in package or any package) - if (pIter == packages.end()) { - } else { - - } - } - - value = ft.get("_objectid"); - if (value.get() && value->convertsTo()) { - oidRepr = value->get(); - oid.reset(new ObjectId()); - oid->impl->fromString(oidRepr); - } - - AgentQueryContext::Ptr context(new AgentQueryContext); - uint32_t contextNum = nextContextNum++; - context->sequence = sequence; - context->exchange = DIR_EXCHANGE; - context->key = replyTo; - contextMap[contextNum] = context; - - eventQueue.push_back(eventQuery(contextNum, userId, pname, cname, oid)); -} - -void AgentImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const string& replyTo, const string& userId) -{ - Mutex::ScopedLock _lock(lock); - string pname; - string method; - ObjectIdImpl* oidImpl = new ObjectIdImpl(buffer); - boost::shared_ptr oid(oidImpl->envelope); - buffer.getShortString(pname); - SchemaClassKey classKey(buffer); - buffer.getShortString(method); - - map::const_iterator pIter = packages.find(pname); - if (pIter == packages.end()) { - sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_PACKAGE, pname); - return; - } - - ObjectClassMap::const_iterator cIter = pIter->second.objectClasses.find(classKey); - if (cIter == pIter->second.objectClasses.end()) { - sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_CLASS, classKey.repr()); - return; - } - - const SchemaObjectClassImpl* schema = cIter->second; - vector::const_iterator mIter = schema->methods.begin(); - for (; mIter != schema->methods.end(); mIter++) { - if ((*mIter)->name == method) - break; - } - - if (mIter == schema->methods.end()) { - sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_METHOD, method); - return; - } - - SchemaMethodImpl* schemaMethod = *mIter; - boost::shared_ptr argMap(new Value(TYPE_MAP)); - ValueImpl* value; - for (vector::const_iterator aIter = schemaMethod->arguments.begin(); - aIter != schemaMethod->arguments.end(); aIter++) { - const SchemaArgumentImpl* schemaArg = *aIter; - if (schemaArg->dir == DIR_IN || schemaArg->dir == DIR_IN_OUT) - value = new ValueImpl(schemaArg->typecode, buffer); - else - value = new ValueImpl(schemaArg->typecode); - argMap->insert(schemaArg->name.c_str(), value->envelope); - } - - AgentQueryContext::Ptr context(new AgentQueryContext); - uint32_t contextNum = nextContextNum++; - context->sequence = sequence; - context->exchange = DIR_EXCHANGE; - context->key = replyTo; - context->schemaMethod = schemaMethod; - contextMap[contextNum] = context; - - eventQueue.push_back(eventMethod(contextNum, userId, method, oid, argMap, schema->envelope)); -} - -void AgentImpl::handleConsoleAddedIndication() -{ - Mutex::ScopedLock _lock(lock); -} - -//================================================================== -// Wrappers -//================================================================== - -Agent::Agent(char* label, bool internalStore) -{ - impl = new AgentImpl(label, internalStore); -} - -Agent::~Agent() -{ - delete impl; -} - -void Agent::setStoreDir(char* path) -{ - impl->setStoreDir(path); -} - -void Agent::setTransferDir(char* path) -{ - impl->setTransferDir(path); -} - -void Agent::handleRcvMessage(Message& message) -{ - impl->handleRcvMessage(message); -} - -bool Agent::getXmtMessage(Message& item) -{ - return impl->getXmtMessage(item); -} - -void Agent::popXmt() -{ - impl->popXmt(); -} - -bool Agent::getEvent(AgentEvent& event) -{ - return impl->getEvent(event); -} - -void Agent::popEvent() -{ - impl->popEvent(); -} - -void Agent::newSession() -{ - impl->newSession(); -} - -void Agent::startProtocol() -{ - impl->startProtocol(); -} - -void Agent::heartbeat() -{ - impl->heartbeat(); -} - -void Agent::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments) -{ - impl->methodResponse(sequence, status, text, arguments); -} - -void Agent::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) -{ - impl->queryResponse(sequence, object, prop, stat); -} - -void Agent::queryComplete(uint32_t sequence) -{ - impl->queryComplete(sequence); -} - -void Agent::registerClass(SchemaObjectClass* cls) -{ - impl->registerClass(cls); -} - -void Agent::registerClass(SchemaEventClass* cls) -{ - impl->registerClass(cls); -} - -const ObjectId* Agent::addObject(Object& obj, uint64_t persistId) -{ - return impl->addObject(obj, persistId); -} - -const ObjectId* Agent::allocObjectId(uint64_t persistId) -{ - return impl->allocObjectId(persistId); -} - -const ObjectId* Agent::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) -{ - return impl->allocObjectId(persistIdLo, persistIdHi); -} - -void Agent::raiseEvent(Event& event) -{ - impl->raiseEvent(event); -} - diff --git a/qpid/cpp/src/qmf/Agent.h b/qpid/cpp/src/qmf/Agent.h deleted file mode 100644 index d8f784e9d8..0000000000 --- a/qpid/cpp/src/qmf/Agent.h +++ /dev/null @@ -1,206 +0,0 @@ -#ifndef _QmfAgent_ -#define _QmfAgent_ - -/* - * 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. - */ - -#include -#include -#include -#include -#include -#include -#include - -namespace qmf { - - /** - * AgentEvent - * - * This structure represents a QMF event coming from the agent to - * the application. - */ - struct AgentEvent { - enum EventKind { - GET_QUERY = 1, - START_SYNC = 2, - END_SYNC = 3, - METHOD_CALL = 4, - DECLARE_QUEUE = 5, - DELETE_QUEUE = 6, - BIND = 7, - UNBIND = 8, - SETUP_COMPLETE = 9 - }; - - EventKind kind; - uint32_t sequence; // Protocol sequence (for all kinds) - char* authUserId; // Authenticated user ID (for all kinds) - char* authToken; // Authentication token if issued (for all kinds) - char* name; // Name of the method/sync query - // (METHOD_CALL, START_SYNC, END_SYNC, DECLARE_QUEUE, BIND, UNBIND) - Object* object; // Object involved in method call (METHOD_CALL) - ObjectId* objectId; // ObjectId for method call (METHOD_CALL) - Query* query; // Query parameters (GET_QUERY, START_SYNC) - Value* arguments; // Method parameters (METHOD_CALL) - char* exchange; // Exchange for bind (BIND, UNBIND) - char* bindingKey; // Key for bind (BIND, UNBIND) - SchemaObjectClass* objectClass; // (METHOD_CALL) - }; - - class AgentImpl; - - /** - * Agent - Protocol engine for the QMF agent - */ - class Agent { - public: - Agent(char* label, bool internalStore=true); - ~Agent(); - - /** - * Configure the directory path for storing persistent data. - *@param path Null-terminated string containing a directory path where files can be - * created, written, and read. If NULL, no persistent storage will be - * attempted. - */ - void setStoreDir(char* path); - - /** - * Configure the directory path for files transferred over QMF. - *@param path Null-terminated string containing a directory path where files can be - * created, deleted, written, and read. If NULL, file transfers shall not - * be permitted. - */ - void setTransferDir(char* path); - - /** - * Pass messages received from the AMQP session to the Agent engine. - *@param message AMQP messages received on the agent session. - */ - void handleRcvMessage(Message& message); - - /** - * Get the next message to be sent to the AMQP network. - *@param item The Message structure describing the message to be produced. - *@return true if the Message is valid, false if there are no messages to send. - */ - bool getXmtMessage(Message& item); - - /** - * Remove and discard one message from the head of the transmit queue. - */ - void popXmt(); - - /** - * Get the next application event from the agent engine. - *@param event The event iff the return value is true - *@return true if event is valid, false if there are no events to process - */ - bool getEvent(AgentEvent& event); - - /** - * Remove and discard one event from the head of the event queue. - */ - void popEvent(); - - /** - * A new AMQP session has been established for Agent communication. - */ - void newSession(); - - /** - * Start the QMF Agent protocol. This should be invoked after a SETUP_COMPLETE event - * is received from the Agent engine. - */ - void startProtocol(); - - /** - * This method is called periodically so the agent can supply a heartbeat. - */ - void heartbeat(); - - /** - * Respond to a method request. - *@param sequence The sequence number from the method request event. - *@param status The method's completion status. - *@param text Status text ("OK" or an error message) - *@param arguments The list of output arguments from the method call. - */ - void methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments); - - /** - * Send a content indication to the QMF bus. This is only needed for objects that are - * managed by the application. This is *NOT* needed for objects managed by the Agent - * (inserted using addObject). - *@param sequence The sequence number of the GET request or the SYNC_START request. - *@param object The object (annotated with "changed" flags) for publication. - *@param prop If true, changed object properties are transmitted. - *@param stat If true, changed object statistics are transmitted. - */ - void queryResponse(uint32_t sequence, Object& object, bool prop = true, bool stat = true); - - /** - * Indicate the completion of a query. This is not used for SYNC_START requests. - *@param sequence The sequence number of the GET request. - */ - void queryComplete(uint32_t sequence); - - /** - * Register a schema class with the Agent. - *@param cls A SchemaObejctClass object that defines data managed by the agent. - */ - void registerClass(SchemaObjectClass* cls); - - /** - * Register a schema class with the Agent. - *@param cls A SchemaEventClass object that defines events sent by the agent. - */ - void registerClass(SchemaEventClass* cls); - - /** - * Give an object to the Agent for storage and management. Once added, the agent takes - * responsibility for the life cycle of the object. - *@param obj The object to be managed by the Agent. - *@param persistId A unique non-zero value if the object-id is to be persistent. - *@return The objectId of the managed object. - */ - const ObjectId* addObject(Object& obj, uint64_t persistId); - - /** - * Allocate an objecc-id for an object that will be managed by the application. - *@param persistId A unique non-zero value if the object-id is to be persistent. - @return The objectId structure for the allocated ID. - */ - const ObjectId* allocObjectId(uint64_t persistId); - const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); - - /** - * Raise an event into the QMF network.. - *@param event The event object for the event to be raised. - */ - void raiseEvent(Event& event); - - private: - AgentImpl* impl; - }; -} - -#endif - diff --git a/qpid/cpp/src/qmf/AgentEngine.cpp b/qpid/cpp/src/qmf/AgentEngine.cpp new file mode 100644 index 0000000000..bef8b3d102 --- /dev/null +++ b/qpid/cpp/src/qmf/AgentEngine.cpp @@ -0,0 +1,958 @@ +/* + * 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. + */ + +#include "qmf/AgentEngine.h" +#include "qmf/MessageImpl.h" +#include "qmf/SchemaImpl.h" +#include "qmf/Typecode.h" +#include "qmf/ObjectImpl.h" +#include "qmf/ObjectIdImpl.h" +#include "qmf/QueryImpl.h" +#include "qmf/ValueImpl.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace qmf; +using namespace qpid::framing; +using namespace qpid::sys; + +namespace qmf { + + struct AgentEventImpl { + typedef boost::shared_ptr Ptr; + AgentEvent::EventKind kind; + uint32_t sequence; + string authUserId; + string authToken; + string name; + Object* object; + boost::shared_ptr objectId; + Query query; + boost::shared_ptr arguments; + string exchange; + string bindingKey; + SchemaObjectClass* objectClass; + + AgentEventImpl(AgentEvent::EventKind k) : + kind(k), sequence(0), object(0), objectClass(0) {} + ~AgentEventImpl() {} + AgentEvent copy(); + }; + + struct AgentQueryContext { + typedef boost::shared_ptr Ptr; + uint32_t sequence; + string exchange; + string key; + SchemaMethodImpl* schemaMethod; + AgentQueryContext() : schemaMethod(0) {} + }; + + class AgentEngineImpl { + public: + AgentEngineImpl(char* label, bool internalStore); + ~AgentEngineImpl(); + + void setStoreDir(const char* path); + void setTransferDir(const char* path); + void handleRcvMessage(Message& message); + bool getXmtMessage(Message& item); + void popXmt(); + bool getEvent(AgentEvent& event); + void popEvent(); + void newSession(); + void startProtocol(); + void heartbeat(); + void methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments); + void queryResponse(uint32_t sequence, Object& object, bool prop, bool stat); + void queryComplete(uint32_t sequence); + void registerClass(SchemaObjectClass* cls); + void registerClass(SchemaEventClass* cls); + const ObjectId* addObject(Object& obj, uint64_t persistId); + const ObjectId* allocObjectId(uint64_t persistId); + const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); + void raiseEvent(Event& event); + + private: + Mutex lock; + Mutex addLock; + string label; + string queueName; + string storeDir; + string transferDir; + bool internalStore; + uint64_t nextTransientId; + Uuid systemId; + uint32_t requestedBrokerBank; + uint32_t requestedAgentBank; + uint32_t assignedBrokerBank; + uint32_t assignedAgentBank; + AgentAttachment attachment; + uint16_t bootSequence; + uint64_t nextObjectId; + uint32_t nextContextNum; + deque eventQueue; + deque xmtQueue; + map contextMap; + + static const char* QMF_EXCHANGE; + static const char* DIR_EXCHANGE; + static const char* BROKER_KEY; + static const uint32_t MERR_UNKNOWN_METHOD = 2; + static const uint32_t MERR_UNKNOWN_PACKAGE = 8; + static const uint32_t MERR_UNKNOWN_CLASS = 9; + static const uint32_t MERR_INTERNAL_ERROR = 10; +# define MA_BUFFER_SIZE 65536 + char outputBuffer[MA_BUFFER_SIZE]; + + struct SchemaClassKey { + string name; + uint8_t hash[16]; + SchemaClassKey(const string& n, const uint8_t* h) : name(n) { + memcpy(hash, h, 16); + } + SchemaClassKey(Buffer& buffer) { + buffer.getShortString(name); + buffer.getBin128(hash); + } + string repr() { + return name; + } + }; + + struct SchemaClassKeyComp { + bool operator() (const SchemaClassKey& lhs, const SchemaClassKey& rhs) const + { + if (lhs.name != rhs.name) + return lhs.name < rhs.name; + else + for (int i = 0; i < 16; i++) + if (lhs.hash[i] != rhs.hash[i]) + return lhs.hash[i] < rhs.hash[i]; + return false; + } + }; + + typedef map ObjectClassMap; + typedef map EventClassMap; + + struct ClassMaps { + ObjectClassMap objectClasses; + EventClassMap eventClasses; + }; + + map packages; + + bool checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq); + void encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq = 0); + AgentEventImpl::Ptr eventDeclareQueue(const string& queueName); + AgentEventImpl::Ptr eventBind(const string& exchange, const string& queue, const string& key); + AgentEventImpl::Ptr eventSetupComplete(); + AgentEventImpl::Ptr eventQuery(uint32_t num, const string& userId, const string& package, const string& cls, + boost::shared_ptr oid); + AgentEventImpl::Ptr eventMethod(uint32_t num, const string& userId, const string& method, + boost::shared_ptr oid, boost::shared_ptr argMap, + SchemaObjectClass* objectClass); + void sendBufferLH(Buffer& buf, const string& destination, const string& routingKey); + + void sendPackageIndicationLH(const string& packageName); + void sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key); + void sendCommandCompleteLH(const string& exchange, const string& key, uint32_t seq, + uint32_t code = 0, const string& text = "OK"); + void sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text=""); + void handleAttachResponse(Buffer& inBuffer); + void handlePackageRequest(Buffer& inBuffer); + void handleClassQuery(Buffer& inBuffer); + void handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, + const string& replyToExchange, const string& replyToKey); + void handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); + void handleMethodRequest(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId); + void handleConsoleAddedIndication(); + }; +} + +const char* AgentEngineImpl::QMF_EXCHANGE = "qpid.management"; +const char* AgentEngineImpl::DIR_EXCHANGE = "amq.direct"; +const char* AgentEngineImpl::BROKER_KEY = "broker"; + +#define STRING_REF(s) {if (!s.empty()) item.s = const_cast(s.c_str());} + +AgentEvent AgentEventImpl::copy() +{ + AgentEvent item; + + ::memset(&item, 0, sizeof(AgentEvent)); + item.kind = kind; + item.sequence = sequence; + item.object = object; + item.objectId = objectId.get(); + item.query = &query; + item.arguments = arguments.get(); + item.objectClass = objectClass; + + STRING_REF(authUserId); + STRING_REF(authToken); + STRING_REF(name); + STRING_REF(exchange); + STRING_REF(bindingKey); + + return item; +} + +AgentEngineImpl::AgentEngineImpl(char* _label, bool i) : + label(_label), queueName("qmfa-"), internalStore(i), nextTransientId(1), + requestedBrokerBank(0), requestedAgentBank(0), + assignedBrokerBank(0), assignedAgentBank(0), + bootSequence(1), nextObjectId(1), nextContextNum(1) +{ + queueName += label; +} + +AgentEngineImpl::~AgentEngineImpl() +{ +} + +void AgentEngineImpl::setStoreDir(const char* path) +{ + Mutex::ScopedLock _lock(lock); + if (path) + storeDir = path; + else + storeDir.clear(); +} + +void AgentEngineImpl::setTransferDir(const char* path) +{ + Mutex::ScopedLock _lock(lock); + if (path) + transferDir = path; + else + transferDir.clear(); +} + +void AgentEngineImpl::handleRcvMessage(Message& message) +{ + Buffer inBuffer(message.body, message.length); + uint8_t opcode; + uint32_t sequence; + string replyToExchange(message.replyExchange ? message.replyExchange : ""); + string replyToKey(message.replyKey ? message.replyKey : ""); + string userId(message.userId ? message.userId : ""); + + if (checkHeader(inBuffer, &opcode, &sequence)) { + if (opcode == 'a') handleAttachResponse(inBuffer); + else if (opcode == 'S') handleSchemaRequest(inBuffer, sequence, replyToExchange, replyToKey); + else if (opcode == 'x') handleConsoleAddedIndication(); + else if (opcode == 'G') handleGetQuery(inBuffer, sequence, replyToKey, userId); + else if (opcode == 'M') handleMethodRequest(inBuffer, sequence, replyToKey, userId); + } +} + +bool AgentEngineImpl::getXmtMessage(Message& item) +{ + Mutex::ScopedLock _lock(lock); + if (xmtQueue.empty()) + return false; + item = xmtQueue.front()->copy(); + return true; +} + +void AgentEngineImpl::popXmt() +{ + Mutex::ScopedLock _lock(lock); + if (!xmtQueue.empty()) + xmtQueue.pop_front(); +} + +bool AgentEngineImpl::getEvent(AgentEvent& event) +{ + Mutex::ScopedLock _lock(lock); + if (eventQueue.empty()) + return false; + event = eventQueue.front()->copy(); + return true; +} + +void AgentEngineImpl::popEvent() +{ + Mutex::ScopedLock _lock(lock); + if (!eventQueue.empty()) + eventQueue.pop_front(); +} + +void AgentEngineImpl::newSession() +{ + Mutex::ScopedLock _lock(lock); + eventQueue.clear(); + xmtQueue.clear(); + eventQueue.push_back(eventDeclareQueue(queueName)); + eventQueue.push_back(eventBind("amq.direct", queueName, queueName)); + eventQueue.push_back(eventSetupComplete()); +} + +void AgentEngineImpl::startProtocol() +{ + Mutex::ScopedLock _lock(lock); + char rawbuffer[512]; + Buffer buffer(rawbuffer, 512); + + encodeHeader(buffer, 'A'); + buffer.putShortString("qmfa"); + systemId.encode(buffer); + buffer.putLong(requestedBrokerBank); + buffer.putLong(requestedAgentBank); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT AttachRequest: reqBroker=" << requestedBrokerBank << + " reqAgent=" << requestedAgentBank); +} + +void AgentEngineImpl::heartbeat() +{ + Mutex::ScopedLock _lock(lock); + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + + encodeHeader(buffer, 'h'); + buffer.putLongLong(uint64_t(Duration(now()))); + stringstream key; + key << "console.heartbeat." << assignedBrokerBank << "." << assignedAgentBank; + sendBufferLH(buffer, QMF_EXCHANGE, key.str()); + QPID_LOG(trace, "SENT HeartbeatIndication"); +} + +void AgentEngineImpl::methodResponse(uint32_t sequence, uint32_t status, char* text, + const Value& argMap) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + AgentQueryContext::Ptr context = iter->second; + contextMap.erase(iter); + + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'm', context->sequence); + buffer.putLong(status); + buffer.putMediumString(text); + if (status == 0) { + for (vector::const_iterator aIter = context->schemaMethod->arguments.begin(); + aIter != context->schemaMethod->arguments.end(); aIter++) { + const SchemaArgumentImpl* schemaArg = *aIter; + if (schemaArg->dir == DIR_OUT || schemaArg->dir == DIR_IN_OUT) { + if (argMap.keyInMap(schemaArg->name.c_str())) { + const Value* val = argMap.byKey(schemaArg->name.c_str()); + val->impl->encode(buffer); + } else { + Value val(schemaArg->typecode); + val.impl->encode(buffer); + } + } + } + } + sendBufferLH(buffer, context->exchange, context->key); + QPID_LOG(trace, "SENT MethodResponse"); +} + +void AgentEngineImpl::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + AgentQueryContext::Ptr context = iter->second; + + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'g', context->sequence); + + object.impl->encodeSchemaKey(buffer); + object.impl->encodeManagedObjectData(buffer); + if (prop) + object.impl->encodeProperties(buffer); + if (stat) + object.impl->encodeStatistics(buffer); + + sendBufferLH(buffer, context->exchange, context->key); + QPID_LOG(trace, "SENT ContentIndication"); +} + +void AgentEngineImpl::queryComplete(uint32_t sequence) +{ + Mutex::ScopedLock _lock(lock); + map::iterator iter = contextMap.find(sequence); + if (iter == contextMap.end()) + return; + + AgentQueryContext::Ptr context = iter->second; + contextMap.erase(iter); + sendCommandCompleteLH(context->exchange, context->key, context->sequence, 0, "OK"); +} + +void AgentEngineImpl::registerClass(SchemaObjectClass* cls) +{ + Mutex::ScopedLock _lock(lock); + SchemaObjectClassImpl* impl = cls->impl; + + map::iterator iter = packages.find(impl->package); + if (iter == packages.end()) { + packages[impl->package] = ClassMaps(); + iter = packages.find(impl->package); + // TODO: Indicate this package if connected + } + + SchemaClassKey key(impl->name, impl->getHash()); + iter->second.objectClasses[key] = impl; + + // TODO: Indicate this schema if connected. +} + +void AgentEngineImpl::registerClass(SchemaEventClass* cls) +{ + Mutex::ScopedLock _lock(lock); + SchemaEventClassImpl* impl = cls->impl; + + map::iterator iter = packages.find(impl->package); + if (iter == packages.end()) { + packages[impl->package] = ClassMaps(); + iter = packages.find(impl->package); + // TODO: Indicate this package if connected + } + + SchemaClassKey key(impl->name, impl->getHash()); + iter->second.eventClasses[key] = impl; + + // TODO: Indicate this schema if connected. +} + +const ObjectId* AgentEngineImpl::addObject(Object&, uint64_t) +{ + Mutex::ScopedLock _lock(lock); + return 0; +} + +const ObjectId* AgentEngineImpl::allocObjectId(uint64_t persistId) +{ + Mutex::ScopedLock _lock(lock); + uint16_t sequence = persistId ? 0 : bootSequence; + uint64_t objectNum = persistId ? persistId : nextObjectId++; + + ObjectIdImpl* oid = new ObjectIdImpl(&attachment, 0, sequence, objectNum); + return oid->envelope; +} + +const ObjectId* AgentEngineImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) +{ + return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo); +} + +void AgentEngineImpl::raiseEvent(Event&) +{ + Mutex::ScopedLock _lock(lock); +} + +void AgentEngineImpl::encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq) +{ + buf.putOctet('A'); + buf.putOctet('M'); + buf.putOctet('3'); + buf.putOctet(opcode); + buf.putLong (seq); +} + +bool AgentEngineImpl::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq) +{ + if (buf.getSize() < 8) + return false; + + uint8_t h1 = buf.getOctet(); + uint8_t h2 = buf.getOctet(); + uint8_t h3 = buf.getOctet(); + + *opcode = buf.getOctet(); + *seq = buf.getLong(); + + return h1 == 'A' && h2 == 'M' && h3 == '3'; +} + +AgentEventImpl::Ptr AgentEngineImpl::eventDeclareQueue(const string& name) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::DECLARE_QUEUE)); + event->name = name; + + return event; +} + +AgentEventImpl::Ptr AgentEngineImpl::eventBind(const string& exchange, const string& queue, + const string& key) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::BIND)); + event->name = queue; + event->exchange = exchange; + event->bindingKey = key; + + return event; +} + +AgentEventImpl::Ptr AgentEngineImpl::eventSetupComplete() +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::SETUP_COMPLETE)); + return event; +} + +AgentEventImpl::Ptr AgentEngineImpl::eventQuery(uint32_t num, const string& userId, const string& package, + const string& cls, boost::shared_ptr oid) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::GET_QUERY)); + event->sequence = num; + event->authUserId = userId; + event->query.impl->packageName = package; + event->query.impl->className = cls; + event->query.impl->oid = oid; + return event; +} + +AgentEventImpl::Ptr AgentEngineImpl::eventMethod(uint32_t num, const string& userId, const string& method, + boost::shared_ptr oid, boost::shared_ptr argMap, + SchemaObjectClass* objectClass) +{ + AgentEventImpl::Ptr event(new AgentEventImpl(AgentEvent::METHOD_CALL)); + event->sequence = num; + event->authUserId = userId; + event->name = method; + event->objectId = oid; + event->arguments = argMap; + event->objectClass = objectClass; + return event; +} + +void AgentEngineImpl::sendBufferLH(Buffer& buf, const string& destination, const string& routingKey) +{ + uint32_t length = buf.getPosition(); + MessageImpl::Ptr message(new MessageImpl); + + buf.reset(); + buf.getRawData(message->body, length); + message->destination = destination; + message->routingKey = routingKey; + message->replyExchange = "amq.direct"; + message->replyKey = queueName; + + xmtQueue.push_back(message); +} + +void AgentEngineImpl::sendPackageIndicationLH(const string& packageName) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'p'); + buffer.putShortString(packageName); + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT PackageIndication: package_name=" << packageName); +} + +void AgentEngineImpl::sendClassIndicationLH(ClassKind kind, const string& packageName, const SchemaClassKey& key) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'q'); + buffer.putOctet((int) kind); + buffer.putShortString(packageName); + buffer.putShortString(key.name); + buffer.putBin128(const_cast(key.hash)); // const_cast needed for older Qpid libraries + sendBufferLH(buffer, QMF_EXCHANGE, BROKER_KEY); + QPID_LOG(trace, "SENT ClassIndication: package_name=" << packageName << " class_name=" << key.name); +} + +void AgentEngineImpl::sendCommandCompleteLH(const string& exchange, const string& replyToKey, + uint32_t sequence, uint32_t code, const string& text) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'z', sequence); + buffer.putLong(code); + buffer.putShortString(text); + sendBufferLH(buffer, exchange, replyToKey); + QPID_LOG(trace, "SENT CommandComplete: seq=" << sequence << " code=" << code << " text=" << text); +} + +void AgentEngineImpl::sendMethodErrorLH(uint32_t sequence, const string& key, uint32_t code, const string& text) +{ + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 'm', sequence); + buffer.putLong(code); + + string fulltext; + switch (code) { + case MERR_UNKNOWN_PACKAGE: fulltext = "Unknown Package"; break; + case MERR_UNKNOWN_CLASS: fulltext = "Unknown Class"; break; + case MERR_UNKNOWN_METHOD: fulltext = "Unknown Method"; break; + case MERR_INTERNAL_ERROR: fulltext = "Internal Error"; break; + default: fulltext = "Unspecified Error"; break; + } + + if (!text.empty()) { + fulltext += " ("; + fulltext += text; + fulltext += ")"; + } + + buffer.putMediumString(fulltext); + sendBufferLH(buffer, DIR_EXCHANGE, key); + QPID_LOG(trace, "SENT MethodResponse: errorCode=" << code << " text=" << fulltext); +} + +void AgentEngineImpl::handleAttachResponse(Buffer& inBuffer) +{ + Mutex::ScopedLock _lock(lock); + + assignedBrokerBank = inBuffer.getLong(); + assignedAgentBank = inBuffer.getLong(); + + QPID_LOG(trace, "RCVD AttachResponse: broker=" << assignedBrokerBank << " agent=" << assignedAgentBank); + + if ((assignedBrokerBank != requestedBrokerBank) || + (assignedAgentBank != requestedAgentBank)) { + if (requestedAgentBank == 0) { + QPID_LOG(notice, "Initial object-id bank assigned: " << assignedBrokerBank << "." << + assignedAgentBank); + } else { + QPID_LOG(warning, "Collision in object-id! New bank assigned: " << assignedBrokerBank << + "." << assignedAgentBank); + } + //storeData(); // TODO + requestedBrokerBank = assignedBrokerBank; + requestedAgentBank = assignedAgentBank; + } + + attachment.setBanks(assignedBrokerBank, assignedAgentBank); + + // Bind to qpid.management to receive commands + stringstream key; + key << "agent." << assignedBrokerBank << "." << assignedAgentBank; + eventQueue.push_back(eventBind(QMF_EXCHANGE, queueName, key.str())); + + // Send package indications for all local packages + for (map::iterator pIter = packages.begin(); + pIter != packages.end(); + pIter++) { + sendPackageIndicationLH(pIter->first); + + // Send class indications for all local classes + ClassMaps cMap = pIter->second; + for (ObjectClassMap::iterator cIter = cMap.objectClasses.begin(); + cIter != cMap.objectClasses.end(); cIter++) + sendClassIndicationLH(CLASS_OBJECT, pIter->first, cIter->first); + for (EventClassMap::iterator cIter = cMap.eventClasses.begin(); + cIter != cMap.eventClasses.end(); cIter++) + sendClassIndicationLH(CLASS_EVENT, pIter->first, cIter->first); + } +} + +void AgentEngineImpl::handlePackageRequest(Buffer&) +{ + Mutex::ScopedLock _lock(lock); +} + +void AgentEngineImpl::handleClassQuery(Buffer&) +{ + Mutex::ScopedLock _lock(lock); +} + +void AgentEngineImpl::handleSchemaRequest(Buffer& inBuffer, uint32_t sequence, + const string& replyExchange, const string& replyKey) +{ + Mutex::ScopedLock _lock(lock); + string rExchange(replyExchange); + string rKey(replyKey); + string packageName; + inBuffer.getShortString(packageName); + SchemaClassKey key(inBuffer); + + if (rExchange.empty()) + rExchange = QMF_EXCHANGE; + if (rKey.empty()) + rKey = BROKER_KEY; + + QPID_LOG(trace, "RCVD SchemaRequest: package=" << packageName << " class=" << key.name); + + map::iterator pIter = packages.find(packageName); + if (pIter == packages.end()) { + sendCommandCompleteLH(rExchange, rKey, sequence, 1, "package not found"); + return; + } + + ClassMaps cMap = pIter->second; + ObjectClassMap::iterator ocIter = cMap.objectClasses.find(key); + if (ocIter != cMap.objectClasses.end()) { + SchemaObjectClassImpl* oImpl = ocIter->second; + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 's', sequence); + oImpl->encode(buffer); + sendBufferLH(buffer, rExchange, rKey); + QPID_LOG(trace, "SENT SchemaResponse: (object) package=" << packageName << " class=" << key.name); + return; + } + + EventClassMap::iterator ecIter = cMap.eventClasses.find(key); + if (ecIter != cMap.eventClasses.end()) { + SchemaEventClassImpl* eImpl = ecIter->second; + Buffer buffer(outputBuffer, MA_BUFFER_SIZE); + encodeHeader(buffer, 's', sequence); + eImpl->encode(buffer); + sendBufferLH(buffer, rExchange, rKey); + QPID_LOG(trace, "SENT SchemaResponse: (event) package=" << packageName << " class=" << key.name); + return; + } + + sendCommandCompleteLH(rExchange, rKey, sequence, 1, "class not found"); +} + +void AgentEngineImpl::handleGetQuery(Buffer& inBuffer, uint32_t sequence, const string& replyTo, const string& userId) +{ + Mutex::ScopedLock _lock(lock); + FieldTable ft; + FieldTable::ValuePtr value; + map::const_iterator pIter = packages.end(); + string pname; + string cname; + string oidRepr; + boost::shared_ptr oid; + + ft.decode(inBuffer); + + QPID_LOG(trace, "RCVD GetQuery: map=" << ft); + + value = ft.get("_package"); + if (value.get() && value->convertsTo()) { + pname = value->get(); + pIter = packages.find(pname); + if (pIter == packages.end()) { + sendCommandCompleteLH(DIR_EXCHANGE, replyTo, sequence); + return; + } + } + + value = ft.get("_class"); + if (value.get() && value->convertsTo()) { + cname = value->get(); + // TODO - check for validity of class (in package or any package) + if (pIter == packages.end()) { + } else { + + } + } + + value = ft.get("_objectid"); + if (value.get() && value->convertsTo()) { + oidRepr = value->get(); + oid.reset(new ObjectId()); + oid->impl->fromString(oidRepr); + } + + AgentQueryContext::Ptr context(new AgentQueryContext); + uint32_t contextNum = nextContextNum++; + context->sequence = sequence; + context->exchange = DIR_EXCHANGE; + context->key = replyTo; + contextMap[contextNum] = context; + + eventQueue.push_back(eventQuery(contextNum, userId, pname, cname, oid)); +} + +void AgentEngineImpl::handleMethodRequest(Buffer& buffer, uint32_t sequence, const string& replyTo, const string& userId) +{ + Mutex::ScopedLock _lock(lock); + string pname; + string method; + ObjectIdImpl* oidImpl = new ObjectIdImpl(buffer); + boost::shared_ptr oid(oidImpl->envelope); + buffer.getShortString(pname); + SchemaClassKey classKey(buffer); + buffer.getShortString(method); + + map::const_iterator pIter = packages.find(pname); + if (pIter == packages.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_PACKAGE, pname); + return; + } + + ObjectClassMap::const_iterator cIter = pIter->second.objectClasses.find(classKey); + if (cIter == pIter->second.objectClasses.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_CLASS, classKey.repr()); + return; + } + + const SchemaObjectClassImpl* schema = cIter->second; + vector::const_iterator mIter = schema->methods.begin(); + for (; mIter != schema->methods.end(); mIter++) { + if ((*mIter)->name == method) + break; + } + + if (mIter == schema->methods.end()) { + sendMethodErrorLH(sequence, replyTo, MERR_UNKNOWN_METHOD, method); + return; + } + + SchemaMethodImpl* schemaMethod = *mIter; + boost::shared_ptr argMap(new Value(TYPE_MAP)); + ValueImpl* value; + for (vector::const_iterator aIter = schemaMethod->arguments.begin(); + aIter != schemaMethod->arguments.end(); aIter++) { + const SchemaArgumentImpl* schemaArg = *aIter; + if (schemaArg->dir == DIR_IN || schemaArg->dir == DIR_IN_OUT) + value = new ValueImpl(schemaArg->typecode, buffer); + else + value = new ValueImpl(schemaArg->typecode); + argMap->insert(schemaArg->name.c_str(), value->envelope); + } + + AgentQueryContext::Ptr context(new AgentQueryContext); + uint32_t contextNum = nextContextNum++; + context->sequence = sequence; + context->exchange = DIR_EXCHANGE; + context->key = replyTo; + context->schemaMethod = schemaMethod; + contextMap[contextNum] = context; + + eventQueue.push_back(eventMethod(contextNum, userId, method, oid, argMap, schema->envelope)); +} + +void AgentEngineImpl::handleConsoleAddedIndication() +{ + Mutex::ScopedLock _lock(lock); +} + +//================================================================== +// Wrappers +//================================================================== + +AgentEngine::AgentEngine(char* label, bool internalStore) +{ + impl = new AgentEngineImpl(label, internalStore); +} + +AgentEngine::~AgentEngine() +{ + delete impl; +} + +void AgentEngine::setStoreDir(const char* path) +{ + impl->setStoreDir(path); +} + +void AgentEngine::setTransferDir(const char* path) +{ + impl->setTransferDir(path); +} + +void AgentEngine::handleRcvMessage(Message& message) +{ + impl->handleRcvMessage(message); +} + +bool AgentEngine::getXmtMessage(Message& item) +{ + return impl->getXmtMessage(item); +} + +void AgentEngine::popXmt() +{ + impl->popXmt(); +} + +bool AgentEngine::getEvent(AgentEvent& event) +{ + return impl->getEvent(event); +} + +void AgentEngine::popEvent() +{ + impl->popEvent(); +} + +void AgentEngine::newSession() +{ + impl->newSession(); +} + +void AgentEngine::startProtocol() +{ + impl->startProtocol(); +} + +void AgentEngine::heartbeat() +{ + impl->heartbeat(); +} + +void AgentEngine::methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments) +{ + impl->methodResponse(sequence, status, text, arguments); +} + +void AgentEngine::queryResponse(uint32_t sequence, Object& object, bool prop, bool stat) +{ + impl->queryResponse(sequence, object, prop, stat); +} + +void AgentEngine::queryComplete(uint32_t sequence) +{ + impl->queryComplete(sequence); +} + +void AgentEngine::registerClass(SchemaObjectClass* cls) +{ + impl->registerClass(cls); +} + +void AgentEngine::registerClass(SchemaEventClass* cls) +{ + impl->registerClass(cls); +} + +const ObjectId* AgentEngine::addObject(Object& obj, uint64_t persistId) +{ + return impl->addObject(obj, persistId); +} + +const ObjectId* AgentEngine::allocObjectId(uint64_t persistId) +{ + return impl->allocObjectId(persistId); +} + +const ObjectId* AgentEngine::allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi) +{ + return impl->allocObjectId(persistIdLo, persistIdHi); +} + +void AgentEngine::raiseEvent(Event& event) +{ + impl->raiseEvent(event); +} + diff --git a/qpid/cpp/src/qmf/AgentEngine.h b/qpid/cpp/src/qmf/AgentEngine.h new file mode 100644 index 0000000000..d18a104e96 --- /dev/null +++ b/qpid/cpp/src/qmf/AgentEngine.h @@ -0,0 +1,207 @@ +#ifndef _QmfAgentEngine_ +#define _QmfAgentEngine_ + +/* + * 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. + */ + +#include +#include +#include +#include +#include +#include +#include + +namespace qmf { + + /** + * AgentEvent + * + * This structure represents a QMF event coming from the agent to + * the application. + */ + struct AgentEvent { + enum EventKind { + GET_QUERY = 1, + START_SYNC = 2, + END_SYNC = 3, + METHOD_CALL = 4, + DECLARE_QUEUE = 5, + DELETE_QUEUE = 6, + BIND = 7, + UNBIND = 8, + SETUP_COMPLETE = 9 + }; + + EventKind kind; + uint32_t sequence; // Protocol sequence (for all kinds) + char* authUserId; // Authenticated user ID (for all kinds) + char* authToken; // Authentication token if issued (for all kinds) + char* name; // Name of the method/sync query + // (METHOD_CALL, START_SYNC, END_SYNC, DECLARE_QUEUE, BIND, UNBIND) + Object* object; // Object involved in method call (METHOD_CALL) + ObjectId* objectId; // ObjectId for method call (METHOD_CALL) + Query* query; // Query parameters (GET_QUERY, START_SYNC) + Value* arguments; // Method parameters (METHOD_CALL) + char* exchange; // Exchange for bind (BIND, UNBIND) + char* bindingKey; // Key for bind (BIND, UNBIND) + SchemaObjectClass* objectClass; // (METHOD_CALL) + }; + + class AgentEngineImpl; + + /** + * AgentEngine - Protocol engine for the QMF agent + */ + class AgentEngine { + public: + AgentEngine(char* label, bool internalStore=true); + ~AgentEngine(); + + /** + * Configure the directory path for storing persistent data. + *@param path Null-terminated string containing a directory path where files can be + * created, written, and read. If NULL, no persistent storage will be + * attempted. + */ + void setStoreDir(const char* path); + + /** + * Configure the directory path for files transferred over QMF. + *@param path Null-terminated string containing a directory path where files can be + * created, deleted, written, and read. If NULL, file transfers shall not + * be permitted. + */ + void setTransferDir(const char* path); + + /** + * Pass messages received from the AMQP session to the Agent engine. + *@param message AMQP messages received on the agent session. + */ + void handleRcvMessage(Message& message); + + /** + * Get the next message to be sent to the AMQP network. + *@param item The Message structure describing the message to be produced. + *@return true if the Message is valid, false if there are no messages to send. + */ + bool getXmtMessage(Message& item); + + /** + * Remove and discard one message from the head of the transmit queue. + */ + void popXmt(); + + /** + * Get the next application event from the agent engine. + *@param event The event iff the return value is true + *@return true if event is valid, false if there are no events to process + */ + bool getEvent(AgentEvent& event); + + /** + * Remove and discard one event from the head of the event queue. + */ + void popEvent(); + + /** + * A new AMQP session has been established for Agent communication. + */ + void newSession(); + + /** + * Start the QMF Agent protocol. This should be invoked after a SETUP_COMPLETE event + * is received from the Agent engine. + */ + void startProtocol(); + + /** + * This method is called periodically so the agent can supply a heartbeat. + */ + void heartbeat(); + + /** + * Respond to a method request. + *@param sequence The sequence number from the method request event. + *@param status The method's completion status. + *@param text Status text ("OK" or an error message) + *@param arguments The list of output arguments from the method call. + */ + void methodResponse(uint32_t sequence, uint32_t status, char* text, const Value& arguments); + + /** + * Send a content indication to the QMF bus. This is only needed for objects that are + * managed by the application. This is *NOT* needed for objects managed by the Agent + * (inserted using addObject). + *@param sequence The sequence number of the GET request or the SYNC_START request. + *@param object The object (annotated with "changed" flags) for publication. + *@param prop If true, changed object properties are transmitted. + *@param stat If true, changed object statistics are transmitted. + */ + void queryResponse(uint32_t sequence, Object& object, bool prop = true, bool stat = true); + + /** + * Indicate the completion of a query. This is not used for SYNC_START requests. + *@param sequence The sequence number of the GET request. + */ + void queryComplete(uint32_t sequence); + + /** + * Register a schema class with the Agent. + *@param cls A SchemaObejctClass object that defines data managed by the agent. + */ + void registerClass(SchemaObjectClass* cls); + + /** + * Register a schema class with the Agent. + *@param cls A SchemaEventClass object that defines events sent by the agent. + */ + void registerClass(SchemaEventClass* cls); + + /** + * Give an object to the Agent for storage and management. Once added, the agent takes + * responsibility for the life cycle of the object. + *@param obj The object to be managed by the Agent. + *@param persistId A unique non-zero value if the object-id is to be persistent. + *@return The objectId of the managed object. + */ + const ObjectId* addObject(Object& obj, uint64_t persistId); + const ObjectId* addObject(Object& obj, uint32_t persistIdLo, uint32_t persistIdHi); + + /** + * Allocate an object-id for an object that will be managed by the application. + *@param persistId A unique non-zero value if the object-id is to be persistent. + *@return The objectId structure for the allocated ID. + */ + const ObjectId* allocObjectId(uint64_t persistId); + const ObjectId* allocObjectId(uint32_t persistIdLo, uint32_t persistIdHi); + + /** + * Raise an event into the QMF network.. + *@param event The event object for the event to be raised. + */ + void raiseEvent(Event& event); + + private: + AgentEngineImpl* impl; + }; +} + +#endif + diff --git a/qpid/cpp/src/qmf/Console.h b/qpid/cpp/src/qmf/Console.h deleted file mode 100644 index de7949e1de..0000000000 --- a/qpid/cpp/src/qmf/Console.h +++ /dev/null @@ -1,82 +0,0 @@ -#ifndef _QmfConsole_ -#define _QmfConsole_ - -/* - * 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. - */ - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -namespace qmf { - - struct ConsoleSettings { - bool rcvObjects; - bool rcvEvents; - bool rcvHeartbeats; - bool userBindings; - uint32_t methodTimeout; - uint32_t getTimeout; - - ConsoleSettings() : - rcvObjects(true), - rcvEvents(true), - rcvHeartbeats(true), - userBindings(false), - methodTimeout(20), - getTimeout(20) {} - }; - - class Console { - public: - Console(ConsoleHandler* handler = 0, ConsoleSettings settings = ConsoleSettings()); - ~Console(); - - Broker* addConnection(ManagedConnection& connection); - void delConnection(Broker* broker); - void delConnection(ManagedConnection& connection); - - const PackageMap& getPackages() const; - - void bindPackage(const Package& package); - void bindPackage(const std::string& packageName); - void bindClass(const SchemaClass& otype); - void bindClass(const std::string& packageName, const std::string& className); - - void getAgents(std::set& agents, Broker* = 0); - void getObjects(std::vector& objects, const std::string& typeName, - const std::string& packageName = "", - Broker* broker = 0, - Agent* agent = 0); - void getObjects(std::vector& objects, - const std::map& query, - Broker* broker = 0, - Agent* agent = 0); - }; -} - -#endif - diff --git a/qpid/cpp/src/qmf/ConsoleEngine.h b/qpid/cpp/src/qmf/ConsoleEngine.h new file mode 100644 index 0000000000..823e281b14 --- /dev/null +++ b/qpid/cpp/src/qmf/ConsoleEngine.h @@ -0,0 +1,83 @@ +#ifndef _QmfConsoleEngine_ +#define _QmfConsoleEngine_ + +/* + * 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. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace qmf { + + struct ConsoleSettings { + bool rcvObjects; + bool rcvEvents; + bool rcvHeartbeats; + bool userBindings; + uint32_t methodTimeout; + uint32_t getTimeout; + + ConsoleSettings() : + rcvObjects(true), + rcvEvents(true), + rcvHeartbeats(true), + userBindings(false), + methodTimeout(20), + getTimeout(20) {} + }; + + class ConsoleEngine { + public: + ConsoleEngine(ConsoleHandler* handler = 0, ConsoleSettings settings = ConsoleSettings()); + ~ConsoleEngine(); + + Broker* addConnection(ManagedConnection& connection); + void delConnection(Broker* broker); + void delConnection(ManagedConnection& connection); + + const PackageMap& getPackages() const; + + void bindPackage(const Package& package); + void bindPackage(const std::string& packageName); + void bindClass(const SchemaClass& otype); + void bindClass(const std::string& packageName, const std::string& className); + + /* + void getAgents(std::set& agents, Broker* = 0); + void getObjects(std::vector& objects, const std::string& typeName, + const std::string& packageName = "", + Broker* broker = 0, + Agent* agent = 0); + void getObjects(std::vector& objects, + const std::map& query, + Broker* broker = 0, + Agent* agent = 0); + */ + }; +} + +#endif + diff --git a/qpid/cpp/src/qmf/Object.h b/qpid/cpp/src/qmf/Object.h index 8caab8d6dc..eb92cbbe45 100644 --- a/qpid/cpp/src/qmf/Object.h +++ b/qpid/cpp/src/qmf/Object.h @@ -31,7 +31,7 @@ namespace qmf { public: Object(const SchemaObjectClass* type); Object(ObjectImpl* impl); - ~Object(); + virtual ~Object(); void destroy(); const ObjectId* getObjectId() const; -- cgit v1.2.1