Playgrounds/TelemetryMonitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix disconnections every 20sec and fix Y zoom

Browse Source
This commit is contained in:
Gabriel Lima
2026-04-22 00:22:50 -03:00
parent 65a57b3429
commit 239ca7c5b4
5 changed files with 225 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
server/main.cpp
View File

@@ -27,6 +27,7 @@
#include <string>
#include <string_view>
#include <thread>
#include <vector>
#include "messages.pb.h"
@@ -205,11 +206,12 @@ bool handshake(int fd) {
return write_all(fd, resp.data(), resp.size());
}
// Server-to-client binary frame (unmasked, FIN set).
bool send_binary(int fd, std::span<const uint8_t> payload) {
// Send one server-to-client frame (unmasked). `op_byte` is the full first
// byte — e.g. 0x82 = FIN+binary, 0x8A = FIN+pong, 0x88 = FIN+close.
bool send_frame(int fd, uint8_t op_byte, std::span<const uint8_t> payload) {
std::array<uint8_t, 10> hdr{};
size_t hdr_len;
hdr[0] = 0x82; // FIN=1, opcode=2 (binary)
hdr[0] = op_byte;
if (payload.size() < 126) {
hdr[1] = static_cast<uint8_t>(payload.size());
hdr_len = 2;
@@ -229,36 +231,138 @@ bool send_binary(int fd, std::span<const uint8_t> payload) {
return write_all(fd, payload.data(), payload.size());
}
bool send_binary(int fd, std::span<const uint8_t> payload) {
return send_frame(fd, 0x82, payload);
}
// Try to parse one inbound (client-masked) frame from `in`.
// Returns bytes consumed, 0 if more data is needed, -1 on protocol error.
ssize_t try_parse_frame(std::span<const uint8_t> in, uint8_t& op, bool& fin,
std::vector<uint8_t>& payload) {
constexpr uint64_t kMaxInbound = 1u << 20; // 1 MiB cap is ample for control frames
if (in.size() < 2) return 0;
const uint8_t b0 = in[0];
const uint8_t b1 = in[1];
fin = (b0 & 0x80) != 0;
op = b0 & 0x0f;
const bool masked = (b1 & 0x80) != 0;
uint64_t len = b1 & 0x7fu;
size_t pos = 2;
if (len == 126) {
if (in.size() < 4) return 0;
len = (uint64_t(in[2]) << 8) | in[3];
pos = 4;
} else if (len == 127) {
if (in.size() < 10) return 0;
len = 0;
for (int i = 0; i < 8; ++i) len = (len << 8) | in[2 + i];
pos = 10;
}
if (!masked) return -1; // RFC 6455: client frames MUST be masked.
if (len > kMaxInbound) return -1;
if (in.size() < pos + 4) return 0;
uint8_t mask[4];
std::memcpy(mask, in.data() + pos, 4);
pos += 4;
if (in.size() < pos + len) return 0;
payload.assign(len, 0);
for (uint64_t i = 0; i < len; ++i)
payload[i] = in[pos + i] ^ mask[i & 3];
return static_cast<ssize_t>(pos + len);
}
// Drain any bytes pending on `fd`, parse complete frames out of `rx`, and
// respond to control frames. Returns false on peer close, protocol error,
// or socket error — caller should exit the serve loop.
bool pump_inbox(int fd, std::vector<uint8_t>& rx) {
uint8_t tmp[4096];
while (true) {
ssize_t n = recv(fd, tmp, sizeof(tmp), MSG_DONTWAIT);
if (n > 0) {
rx.insert(rx.end(), tmp, tmp + n);
continue;
}
if (n == 0) return false; // peer closed
if (errno == EAGAIN || errno == EWOULDBLOCK) break;
if (errno == EINTR) continue;
return false;
}
while (true) {
uint8_t op;
bool fin;
std::vector<uint8_t> payload;
ssize_t consumed = try_parse_frame(
{rx.data(), rx.size()}, op, fin, payload);
if (consumed == 0) break; // need more data
if (consumed < 0) return false; // protocol error
rx.erase(rx.begin(), rx.begin() + consumed);
switch (op) {
case 0x9: // ping → pong with echoed payload
if (!send_frame(fd, 0x8A, payload)) return false;
break;
case 0x8: // close → echo close and exit
send_frame(fd, 0x88, payload);
return false;
default:
// continuation/text/binary/pong — ignore for this test server
break;
}
}
return true;
}
// ---------------------------------------------------------------------------
// Synthetic signals.
// Synthetic signals. Each channel has an independent on/off gate that
// simulates sensor outages of varying length, so the client's missing-data
// hatching renders at short, medium, and long gap scales.
// ---------------------------------------------------------------------------
struct ChannelGate {
bool emitting = true;
int remaining = 0; // ticks (ms) until state flip
};
void advance_gate(ChannelGate& g, std::mt19937_64& rng) {
if (--g.remaining > 0) return;
g.emitting = !g.emitting;
std::uniform_real_distribution<double> u(0.0, 1.0);
if (g.emitting) {
g.remaining = std::uniform_int_distribution<int>(3000, 15000)(rng);
} else {
const double r = u(rng);
if (r < 0.60) g.remaining = std::uniform_int_distribution<int>(30, 200)(rng); // short
else if (r < 0.90) g.remaining = std::uniform_int_distribution<int>(300, 1500)(rng); // medium
else g.remaining = std::uniform_int_distribution<int>(2000, 8000)(rng); // long
}
}
void fill_data(pb::DataPacket& p, int64_t now_us, uint64_t tick,
std::mt19937_64& rng) {
std::mt19937_64& rng,
const std::array<ChannelGate, 16>& gates) {
std::normal_distribution<double> noise(0.0, 0.05);
const double t = now_us * 1e-6;
p.set_timestamp_us(now_us);
p.set_ch1(std::sin(2 * M_PI * 1.0 * t) + noise(rng));
p.set_ch2(std::sin(2 * M_PI * 0.3 * t) * 2.0 + noise(rng));
p.set_ch3(std::cos(2 * M_PI * 0.7 * t) + noise(rng));
p.set_ch4(std::fmod(t * 0.5, 1.0) * 2.0 - 1.0); // saw
p.set_ch5(std::fabs(std::fmod(t * 0.3, 2.0) - 1.0) * 2.0 - 1.0); // tri
p.set_ch6(std::sin(2 * M_PI * 0.5 * t) > 0 ? 1.0 : -1.0); // square
p.set_ch7(std::fmod(t * 0.1, 1.0)); // slow ramp
p.set_ch8(noise(rng) * 4.0); // noise
if (gates[0].emitting) p.set_ch1(std::sin(2 * M_PI * 1.0 * t) + noise(rng));
if (gates[1].emitting) p.set_ch2(std::sin(2 * M_PI * 0.3 * t) * 2.0 + noise(rng));
if (gates[2].emitting) p.set_ch3(std::cos(2 * M_PI * 0.7 * t) + noise(rng));
if (gates[3].emitting) p.set_ch4(std::fmod(t * 0.5, 1.0) * 2.0 - 1.0); // saw
if (gates[4].emitting) p.set_ch5(std::fabs(std::fmod(t * 0.3, 2.0) - 1.0) * 2.0 - 1.0); // tri
if (gates[5].emitting) p.set_ch6(std::sin(2 * M_PI * 0.5 * t) > 0 ? 1.0 : -1.0); // square
if (gates[6].emitting) p.set_ch7(std::fmod(t * 0.1, 1.0)); // slow ramp
if (gates[7].emitting) p.set_ch8(noise(rng) * 4.0); // noise
auto sine = [&](int i) {
const double freq = 0.1 * i;
const double phase = i * 0.2;
return std::sin(2 * M_PI * freq * t + phase) + noise(rng) * 0.3;
};
p.set_ch9(sine(9));
p.set_ch10(sine(10));
p.set_ch11(sine(11));
p.set_ch12(sine(12));
p.set_ch13(sine(13));
p.set_ch14(sine(14));
p.set_ch15(sine(15));
p.set_ch16(sine(16));
if (gates[8].emitting) p.set_ch9(sine(9));
if (gates[9].emitting) p.set_ch10(sine(10));
if (gates[10].emitting) p.set_ch11(sine(11));
if (gates[11].emitting) p.set_ch12(sine(12));
if (gates[12].emitting) p.set_ch13(sine(13));
if (gates[13].emitting) p.set_ch14(sine(14));
if (gates[14].emitting) p.set_ch15(sine(15));
if (gates[15].emitting) p.set_ch16(sine(16));
// Statuses rotate once every 2 s, each offset by one step.
if (tick % 2000 == 0) {
@@ -314,17 +418,31 @@ void serve(int fd) {
uint32_t log_seq = 0;
std::string buf;
// Stagger each channel's initial "on" window so outages don't line up.
std::array<ChannelGate, 16> gates;
for (auto& g : gates) {
g.emitting = true;
g.remaining = std::uniform_int_distribution<int>(500, 12000)(rng);
}
// Inbound byte buffer for parsing client control frames (pings, close).
std::vector<uint8_t> rx;
while (true) {
std::this_thread::sleep_until(next);
next += 1ms;
if (!pump_inbox(fd, rx)) break;
const int64_t now_us =
std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::steady_clock::now() - start)
.count();
for (auto& g : gates) advance_gate(g, rng);
pb::Envelope env;
fill_data(*env.mutable_data(), now_us, tick, rng);
fill_data(*env.mutable_data(), now_us, tick, rng, gates);
buf.clear();
env.SerializeToString(&buf);
if (!send_binary(fd, {reinterpret_cast<const uint8_t*>(buf.data()),