Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 08:41:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

SDK: Implement DDS helpers

Browse Source
This commit is contained in:
Dennis Klein 2019-07-19 16:34:44 +02:00 committed by Dennis Klein
parent 90496c89fe
commit cfcdd666bf
3 changed files with 36 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
fairmq/sdk/DDSEnvironment.cxx
View File

@ -33,6 +33,7 @@ auto LoadDDSEnv(const boost::filesystem::path& config_home, const boost::filesys
std::istringstream cmd;
cmd.str("DDS_CFG=`dds-user-defaults --ignore-default-sid -p`\n"
"if [ -z \"$DDS_CFG\" ]; then\n"
" mkdir -p \"$HOME/.DDS\"\n"
" dds-user-defaults --ignore-default-sid -d -c \"$HOME/.DDS/DDS.cfg\"\n"
"fi");
std::system(cmd.str().c_str());

34
fairmq/sdk/DDSSession.cxx
View File

@ -69,7 +69,6 @@ struct DDSSession::Impl
~Impl()
{
fSession.stop();
fSession.shutdown();
}
@ -136,6 +135,39 @@ auto DDSSession::SubmitAgents(Quantity agents, DDSRMSPlugin plugin, const Path&
blocker.Wait();
}
auto DDSSession::RequestAgentInfo() -> void
{
dds::tools_api::SAgentInfoRequestData agentInfoInfo;
tools::Semaphore blocker;
auto agentInfoRequest = dds::tools_api::SAgentInfoRequest::makeRequest(agentInfoInfo);
agentInfoRequest->setResponseCallback(
[&](const dds::tools_api::SAgentInfoResponseData& _response) {
LOG(debug) << "agent: " << _response.m_index << "/" << _response.m_activeAgentsCount;
LOG(debug) << "info: " << _response.m_agentInfo;
});
agentInfoRequest->setMessageCallback(
[](const dds::tools_api::SMessageResponseData& _message) { LOG(debug) << _message; });
agentInfoRequest->setDoneCallback([&]() { blocker.Signal(); });
fImpl->fSession.sendRequest<dds::tools_api::SAgentInfoRequest>(agentInfoRequest);
blocker.Wait();
}
auto DDSSession::ActivateTopology(Path topologyFile) -> void
{
dds::tools_api::STopologyRequestData topologyInfo;
topologyInfo.m_updateType = dds::tools_api::STopologyRequestData::EUpdateType::ACTIVATE;
topologyInfo.m_topologyFile = topologyFile.string();
LOG(warn) << topologyFile.string() << " :::: " << topologyFile;
tools::Semaphore blocker;
auto topologyRequest = dds::tools_api::STopologyRequest::makeRequest(topologyInfo);
topologyRequest->setMessageCallback(
[](const dds::tools_api::SMessageResponseData& _message) { LOG(debug) << _message; });
topologyRequest->setDoneCallback([&]() { blocker.Signal(); });
fImpl->fSession.sendRequest<dds::tools_api::STopologyRequest>(topologyRequest);
blocker.Wait();
}
auto operator<<(std::ostream& os, DDSSession session) -> std::ostream&
{
return os << "$DDS_SESSION_ID: " << session.GetId();

2
fairmq/sdk/DDSSession.h
View File

@ -58,6 +58,8 @@ class DDSSession
auto SubmitAgents(Quantity agents, DDSRMSPlugin plugin) -> void;
auto SubmitAgents(Quantity agents, DDSRMSPlugin plugin, const Path& config) -> void;
auto SubmitAgents(Quantity agents, const Path& config) -> void;
auto RequestAgentInfo() -> void;
auto ActivateTopology(Path topologyFile) -> void;
friend auto operator<<(std::ostream& os, DDSSession session) -> std::ostream&;
private: