#pragma once #include "ddkit/utils/assert.hpp" #include "ddkit/access/TopicPoolRead.hpp" #include "ddkit/access/TopicPoolWrite.hpp" #include "ddkit/actors/CallbackInfo.hpp" #include "ddkit/actors/IActor.hpp" #include "ddkit/actors/SessionImpl.hpp" #include "ddkit/idls/ActorInfo.hpp" #include "ddkit/actors/Key.hpp" #include "ddkit/utils/CallbackWrapper.hpp" #include #include namespace ddkit { namespace actors { class SimpleActor : public IActor { public: SimpleActor(const SessionImplPtr& sessionImpl, const std::string& serviceName, bool isService, const std::string& module); ~SimpleActor(); bool hasObserver(); bool hasSender(); access::TopicPoolRead& topicPoolRead(); access::TopicPoolWrite& topicPoolWrite(); std::string moduleName() const; bool isService() const; int domain() const; std::string serviceName() const; ActorConfiguration config() const; std::string actorId() const; bool operator==(const SimpleActor& other) const; void setClientName(const std::string& clientName); void waitUntilConnected(); void addConnectCheck(access::IConnectCheckPtr connectCheck); // template-class-parameters Observer and Sender in the following methods must be the 'reader-class' // respectively the 'writer-class' of one of the rmodel-modules. They are called Observer and Sender // in the corresponding namespace and inherit the interface IObserver/ISender. They define a typedef TypeUser // to refer to the API-Interface-Type IClient/IService. Sender also inherits its TypeUser-Type /// @brief - needed for access from generated Service/Client, to call write-stuff template ISenderPtr sender() { DDKIT_LOG_TRACE; checkSenderType(); if (hasSender()) { std::shared_ptr sender = std::dynamic_pointer_cast(m_sender); // Sender inherits ISender if (sender) { checkTypeOfSender(sender.get()); return sender; } } return nullptr; } ISenderPtr sender() { return m_sender; } IObserverPtr observer() { return m_observer; } bool hasSession() { return !m_sessionImplWeak.expired(); } template IObserverPtr createObserver(typename Observer::TypeUser& callbacks, const std::vector& methods) { ddkit::log::info() << "Create observer"; SessionImplPtr sessionImpl = m_sessionImplWeak.lock(); if (!sessionImpl) { throw std::runtime_error("Failure locking std::weak_ptr in ActorImpl"); } checkObserverType(); typename Observer::TypeUser* userCallback = &callbacks; // (keep * inside ... else problems with capturing) // insert observer DDKIT_THROW_ON_FALSE_INFO( !hasObserver(), "Inserting an observer to an actor already containing one." "Called ObsFactory::add() twice? (same serviceName + domain)"); m_observer = std::make_shared(*this, m_topicPoolRead, methods, userCallback); DDKIT_THROW_ON_FALSE_INFO(hasObserver(), "creation of observer failed"); // update actorInfo if (m_actorInfo.isReading()) { ddkit::log::warn() << "SimpleActor::createObserver: " << "actor is already reading"; } m_actorInfo.isReading() = true; m_actorInfo.openReaders() = methods; // only vectors on dds sessionImpl->publishActorInfo(m_actorInfo); return m_observer; } template IObserverPtr createObserver(std::map> callbacks, const std::vector methods) { ddkit::log::info() << "Create observer"; SessionImplPtr sessionImpl = m_sessionImplWeak.lock(); if (!sessionImpl) { throw std::runtime_error("Failure locking std::weak_ptr in ActorImpl"); } checkObserverType(); // insert observer DDKIT_THROW_ON_FALSE_INFO( !hasObserver(), "Inserting an observer to an actor already containing one." "Called ObsFactory::add() twice? (same serviceName + domain)"); m_observer = std::make_shared(*this, m_topicPoolRead, methods, callbacks); DDKIT_THROW_ON_FALSE_INFO(hasObserver(), "creation of observer failed"); // update actorInfo if(m_actorInfo.isReading()) { ddkit::log::warn() << "SimpleActor::createObserver: actor is already reading"; } m_actorInfo.isReading() = true; m_actorInfo.openReaders() = methods; // only vectors on dds sessionImpl->publishActorInfo(m_actorInfo); return m_observer; } template ISenderPtr createSender(const std::vector& methods) { ddkit::log::info() << "Create sender"; SessionImplPtr sessionImpl = m_sessionImplWeak.lock(); if (!sessionImpl) { throw std::runtime_error("Failure locking std::weak_ptr in ActorImpl"); } checkSenderType(); DDKIT_THROW_ON_FALSE_INFO( !hasSender(), "Inserting a sender to an actor already containing one. " "Called Factory::create() twice? (same serviceName + domain)"); m_sender = std::make_shared(*this, m_topicPoolWrite); Sender* sender2 = dynamic_cast(m_sender.get()); checkTypeOfSender(sender2); DDKIT_THROW_ON_FALSE(sender2); DDKIT_THROW_ON_FALSE(hasSender()); // update actorInfo if (m_actorInfo.isWriting()) { ddkit::log::warn() << "SimpleActor::createSender: " << "actor is already writing"; } m_actorInfo.isWriting() = true; m_actorInfo.openWriters() = methods; // only vectors on dds sessionImpl->publishActorInfo(m_actorInfo); return m_sender; } void discardSender(); void discardObserver(); bool isValid() override; void shutdown() override; bool shutdownRequested(); CallbackInfo callbackInfoBase(); CallbackInfo callbackInfo(); void setCallbackInfo(CallbackInfo callbackInfo); void resetCallbackInfo(); private: ddkit_intern::ActorInfo m_actorInfo; access::IQosStrategyPtr m_qosStrategy; access::ITopicStrategyPtr m_topicStrategy; access::TopicPoolRead m_topicPoolRead; access::TopicPoolWrite m_topicPoolWrite; std::vector m_connectCheckSet; IObserverPtr m_observer; // or nullptr, if none ISenderPtr m_sender; // or nullptr, if none CallbackInfo m_callbackInfoBase; std::atomic m_shutdownRequested; }; } // namespace actors } // namespace ddkit