diff options
Diffstat (limited to 'qpid/cpp/src/qmf/AgentSessionImpl.h')
| -rw-r--r-- | qpid/cpp/src/qmf/AgentSessionImpl.h | 168 |
1 files changed, 168 insertions, 0 deletions
diff --git a/qpid/cpp/src/qmf/AgentSessionImpl.h b/qpid/cpp/src/qmf/AgentSessionImpl.h new file mode 100644 index 0000000000..64a39ab2e8 --- /dev/null +++ b/qpid/cpp/src/qmf/AgentSessionImpl.h @@ -0,0 +1,168 @@ +#ifndef __QMF_AGENT_SESSION_IMPL_H +#define __QMF_AGENT_SESSION_IMPL_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. + * + */ + +#include "qpid/RefCounted.h" +#include "qmf/PrivateImplRef.h" +#include "qmf/exceptions.h" +#include "qmf/AgentSession.h" +#include "qmf/AgentEventImpl.h" +#include "qmf/EventNotifierImpl.h" +#include "qpid/messaging/Connection.h" +#include "qpid/sys/Runnable.h" +#include "qpid/sys/Mutex.h" +#include "qpid/sys/Condition.h" +#include "qpid/sys/Thread.h" +#include "qpid/sys/Runnable.h" +#include "qpid/log/Statement.h" +#include "qpid/messaging/Connection.h" +#include "qpid/messaging/Session.h" +#include "qpid/messaging/Receiver.h" +#include "qpid/messaging/Sender.h" +#include "qpid/messaging/Message.h" +#include "qpid/messaging/AddressParser.h" +#include "qpid/management/Buffer.h" +#include "qpid/RefCounted.h" +#include "qmf/PrivateImplRef.h" +#include "qmf/AgentSession.h" +#include "qmf/exceptions.h" +#include "qmf/AgentSession.h" +#include "qmf/SchemaIdImpl.h" +#include "qmf/SchemaImpl.h" +#include "qmf/DataAddrImpl.h" +#include "qmf/DataImpl.h" +#include "qmf/QueryImpl.h" +#include "qmf/agentCapability.h" +#include "qmf/constants.h" + +#include <queue> +#include <map> + +namespace qmf { + typedef qmf::PrivateImplRef<AgentSession> PI; + + class AgentSessionImpl : public virtual qpid::RefCounted, public qpid::sys::Runnable { + public: + ~AgentSessionImpl(); + + // + // Methods from API handle + // + AgentSessionImpl(qpid::messaging::Connection& c, const std::string& o); + void setDomain(const std::string& d) { checkOpen(); domain = d; } + void setVendor(const std::string& v) { checkOpen(); attributes["_vendor"] = v; } + void setProduct(const std::string& p) { checkOpen(); attributes["_product"] = p; } + void setInstance(const std::string& i) { checkOpen(); attributes["_instance"] = i; } + void setAttribute(const std::string& k, const qpid::types::Variant& v) { checkOpen(); attributes[k] = v; } + const std::string& getName() const { return agentName; } + void open(); + void closeAsync(); + void close(); + bool nextEvent(AgentEvent& e, qpid::messaging::Duration t); + int pendingEvents() const; + + void setEventNotifier(EventNotifierImpl* eventNotifier); + EventNotifierImpl* getEventNotifier() const; + + void registerSchema(Schema& s); + DataAddr addData(Data& d, const std::string& n, bool persist); + void delData(const DataAddr&); + + void authAccept(AgentEvent& e); + void authReject(AgentEvent& e, const std::string& m); + void raiseException(AgentEvent& e, const std::string& s); + void raiseException(AgentEvent& e, const Data& d); + void response(AgentEvent& e, const Data& d); + void complete(AgentEvent& e); + void methodSuccess(AgentEvent& e); + void raiseEvent(const Data& d); + void raiseEvent(const Data& d, int s); + + private: + typedef std::map<DataAddr, Data, DataAddrCompare> DataIndex; + typedef std::map<SchemaId, Schema, SchemaIdCompare> SchemaMap; + + mutable qpid::sys::Mutex lock; + qpid::sys::Condition cond; + qpid::messaging::Connection connection; + qpid::messaging::Session session; + qpid::messaging::Sender directSender; + qpid::messaging::Sender topicSender; + std::string domain; + qpid::types::Variant::Map attributes; + qpid::types::Variant::Map options; + std::string agentName; + bool opened; + std::queue<AgentEvent> eventQueue; + EventNotifierImpl* eventNotifier; + qpid::sys::Thread* thread; + bool threadCanceled; + uint32_t bootSequence; + uint32_t interval; + uint64_t lastHeartbeat; + uint64_t lastVisit; + bool forceHeartbeat; + bool externalStorage; + bool autoAllowQueries; + bool autoAllowMethods; + uint32_t maxSubscriptions; + uint32_t minSubInterval; + uint32_t subLifetime; + bool publicEvents; + bool listenOnDirect; + bool strictSecurity; + uint32_t maxThreadWaitTime; + uint64_t schemaUpdateTime; + std::string directBase; + std::string topicBase; + + SchemaMap schemata; + DataIndex globalIndex; + std::map<SchemaId, DataIndex, SchemaIdCompareNoHash> schemaIndex; + + void checkOpen(); + void setAgentName(); + void enqueueEvent(const AgentEvent&); + void alertEventNotifierLH(bool readable); + void handleLocateRequest(const qpid::types::Variant::List& content, const qpid::messaging::Message& msg); + void handleMethodRequest(const qpid::types::Variant::Map& content, const qpid::messaging::Message& msg); + void handleQueryRequest(const qpid::types::Variant::Map& content, const qpid::messaging::Message& msg); + void handleSchemaRequest(AgentEvent&); + void handleV1SchemaRequest(qpid::management::Buffer&, uint32_t, const qpid::messaging::Message&); + void dispatch(qpid::messaging::Message); + void sendHeartbeat(); + void send(qpid::messaging::Message, const qpid::messaging::Address&); + void flushResponses(AgentEvent&, bool); + void periodicProcessing(uint64_t); + void run(); + }; + + struct AgentSessionImplAccess { + static AgentSessionImpl& get(AgentSession& session); + static const AgentSessionImpl& get(const AgentSession& session); + }; +} + + +#endif + |