137 lines
4.2 KiB
C++
137 lines
4.2 KiB
C++
#pragma once
|
|
|
|
#include "ddkit/access/TopicPool.hpp"
|
|
#include "ddkit/access/Topic.hpp"
|
|
#include "ddkit/access/Writer.hpp"
|
|
|
|
#include "ddkit/utils/system.hpp"
|
|
|
|
namespace ddkit {
|
|
namespace access {
|
|
|
|
class TopicPoolWrite final {
|
|
|
|
public:
|
|
TopicPoolWrite(
|
|
Participant& participant, const IQosStrategyPtr& qosStrategy, const ITopicStrategyPtr& topicStrategy);
|
|
~TopicPoolWrite();
|
|
|
|
// create or get writer
|
|
template <typename T>
|
|
std::shared_ptr<Writer<T>> writer() {
|
|
DDKIT_LOG_TRACE;
|
|
std::unique_lock<std::recursive_mutex> lock(m_mutex);
|
|
|
|
createTopicWriteIf<T>(); // must be done here and not in Topic (because TopicPool and qos access needed)
|
|
|
|
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
|
|
if(topic) {
|
|
auto writer = topic->writer();
|
|
if(!writer){
|
|
ddkit::log::error() << "TopicPoolWrite::writer: no writer in topic";
|
|
}
|
|
return writer;
|
|
} else {
|
|
ddkit::log::error() << "TopicPoolWrite::writer: no topic in topic pool";
|
|
}
|
|
return nullptr;
|
|
}
|
|
|
|
// shortcut, no interface+virtual possible (templates)
|
|
template <class T>
|
|
void write(T const& t) {
|
|
bool isOkUnconnected = false;
|
|
bool isThrowUnconnected = false;
|
|
writeIntern(t, isOkUnconnected, isThrowUnconnected);
|
|
}
|
|
|
|
template <class T>
|
|
void writeUnconnectedOk(T const& t) {
|
|
bool isOkUnconnected = true;
|
|
bool isThrowUnconnected = false;
|
|
writeIntern(t, isOkUnconnected, isThrowUnconnected);
|
|
}
|
|
|
|
template <class T>
|
|
void writeUnconnectedThrow(T const& t) {
|
|
bool isOkUnconnected = false;
|
|
bool isThrowUnconnected = false;
|
|
writeIntern(t, isOkUnconnected, isThrowUnconnected);
|
|
}
|
|
|
|
private:
|
|
template <typename T>
|
|
void writeIntern(T const& t, bool isOkUnconnected, bool isThrowUnconnected) {
|
|
ddkit::log::trace() << "writing " << dds::topic::topic_type_name<T>::value() << ":" << t;
|
|
|
|
createTopicWriteIf<T>();
|
|
|
|
std::unique_lock<std::recursive_mutex> lock(m_mutex);
|
|
|
|
auto currentWriter = writer<T>();
|
|
if(currentWriter){
|
|
currentWriter->send(t, isOkUnconnected, isThrowUnconnected);
|
|
} else {
|
|
ddkit::log::error() << "TopicPoolWrite::writeIntern: no writer in topic pool";
|
|
}
|
|
}
|
|
|
|
template <typename T>
|
|
void createTopicWrite() {
|
|
std::unique_lock<std::recursive_mutex> lock(m_mutex);
|
|
std::string topicName = m_topicPool.getTopicName<T>();
|
|
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
|
|
std::string typeName = topic->typeName();
|
|
|
|
if(!topic) {
|
|
ddkit::log::error() << "TopicPoolWrite::createTopicWrite: no topic in topic pool.";
|
|
return;
|
|
}
|
|
if(topic->hasWriter()) {
|
|
ddkit::log::error() << "TopicPoolWrite::createTopicWrite: Writer (" << topicName << ") already exists.";
|
|
return;
|
|
}
|
|
|
|
|
|
QosPattern qosPattern = m_topicPool.qosPattern(typeName);
|
|
dds::pub::qos::DataWriterQos dataWriterQos = qosPattern.qosWrite();
|
|
dds::pub::qos::PublisherQos pubQos = qosPattern.qosPub();
|
|
|
|
auto writer = std::make_shared<Writer<T>>(topic, dataWriterQos, pubQos);
|
|
topic->setWriter(writer);
|
|
}
|
|
|
|
template <class T>
|
|
void createTopicWriteIf() {
|
|
std::unique_lock<std::recursive_mutex> lock(m_mutex);
|
|
std::shared_ptr<Topic<T>> topic = m_topicPool.topic<T>();
|
|
if (topic && !topic->hasWriter()) {
|
|
createTopicWrite<T>();
|
|
}
|
|
}
|
|
|
|
public:
|
|
template <class T>
|
|
void dispose(T const& t) {
|
|
std::unique_lock<std::recursive_mutex> lock(m_mutex);
|
|
|
|
auto currentWriter = writer<T>();
|
|
if(currentWriter){
|
|
currentWriter->dispose(t);
|
|
} else {
|
|
ddkit::log::error() << "TopicPoolWrite::dispose: no writer in topic pool";
|
|
}
|
|
}
|
|
|
|
void closeWriters();
|
|
|
|
private:
|
|
std::recursive_mutex m_mutex;
|
|
TopicPool m_topicPool;
|
|
|
|
TopicPoolWrite(TopicPoolWrite const&) = delete;
|
|
void operator=(TopicPoolWrite const&) = delete;
|
|
};
|
|
|
|
} // namespace access
|
|
} // namespace ddkit
|