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

Commands: Add task id to subscription status cmds

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-03-31 17:00:42 +02:00 committed by Dennis Klein
parent c5efd3e4a6
commit 274ba5ec00
6 changed files with 57 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -367,7 +367,7 @@ auto DDS::HandleCmd(const string& id, sdk::cmd::Cmd& cmd, const string& cond, ui
LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState << " to " << senderId; LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState << " to " << senderId;
Cmds outCmds(make<StateChangeSubscription>(id, Result::Ok), Cmds outCmds(make<StateChangeSubscription>(id, fDDSTaskId, Result::Ok),
make<StateChange>(id, fDDSTaskId, fLastState, fCurrentState)); make<StateChange>(id, fDDSTaskId, fLastState, fCurrentState));
fDDS.Send(outCmds.Serialize(), to_string(senderId)); fDDS.Send(outCmds.Serialize(), to_string(senderId));
@ -377,7 +377,7 @@ auto DDS::HandleCmd(const string& id, sdk::cmd::Cmd& cmd, const string& cond, ui
lock_guard<mutex> lock{fStateChangeSubscriberMutex}; lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.erase(senderId); fStateChangeSubscribers.erase(senderId);
} }
Cmds outCmds(make<StateChangeUnsubscription>(id, Result::Ok)); Cmds outCmds(make<StateChangeUnsubscription>(id, fDDSTaskId, Result::Ok));
fDDS.Send(outCmds.Serialize(), to_string(senderId)); fDDS.Send(outCmds.Serialize(), to_string(senderId));
} break; } break;
case Type::get_properties: { case Type::get_properties: {

37
fairmq/sdk/Topology.h
View File

@ -71,7 +71,7 @@ const std::map<DeviceTransition, DeviceState> expectedState =
struct DeviceStatus struct DeviceStatus
{ {
bool initialized; bool subscribed_to_state_changes;
DeviceState lastState; DeviceState lastState;
DeviceState state; DeviceState state;
DDSTask::Id taskId; DDSTask::Id taskId;
@ -265,15 +265,35 @@ class BasicTopology : public AsioBase<Executor, Allocator>
auto HandleCmd(cmd::StateChangeSubscription const& cmd) -> void auto HandleCmd(cmd::StateChangeSubscription const& cmd) -> void
{ {
if (cmd.GetResult() != cmd::Result::Ok) { if (cmd.GetResult() == cmd::Result::Ok) {
FAIR_LOG(error) << "State change subscription failed for " << cmd.GetDeviceId(); DDSTask::Id taskId(cmd.GetTaskId());
std::lock_guard<std::mutex> lk(fMtx);
try {
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.subscribed_to_state_changes = true;
} catch (const std::exception& e) {
FAIR_LOG(error) << "Exception in HandleCmd(cmd::StateChangeSubscription const&): " << e.what();
}
} else {
FAIR_LOG(error) << "State change subscription failed for device: " << cmd.GetDeviceId() << ", task id: " << cmd.GetTaskId();
} }
} }
auto HandleCmd(cmd::StateChangeUnsubscription const& cmd) -> void auto HandleCmd(cmd::StateChangeUnsubscription const& cmd) -> void
{ {
if (cmd.GetResult() != cmd::Result::Ok) { if (cmd.GetResult() == cmd::Result::Ok) {
FAIR_LOG(error) << "State change unsubscription failed for " << cmd.GetDeviceId(); DDSTask::Id taskId(cmd.GetTaskId());
std::lock_guard<std::mutex> lk(fMtx);
try {
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.subscribed_to_state_changes = false;
} catch (const std::exception& e) {
FAIR_LOG(error) << "Exception in HandleCmd(cmd::StateChangeUnsubscription const&): " << e.what();
}
} else {
FAIR_LOG(error) << "State change unsubscription failed for device: " << cmd.GetDeviceId() << ", task id: " << cmd.GetTaskId();
} }
} }
@ -290,7 +310,6 @@ class BasicTopology : public AsioBase<Executor, Allocator>
try { try {
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId)); DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.initialized = true;
task.lastState = cmd.GetLastState(); task.lastState = cmd.GetLastState();
task.state = cmd.GetCurrentState(); task.state = cmd.GetCurrentState();
// FAIR_LOG(debug) << "Updated state entry: taskId=" << taskId << ", state=" << state; // FAIR_LOG(debug) << "Updated state entry: taskId=" << taskId << ", state=" << state;
@ -310,11 +329,11 @@ class BasicTopology : public AsioBase<Executor, Allocator>
{ {
if (cmd.GetResult() != cmd::Result::Ok) { if (cmd.GetResult() != cmd::Result::Ok) {
FAIR_LOG(error) << cmd.GetTransition() << " transition failed for " << cmd.GetDeviceId(); FAIR_LOG(error) << cmd.GetTransition() << " transition failed for " << cmd.GetDeviceId();
DDSTask::Id id(cmd.GetTaskId()); DDSTask::Id taskId(cmd.GetTaskId());
std::lock_guard<std::mutex> lk(fMtx); std::lock_guard<std::mutex> lk(fMtx);
for (auto& op : fChangeStateOps) { for (auto& op : fChangeStateOps) {
if (!op.second.IsCompleted() && op.second.ContainsTask(id) && if (!op.second.IsCompleted() && op.second.ContainsTask(taskId) &&
fStateData.at(fStateIndex.at(id)).state != op.second.GetTargetState()) { fStateData.at(fStateIndex.at(taskId)).state != op.second.GetTargetState()) {
op.second.Complete(MakeErrorCode(ErrorCode::DeviceChangeStateFailed)); op.second.Complete(MakeErrorCode(ErrorCode::DeviceChangeStateFailed));
} }
} }

6
fairmq/sdk/commands/Commands.cxx
View File

@ -293,6 +293,7 @@ string Cmds::Serialize(const Format type) const
auto deviceId = fbb.CreateString(_cmd.GetDeviceId()); auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb); cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId); cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult())); cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
} }
break; break;
@ -301,6 +302,7 @@ string Cmds::Serialize(const Format type) const
auto deviceId = fbb.CreateString(_cmd.GetDeviceId()); auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb); cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId); cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult())); cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
} }
break; break;
@ -433,10 +435,10 @@ void Cmds::Deserialize(const string& str, const Format type)
fCmds.emplace_back(make<Config>(cmdPtr.device_id()->str(), cmdPtr.config_string()->str())); fCmds.emplace_back(make<Config>(cmdPtr.device_id()->str(), cmdPtr.config_string()->str()));
break; break;
case FBCmd_state_change_subscription: case FBCmd_state_change_subscription:
fCmds.emplace_back(make<StateChangeSubscription>(cmdPtr.device_id()->str(), GetResult(cmdPtr.result()))); fCmds.emplace_back(make<StateChangeSubscription>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetResult(cmdPtr.result())));
break; break;
case FBCmd_state_change_unsubscription: case FBCmd_state_change_unsubscription:
fCmds.emplace_back(make<StateChangeUnsubscription>(cmdPtr.device_id()->str(), GetResult(cmdPtr.result()))); fCmds.emplace_back(make<StateChangeUnsubscription>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetResult(cmdPtr.result())));
break; break;
case FBCmd_state_change: case FBCmd_state_change:
fCmds.emplace_back(make<StateChange>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetMQState(cmdPtr.last_state()), GetMQState(cmdPtr.current_state()))); fCmds.emplace_back(make<StateChange>(cmdPtr.device_id()->str(), cmdPtr.task_id(), GetMQState(cmdPtr.last_state()), GetMQState(cmdPtr.current_state())));

16
fairmq/sdk/commands/Commands.h
View File

@ -51,8 +51,8 @@ enum class Type : int
current_state, // args: { device_id, current_state } current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, task_id, Result, transition } transition_status, // args: { device_id, task_id, Result, transition }
config, // args: { device_id, config_string } config, // args: { device_id, config_string }
state_change_subscription, // args: { device_id, Result } state_change_subscription, // args: { device_id, task_id, Result }
state_change_unsubscription, // args: { device_id, Result } state_change_unsubscription, // args: { device_id, task_id, Result }
state_change, // args: { device_id, task_id, last_state, current_state } state_change, // args: { device_id, task_id, last_state, current_state }
properties, // args: { device_id, request_id, Result, properties } properties, // args: { device_id, request_id, Result, properties }
properties_set // args: { device_id, request_id, Result } properties_set // args: { device_id, request_id, Result }
@ -208,37 +208,45 @@ struct Config : Cmd
struct StateChangeSubscription : Cmd struct StateChangeSubscription : Cmd
{ {
explicit StateChangeSubscription(const std::string& id, const Result result) explicit StateChangeSubscription(const std::string& id, const uint64_t taskId, const Result result)
: Cmd(Type::state_change_subscription) : Cmd(Type::state_change_subscription)
, fDeviceId(id) , fDeviceId(id)
, fTaskId(taskId)
, fResult(result) , fResult(result)
{} {}
std::string GetDeviceId() const { return fDeviceId; } std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; } void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
Result GetResult() const { return fResult; } Result GetResult() const { return fResult; }
void SetResult(const Result result) { fResult = result; } void SetResult(const Result result) { fResult = result; }
private: private:
std::string fDeviceId; std::string fDeviceId;
uint64_t fTaskId;
Result fResult; Result fResult;
}; };
struct StateChangeUnsubscription : Cmd struct StateChangeUnsubscription : Cmd
{ {
explicit StateChangeUnsubscription(const std::string& id, const Result result) explicit StateChangeUnsubscription(const std::string& id, const uint64_t taskId, const Result result)
: Cmd(Type::state_change_unsubscription) : Cmd(Type::state_change_unsubscription)
, fDeviceId(id) , fDeviceId(id)
, fTaskId(taskId)
, fResult(result) , fResult(result)
{} {}
std::string GetDeviceId() const { return fDeviceId; } std::string GetDeviceId() const { return fDeviceId; }
void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; } void SetDeviceId(const std::string& deviceId) { fDeviceId = deviceId; }
uint64_t GetTaskId() const { return fTaskId; }
void SetTaskId(const uint64_t taskId) { fTaskId = taskId; }
Result GetResult() const { return fResult; } Result GetResult() const { return fResult; }
void SetResult(const Result result) { fResult = result; } void SetResult(const Result result) { fResult = result; }
private: private:
std::string fDeviceId; std::string fDeviceId;
uint64_t fTaskId;
Result fResult; Result fResult;
}; };

6
fairmq/sdk/commands/CommandsFormat.fbs
View File

@ -54,10 +54,10 @@ enum FBCmd:byte {
set_properties, // args: { request_id, properties } set_properties, // args: { request_id, properties }
current_state, // args: { device_id, current_state } current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, Result, transition } transition_status, // args: { device_id, task_id, Result, transition }
config, // args: { device_id, config_string } config, // args: { device_id, config_string }
state_change_subscription, // args: { device_id, Result } state_change_subscription, // args: { device_id, task_id, Result }
state_change_unsubscription, // args: { device_id, Result } state_change_unsubscription, // args: { device_id, task_id, Result }
state_change, // args: { device_id, task_id, last_state, current_state } state_change, // args: { device_id, task_id, last_state, current_state }
properties, // args: { device_id, request_id, Result, properties } properties, // args: { device_id, request_id, Result, properties }
properties_set // args: { device_id, request_id, Result } properties_set // args: { device_id, request_id, Result }

12
test/commands/_commands.cxx
View File

@ -32,8 +32,8 @@ TEST(Format, Construction)
Cmds currentStateCmds(make<CurrentState>("somedeviceid", State::Running)); Cmds currentStateCmds(make<CurrentState>("somedeviceid", State::Running));
Cmds transitionStatusCmds(make<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop)); Cmds transitionStatusCmds(make<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop));
Cmds configCmds(make<Config>("somedeviceid", "someconfig")); Cmds configCmds(make<Config>("somedeviceid", "someconfig"));
Cmds stateChangeSubscriptionCmds(make<StateChangeSubscription>("somedeviceid", Result::Ok)); Cmds stateChangeSubscriptionCmds(make<StateChangeSubscription>("somedeviceid", 123456, Result::Ok));
Cmds stateChangeUnsubscriptionCmds(make<StateChangeUnsubscription>("somedeviceid", Result::Ok)); Cmds stateChangeUnsubscriptionCmds(make<StateChangeUnsubscription>("somedeviceid", 123456, Result::Ok));
Cmds stateChangeCmds(make<StateChange>("somedeviceid", 123456, State::Running, State::Ready)); Cmds stateChangeCmds(make<StateChange>("somedeviceid", 123456, State::Running, State::Ready));
Cmds propertiesCmds(make<Properties>("somedeviceid", 66, Result::Ok, props)); Cmds propertiesCmds(make<Properties>("somedeviceid", 66, Result::Ok, props));
Cmds propertiesSetCmds(make<PropertiesSet>("somedeviceid", 42, Result::Ok)); Cmds propertiesSetCmds(make<PropertiesSet>("somedeviceid", 42, Result::Ok));
@ -64,9 +64,11 @@ TEST(Format, Construction)
ASSERT_EQ(static_cast<Config&>(configCmds.At(0)).GetConfig(), "someconfig"); ASSERT_EQ(static_cast<Config&>(configCmds.At(0)).GetConfig(), "someconfig");
ASSERT_EQ(stateChangeSubscriptionCmds.At(0).GetType(), Type::state_change_subscription); ASSERT_EQ(stateChangeSubscriptionCmds.At(0).GetType(), Type::state_change_subscription);
ASSERT_EQ(static_cast<StateChangeSubscription&>(stateChangeSubscriptionCmds.At(0)).GetDeviceId(), "somedeviceid"); ASSERT_EQ(static_cast<StateChangeSubscription&>(stateChangeSubscriptionCmds.At(0)).GetDeviceId(), "somedeviceid");
ASSERT_EQ(static_cast<StateChangeSubscription&>(stateChangeSubscriptionCmds.At(0)).GetTaskId(), 123456);
ASSERT_EQ(static_cast<StateChangeSubscription&>(stateChangeSubscriptionCmds.At(0)).GetResult(), Result::Ok); ASSERT_EQ(static_cast<StateChangeSubscription&>(stateChangeSubscriptionCmds.At(0)).GetResult(), Result::Ok);
ASSERT_EQ(stateChangeUnsubscriptionCmds.At(0).GetType(), Type::state_change_unsubscription); ASSERT_EQ(stateChangeUnsubscriptionCmds.At(0).GetType(), Type::state_change_unsubscription);
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(stateChangeUnsubscriptionCmds.At(0)).GetDeviceId(), "somedeviceid"); ASSERT_EQ(static_cast<StateChangeUnsubscription&>(stateChangeUnsubscriptionCmds.At(0)).GetDeviceId(), "somedeviceid");
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(stateChangeUnsubscriptionCmds.At(0)).GetTaskId(), 123456);
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(stateChangeUnsubscriptionCmds.At(0)).GetResult(), Result::Ok); ASSERT_EQ(static_cast<StateChangeUnsubscription&>(stateChangeUnsubscriptionCmds.At(0)).GetResult(), Result::Ok);
ASSERT_EQ(stateChangeCmds.At(0).GetType(), Type::state_change); ASSERT_EQ(stateChangeCmds.At(0).GetType(), Type::state_change);
ASSERT_EQ(static_cast<StateChange&>(stateChangeCmds.At(0)).GetDeviceId(), "somedeviceid"); ASSERT_EQ(static_cast<StateChange&>(stateChangeCmds.At(0)).GetDeviceId(), "somedeviceid");
@ -99,8 +101,8 @@ void fillCommands(Cmds& cmds)
cmds.Add<CurrentState>("somedeviceid", State::Running); cmds.Add<CurrentState>("somedeviceid", State::Running);
cmds.Add<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop); cmds.Add<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop);
cmds.Add<Config>("somedeviceid", "someconfig"); cmds.Add<Config>("somedeviceid", "someconfig");
cmds.Add<StateChangeSubscription>("somedeviceid", Result::Ok); cmds.Add<StateChangeSubscription>("somedeviceid", 123456, Result::Ok);
cmds.Add<StateChangeUnsubscription>("somedeviceid", Result::Ok); cmds.Add<StateChangeUnsubscription>("somedeviceid", 123456, Result::Ok);
cmds.Add<StateChange>("somedeviceid", 123456, State::Running, State::Ready); cmds.Add<StateChange>("somedeviceid", 123456, State::Running, State::Ready);
cmds.Add<Properties>("somedeviceid", 66, Result::Ok, props); cmds.Add<Properties>("somedeviceid", 66, Result::Ok, props);
cmds.Add<PropertiesSet>("somedeviceid", 42, Result::Ok); cmds.Add<PropertiesSet>("somedeviceid", 42, Result::Ok);
@ -164,11 +166,13 @@ void checkCommands(Cmds& cmds)
case Type::state_change_subscription: case Type::state_change_subscription:
++count; ++count;
ASSERT_EQ(static_cast<StateChangeSubscription&>(*cmd).GetDeviceId(), "somedeviceid"); ASSERT_EQ(static_cast<StateChangeSubscription&>(*cmd).GetDeviceId(), "somedeviceid");
ASSERT_EQ(static_cast<StateChangeSubscription&>(*cmd).GetTaskId(), 123456);
ASSERT_EQ(static_cast<StateChangeSubscription&>(*cmd).GetResult(), Result::Ok); ASSERT_EQ(static_cast<StateChangeSubscription&>(*cmd).GetResult(), Result::Ok);
break; break;
case Type::state_change_unsubscription: case Type::state_change_unsubscription:
++count; ++count;
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(*cmd).GetDeviceId(), "somedeviceid"); ASSERT_EQ(static_cast<StateChangeUnsubscription&>(*cmd).GetDeviceId(), "somedeviceid");
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(*cmd).GetTaskId(), 123456);
ASSERT_EQ(static_cast<StateChangeUnsubscription&>(*cmd).GetResult(), Result::Ok); ASSERT_EQ(static_cast<StateChangeUnsubscription&>(*cmd).GetResult(), Result::Ok);
break; break;
case Type::state_change: case Type::state_change: