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

Add SubscriptionHeartbeat command

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-04-08 15:15:17 +02:00 committed by Dennis Klein
parent 7cbd154344
commit 330687772f
4 changed files with 60 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -46,7 +46,7 @@ array<string, 2> resultNames =
}
};
array<string, 16> typeNames =
array<string, 17> typeNames =
{
{
"CheckState",
@ -57,6 +57,7 @@ array<string, 16> typeNames =
"StateChangeExitingReceived",
"GetProperties",
"SetProperties",
"SubscriptionHeartbeat",
"CurrentState",
"TransitionStatus",
@ -147,7 +148,7 @@ array<sdk::cmd::FBTransition, 12> mqTransitionToFBTransition =
}
};
array<FBCmd, 16> typeToFBCmd =
array<FBCmd, 17> typeToFBCmd =
{
{
FBCmd::FBCmd_check_state,
@ -158,6 +159,7 @@ array<FBCmd, 16> typeToFBCmd =
FBCmd::FBCmd_state_change_exiting_received,
FBCmd::FBCmd_get_properties,
FBCmd::FBCmd_set_properties,
FBCmd::FBCmd_subscription_heartbeat,
FBCmd::FBCmd_current_state,
FBCmd::FBCmd_transition_status,
FBCmd::FBCmd_config,
@ -169,7 +171,7 @@ array<FBCmd, 16> typeToFBCmd =
}
};
array<Type, 16> fbCmdToType =
array<Type, 17> fbCmdToType =
{
{
Type::check_state,
@ -180,6 +182,7 @@ array<Type, 16> fbCmdToType =
Type::state_change_exiting_received,
Type::get_properties,
Type::set_properties,
Type::subscription_heartbeat,
Type::current_state,
Type::transition_status,
Type::config,
@ -228,7 +231,9 @@ string Cmds::Serialize(const Format type) const
break;
break;
case Type::subscribe_to_state_change: {
auto _cmd = static_cast<SubscribeToStateChange&>(*cmd);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::unsubscribe_from_state_change: {
@ -261,6 +266,12 @@ string Cmds::Serialize(const Format type) const
cmdBuilder->add_properties(props);
}
break;
case Type::subscription_heartbeat: {
auto _cmd = static_cast<SubscriptionHeartbeat&>(*cmd);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::current_state: {
auto _cmd = static_cast<CurrentState&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
@ -406,7 +417,7 @@ void Cmds::Deserialize(const string& str, const Format type)
fCmds.emplace_back(make<DumpConfig>());
break;
case FBCmd_subscribe_to_state_change:
fCmds.emplace_back(make<SubscribeToStateChange>());
fCmds.emplace_back(make<SubscribeToStateChange>(cmdPtr.interval()));
break;
case FBCmd_unsubscribe_from_state_change:
fCmds.emplace_back(make<UnsubscribeFromStateChange>());
@ -425,6 +436,9 @@ void Cmds::Deserialize(const string& str, const Format type)
}
fCmds.emplace_back(make<SetProperties>(cmdPtr.request_id(), properties));
} break;
case FBCmd_subscription_heartbeat:
fCmds.emplace_back(make<SubscriptionHeartbeat>(cmdPtr.interval()));
break;
case FBCmd_current_state:
fCmds.emplace_back(make<CurrentState>(cmdPtr.device_id()->str(), GetMQState(cmdPtr.current_state())));
break;

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

@ -47,6 +47,7 @@ enum class Type : int
state_change_exiting_received, // args: { }
get_properties, // args: { request_id, property_query }
set_properties, // args: { request_id, properties }
subscription_heartbeat, // args: { interval }
current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, task_id, Result, transition }
@ -95,7 +96,16 @@ struct DumpConfig : Cmd
struct SubscribeToStateChange : Cmd
{
explicit SubscribeToStateChange() : Cmd(Type::subscribe_to_state_change) {}
explicit SubscribeToStateChange(int64_t interval)
: Cmd(Type::subscribe_to_state_change)
, fInterval(interval)
{}
int64_t GetInterval() const { return fInterval; }
void SetInterval(int64_t interval) { fInterval = interval; }
private:
int64_t fInterval;
};
struct UnsubscribeFromStateChange : Cmd
@ -144,6 +154,20 @@ struct SetProperties : Cmd
std::vector<std::pair<std::string, std::string>> fProperties;
};
struct SubscriptionHeartbeat : Cmd
{
explicit SubscriptionHeartbeat(int64_t interval)
: Cmd(Type::subscription_heartbeat)
, fInterval(interval)
{}
int64_t GetInterval() const { return fInterval; }
void SetInterval(int64_t interval) { fInterval = interval; }
private:
int64_t fInterval;
};
struct CurrentState : Cmd
{
explicit CurrentState(const std::string& id, State currentState)

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

@ -47,11 +47,12 @@ enum FBCmd:byte {
check_state, // args: { }
change_state, // args: { transition }
dump_config, // args: { }
subscribe_to_state_change, // args: { }
subscribe_to_state_change, // args: { interval }
unsubscribe_from_state_change, // args: { }
state_change_exiting_received, // args: { }
get_properties, // args: { request_id, property_query }
set_properties, // args: { request_id, properties }
subscription_heartbeat, // args: { interval }
current_state, // args: { device_id, current_state }
transition_status, // args: { device_id, task_id, Result, transition }
@ -68,6 +69,7 @@ table FBCommand {
device_id:string;
task_id:uint64;
request_id:uint64;
interval:int64;
state:FBState;
transition:FBTransition;
result:FBResult;

18
test/commands/_commands.cxx
View File

@ -24,11 +24,12 @@ TEST(Format, Construction)
Cmds checkStateCmds(make<CheckState>());
Cmds changeStateCmds(make<ChangeState>(Transition::Stop));
Cmds dumpConfigCmds(make<DumpConfig>());
Cmds subscribeToStateChangeCmds(make<SubscribeToStateChange>());
Cmds subscribeToStateChangeCmds(make<SubscribeToStateChange>(60000));
Cmds unsubscribeFromStateChangeCmds(make<UnsubscribeFromStateChange>());
Cmds stateChangeExitingReceivedCmds(make<StateChangeExitingReceived>());
Cmds getPropertiesCmds(make<GetProperties>(66, "k[12]"));
Cmds setPropertiesCmds(make<SetProperties>(42, props));
Cmds subscriptionHeartbeatCmds(make<SubscriptionHeartbeat>(60000));
Cmds currentStateCmds(make<CurrentState>("somedeviceid", State::Running));
Cmds transitionStatusCmds(make<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop));
Cmds configCmds(make<Config>("somedeviceid", "someconfig"));
@ -43,6 +44,7 @@ TEST(Format, Construction)
ASSERT_EQ(static_cast<ChangeState&>(changeStateCmds.At(0)).GetTransition(), Transition::Stop);
ASSERT_EQ(dumpConfigCmds.At(0).GetType(), Type::dump_config);
ASSERT_EQ(subscribeToStateChangeCmds.At(0).GetType(), Type::subscribe_to_state_change);
ASSERT_EQ(static_cast<SubscribeToStateChange&>(subscribeToStateChangeCmds.At(0)).GetInterval(), 60000);
ASSERT_EQ(unsubscribeFromStateChangeCmds.At(0).GetType(), Type::unsubscribe_from_state_change);
ASSERT_EQ(stateChangeExitingReceivedCmds.At(0).GetType(), Type::state_change_exiting_received);
ASSERT_EQ(getPropertiesCmds.At(0).GetType(), Type::get_properties);
@ -51,6 +53,8 @@ TEST(Format, Construction)
ASSERT_EQ(setPropertiesCmds.At(0).GetType(), Type::set_properties);
ASSERT_EQ(static_cast<SetProperties&>(setPropertiesCmds.At(0)).GetRequestId(), 42);
ASSERT_EQ(static_cast<SetProperties&>(setPropertiesCmds.At(0)).GetProps(), props);
ASSERT_EQ(subscriptionHeartbeatCmds.At(0).GetType(), Type::subscription_heartbeat);
ASSERT_EQ(static_cast<SubscriptionHeartbeat&>(subscriptionHeartbeatCmds.At(0)).GetInterval(), 60000);
ASSERT_EQ(currentStateCmds.At(0).GetType(), Type::current_state);
ASSERT_EQ(static_cast<CurrentState&>(currentStateCmds.At(0)).GetDeviceId(), "somedeviceid");
ASSERT_EQ(static_cast<CurrentState&>(currentStateCmds.At(0)).GetCurrentState(), State::Running);
@ -93,11 +97,12 @@ void fillCommands(Cmds& cmds)
cmds.Add<CheckState>();
cmds.Add<ChangeState>(Transition::Stop);
cmds.Add<DumpConfig>();
cmds.Add<SubscribeToStateChange>();
cmds.Add<SubscribeToStateChange>(60000);
cmds.Add<UnsubscribeFromStateChange>();
cmds.Add<StateChangeExitingReceived>();
cmds.Add<GetProperties>(66, "k[12]");
cmds.Add<SetProperties>(42, props);
cmds.Add<SubscriptionHeartbeat>(60000);
cmds.Add<CurrentState>("somedeviceid", State::Running);
cmds.Add<TransitionStatus>("somedeviceid", 123456, Result::Ok, Transition::Stop);
cmds.Add<Config>("somedeviceid", "someconfig");
@ -110,7 +115,7 @@ void fillCommands(Cmds& cmds)
void checkCommands(Cmds& cmds)
{
ASSERT_EQ(cmds.Size(), 16);
ASSERT_EQ(cmds.Size(), 17);
int count = 0;
auto const props(std::vector<std::pair<std::string, std::string>>({{"k1", "v1"}, {"k2", "v2"}}));
@ -129,6 +134,7 @@ void checkCommands(Cmds& cmds)
break;
case Type::subscribe_to_state_change:
++count;
ASSERT_EQ(static_cast<SubscribeToStateChange&>(*cmd).GetInterval(), 60000);
break;
case Type::unsubscribe_from_state_change:
++count;
@ -146,6 +152,10 @@ void checkCommands(Cmds& cmds)
ASSERT_EQ(static_cast<SetProperties&>(*cmd).GetRequestId(), 42);
ASSERT_EQ(static_cast<SetProperties&>(*cmd).GetProps(), props);
break;
case Type::subscription_heartbeat:
++count;
ASSERT_EQ(static_cast<SubscriptionHeartbeat&>(*cmd).GetInterval(), 60000);
break;
case Type::current_state:
++count;
ASSERT_EQ(static_cast<CurrentState&>(*cmd).GetDeviceId(), "somedeviceid");
@ -201,7 +211,7 @@ void checkCommands(Cmds& cmds)
}
}
ASSERT_EQ(count, 16);
ASSERT_EQ(count, 17);
}
TEST(Format, SerializationBinary)