Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

TEL - Create Latency Node #127

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

TEL - Create Latency Node #127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions lib/core/mqtt_topics.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@ enum class MqttTopic {
kDisplacement,
kVelocity,
kAcceleration,
kLogs
kLogs,
kLatencyRequest,
kLatencyResponse
};

const std::unordered_map<MqttTopic, std::string> mqtt_topic_to_string
Expand All @@ -27,7 +29,9 @@ const std::unordered_map<MqttTopic, std::string> mqtt_topic_to_string
{MqttTopic::kDisplacement, "hyped/cart_2024/navigation/displacement"},
{MqttTopic::kVelocity, "hyped/cart_2024/navigation/velocity"},
{MqttTopic::kAcceleration, "hyped/cart_2024/navigation/acceleration"},
{MqttTopic::kLogs, "hyped/cart_2024/logs"}};
{MqttTopic::kLogs, "hyped/cart_2024/logs"},
{MqttTopic::kLatencyRequest, "hyped/cart_2024/latency/request"},
{MqttTopic::kLatencyResponse, "hyped/cart_2024/latency/response"}};

const std::unordered_map<std::string, MqttTopic> mqtt_string_to_topic
= {{"hyped/cart_2024/state/state", MqttTopic::kState},
Expand All @@ -38,6 +42,8 @@ const std::unordered_map<std::string, MqttTopic> mqtt_string_to_topic
{"hyped/cart_2024/navigation/displacement", MqttTopic::kDisplacement},
{"hyped/cart_2024/navigation/velocity", MqttTopic::kVelocity},
{"hyped/cart_2024/navigation/acceleration", MqttTopic::kAcceleration},
{"hyped/cart_2024/logs", MqttTopic::kLogs}};
{"hyped/cart_2024/logs", MqttTopic::kLogs},
{"hyped/cart_2024/latency/request", MqttTopic::kLatencyRequest},
{"hyped/cart_2024/latency/response", MqttTopic::kLatencyResponse}};

} // namespace hyped::core
46 changes: 46 additions & 0 deletions lib/telemetry/latency.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
#include "latency.hpp"

#include <core/wall_clock.hpp>

namespace hyped::telemetry {

Latency::Latency(std::shared_ptr<core::IMqtt> mqtt)
{
mqtt_->subscribe(core::MqttTopic::kStateRequest);
}

void Latency::respond()
{
const auto nextMessage = mqtt_->getMessage();
if (!nextMessage) { return; }

const auto payload = nextMessage->payload;
const auto topic = core::MqttTopic::kLatencyResponse;
const core::MqttMessage::Header header{.timestamp = 0,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you need a WallClock as a data member, then:
.timestamp = time_.now().time_since_epoch().count()

.priority = core::MqttMessagePriority::kNormal};
const core::MqttMessage message{topic, header, payload};
mqtt_->publish(message, core::MqttMessageQos::kAtLeastOnce);
}

void Latency::run()
{
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

while (true) otherwise this runs once lol

mqtt_->consume();
respond();
}

core::Result Latency::startNode(const std::string &mqtt_ip, const std::uint32_t mqtt_port)
{
core::WallClock wall_clock;
core::Logger logger("LATENCY", core::LogLevel::kDebug, wall_clock);
auto optional_mqtt = core::Mqtt::create(logger, "latency", mqtt_ip, mqtt_port);
if (!optional_mqtt) {
logger.log(core::LogLevel::kFatal, "Failed to create MQTT client");
return core::Result::kFailure;
}
auto mqtt = *optional_mqtt;
Latency latency(mqtt);
latency.run();
return core::Result::kSuccess;
}

} // namespace hyped::telemetry
18 changes: 18 additions & 0 deletions lib/telemetry/latency.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
#pragma once

#include <core/mqtt.hpp>
#include <core/types.hpp>

namespace hyped::telemetry {

class Latency {
public:
Latency(std::shared_ptr<core::IMqtt> mqtt);
void respond();
void run();
static core::Result startNode(const std::string &mqtt_ip, const std::uint32_t mqtt_port);

const std::shared_ptr<core::IMqtt> mqtt_;
};

} // namespace hyped::telemetry
4 changes: 2 additions & 2 deletions telemetry/packages/ui/app/context/pods.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ export const PodsProvider = ({ children }: { children: React.ReactNode }) => {
publish(
'latency/request',
JSON.stringify({
latency: new Date().getTime().toString(),
timestamp: new Date().getTime().toString(),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

here and elsewhere, are you remembering that all mqtt messages are of the format:

{
  header : {
    timestamp: int,
    priority: int
  },
  payload: { rest of data }
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ahhh I see. Will need to change this throughout the telemetry code

}),
podId,
);
Expand Down Expand Up @@ -233,7 +233,7 @@ export const PodsProvider = ({ children }: { children: React.ReactNode }) => {
// calculate the latency
const latency =
new Date().getTime() -
parseInt(JSON.parse(message.toString())['latency'] as string);
parseInt(JSON.parse(message.toString())['timestamp'] as string);

// send warning to the server if the latency is too high
if (latency > POD_WARNING_LATENCY) {
Expand Down
Loading