Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-14 09:06:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

DDS plugin: refactor to load DDS task id only once

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-03-24 14:37:43 +01:00
parent 38eb9d22e4
commit cd83efadea
2 changed files with 10 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
fairmq/plugins/DDS/DDS.cxx
View File

@ -37,6 +37,7 @@ DDS::DDS(const string& name,
const string& homepage, const string& homepage,
PluginServices* pluginServices) PluginServices* pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices) : Plugin(name, version, maintainer, homepage, pluginServices)
, fDDSTaskId(dds::env_prop<dds::task_id>())
, fCurrentState(DeviceState::Idle) , fCurrentState(DeviceState::Idle)
, fLastState(DeviceState::Idle) , fLastState(DeviceState::Idle)
, fDeviceTerminationRequested(false) , fDeviceTerminationRequested(false)
@ -112,7 +113,7 @@ DDS::DDS(const string& name,
for (auto subscriberId : fStateChangeSubscribers) { for (auto subscriberId : fStateChangeSubscribers) {
LOG(debug) << "Publishing state-change: " << fLastState << "->" << newState << " to " << subscriberId; LOG(debug) << "Publishing state-change: " << fLastState << "->" << newState << " to " << subscriberId;
Cmds cmds(make<StateChange>(id, dds::env_prop<dds::task_id>(), fLastState, fCurrentState)); Cmds cmds(make<StateChange>(id, fDDSTaskId, fLastState, fCurrentState));
fDDS.Send(cmds.Serialize(), to_string(subscriberId)); fDDS.Send(cmds.Serialize(), to_string(subscriberId));
} }
}); });
@ -143,10 +144,10 @@ auto DDS::StartWorkerThread() -> void
auto DDS::WaitForExitingAck() -> void auto DDS::WaitForExitingAck() -> void
{ {
unique_lock<mutex> lock(fStateChangeSubscriberMutex); unique_lock<mutex> lock(fStateChangeSubscriberMutex);
fExitingAcked.wait_for( auto timeout = GetProperty<unsigned int>("wait-for-exiting-ack-timeout");
lock, fExitingAcked.wait_for(lock, chrono::milliseconds(timeout), [this]() {
chrono::milliseconds(GetProperty<unsigned int>("wait-for-exiting-ack-timeout")), return fExitingAckedByLastExternalController;
[this]() { return fExitingAckedByLastExternalController; }); });
} }
auto DDS::FillChannelContainers() -> void auto DDS::FillChannelContainers() -> void
@ -321,11 +322,11 @@ auto DDS::SubscribeForCustomCommands() -> void
Transition transition = static_cast<cmd::ChangeState&>(*cmd).GetTransition(); Transition transition = static_cast<cmd::ChangeState&>(*cmd).GetTransition();
if (ChangeDeviceState(transition)) { if (ChangeDeviceState(transition)) {
cmd::Cmds outCmds( cmd::Cmds outCmds(
cmd::make<cmd::TransitionStatus>(id, dds::env_prop<dds::task_id>(), cmd::Result::Ok, transition)); cmd::make<cmd::TransitionStatus>(id, fDDSTaskId, cmd::Result::Ok, transition));
fDDS.Send(outCmds.Serialize(), to_string(senderId)); fDDS.Send(outCmds.Serialize(), to_string(senderId));
} else { } else {
sdk::cmd::Cmds outCmds( sdk::cmd::Cmds outCmds(
cmd::make<cmd::TransitionStatus>(id, dds::env_prop<dds::task_id>(), cmd::Result::Failure, transition)); cmd::make<cmd::TransitionStatus>(id, fDDSTaskId, cmd::Result::Failure, transition));
fDDS.Send(outCmds.Serialize(), to_string(senderId)); fDDS.Send(outCmds.Serialize(), to_string(senderId));
} }
{ {
@ -359,8 +360,7 @@ auto DDS::SubscribeForCustomCommands() -> void
cmd::Cmds outCmds( cmd::Cmds outCmds(
cmd::make<cmd::StateChangeSubscription>(id, cmd::Result::Ok), cmd::make<cmd::StateChangeSubscription>(id, cmd::Result::Ok),
cmd::make<cmd::StateChange>( cmd::make<cmd::StateChange>(id, fDDSTaskId, fLastState, fCurrentState));
id, dds::env_prop<dds::task_id>(), fLastState, fCurrentState));
fDDS.Send(outCmds.Serialize(), to_string(senderId)); fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break; } break;

1
fairmq/plugins/DDS/DDS.h
View File

@ -144,6 +144,7 @@ class DDS : public Plugin
auto SubscribeForCustomCommands() -> void; auto SubscribeForCustomCommands() -> void;
DDSSubscription fDDS; DDSSubscription fDDS;
size_t fDDSTaskId;
std::unordered_map<std::string, std::vector<std::string>> fBindingChans; std::unordered_map<std::string, std::vector<std::string>> fBindingChans;
std::unordered_map<std::string, DDSConfig> fConnectingChans; std::unordered_map<std::string, DDSConfig> fConnectingChans;