martino/marte-debug
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
marte-debug/Source/Components/Interfaces/WebDebugService/WebDebugService.cpp
T
Martino Ferrari f6eb0a7056 Updated without Vector
2026-05-20 11:30:05 +02:00

565 lines
21 KiB
C++
Raw Permalink Blame History

#include "Atomic.h"
#include "Logger.h"
#include "BasicTCPSocket.h"
#include "ConfigurationDatabase.h"
#include "DebugBrokerWrapper.h"
#include "GAM.h"
#include "HighResolutionTimer.h"
#include "ObjectRegistryDatabase.h"
#include "ReferenceT.h"
#include "Sleep.h"
#include "StreamString.h"
#include "Threads.h"
#include "TimeoutType.h"
#include "TypeConversion.h"
#include "WebDebugService.h"
#include "WebUI.h"
namespace MARTe {
// ---------------------------------------------------------------------------
// CLASS_REGISTER
// ---------------------------------------------------------------------------
CLASS_REGISTER(WebDebugService, "1.0")
// ---------------------------------------------------------------------------
// Helper: convert raw bytes to float64 for SSE JSON
// ---------------------------------------------------------------------------
static float64 WDS_ToFloat64(const uint8 *data, TypeDescriptor td) {
float64 v = 0.0;
if (td == Float32Bit) { float32 f; memcpy(&f, data, 4u); v = (float64)f; }
else if (td == Float64Bit) { memcpy(&v, data, 8u); }
else if (td == SignedInteger8Bit) { int8 x; memcpy(&x, data, 1u); v = (float64)x; }
else if (td == UnsignedInteger8Bit) { uint8 x; memcpy(&x, data, 1u); v = (float64)x; }
else if (td == SignedInteger16Bit) { int16 x; memcpy(&x, data, 2u); v = (float64)x; }
else if (td == UnsignedInteger16Bit){ uint16 x; memcpy(&x, data, 2u); v = (float64)x; }
else if (td == SignedInteger32Bit) { int32 x; memcpy(&x, data, 4u); v = (float64)x; }
else if (td == UnsignedInteger32Bit){ uint32 x; memcpy(&x, data, 4u); v = (float64)x; }
else if (td == SignedInteger64Bit) { int64 x; memcpy(&x, data, 8u); v = (float64)x; }
else if (td == UnsignedInteger64Bit){ uint64 x; memcpy(&x, data, 8u); v = (float64)(int64)x; }
return v;
}
static void WDS_EscapeJson(const char8 *src, StreamString &dst) {
if (src == NULL_PTR(const char8 *)) return;
while (*src != '\0') {
if (*src == '"') dst += "\\\"";
else if (*src == '\\') dst += "\\\\";
else if (*src == '\n') dst += "\\n";
else if (*src == '\r') dst += "\\r";
else if (*src == '\t') dst += "\\t";
else dst += *src;
src++;
}
}
// ---------------------------------------------------------------------------
// Constructor / Destructor
// ---------------------------------------------------------------------------
WebDebugService::WebDebugService()
: DebugServiceBase(),
EmbeddedServiceMethodBinderI(),
binderHttp(this, WdsBinder::Http),
binderStream(this, WdsBinder::Stream),
httpService(binderHttp),
streamerService(binderStream) {
httpPort = 8090u;
streamerSeq = 0u;
streamerT0Ns = 0u;
logHistoryWriteIdx = 0u;
logHistoryFill = 0u;
memset(logHistory, 0, sizeof(logHistory));
for (uint32 i = 0u; i < MAX_SSE_CLIENTS; i++) {
sseClients[i] = NULL_PTR(BasicTCPSocket *);
}
}
WebDebugService::~WebDebugService() {
if (DebugServiceI::GetInstance() == this) {
DebugServiceI::SetInstance(NULL_PTR(DebugServiceI *));
}
httpService.Stop();
streamerService.Stop();
tcpServer.Close();
sseMutex.FastLock();
for (uint32 i = 0u; i < MAX_SSE_CLIENTS; i++) {
if (sseClients[i] != NULL_PTR(BasicTCPSocket *)) {
sseClients[i]->Close();
delete sseClients[i];
sseClients[i] = NULL_PTR(BasicTCPSocket *);
}
}
sseMutex.FastUnLock();
// signals owned by DebugServiceBase destructor
}
// ---------------------------------------------------------------------------
// Initialise
// ---------------------------------------------------------------------------
bool WebDebugService::Initialise(StructuredDataI &data) {
if (!ReferenceContainer::Initialise(data)) return false;
uint32 port = 8090u;
(void)data.Read("HttpPort", port);
httpPort = (uint16)port;
if (!traceBuffer.Init(4 * 1024 * 1024)) return false;
DebugServiceI::SetInstance(this);
PatchRegistry();
// Initialise Logger early so REPORT_ERROR calls are captured.
(void)Logger::Instance();
REPORT_ERROR(ErrorManagement::Information,
"WebDebugService initialised on port %d", (int)httpPort);
// Record time origin for relative SSE timestamps.
streamerT0Ns = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1.0e9);
ConfigurationDatabase threadData;
threadData.Write("Timeout", (uint32)1000);
httpService.Initialise(threadData);
streamerService.Initialise(threadData);
if (!tcpServer.Open()) return false;
if (!tcpServer.Listen(httpPort)) return false;
if (httpService.Start() != ErrorManagement::NoError) return false;
if (streamerService.Start() != ErrorManagement::NoError) return false;
return true;
}
// ---------------------------------------------------------------------------
// EmbeddedServiceMethodBinderI — never actually called (sub-binders used)
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType WebDebugService::Execute(ExecutionInfo &info) {
(void)info;
return ErrorManagement::FatalError;
}
// ---------------------------------------------------------------------------
// SERVICE_INFO hook
// ---------------------------------------------------------------------------
void WebDebugService::GetServiceInfo(StreamString &out) {
out.Printf("OK SERVICE_INFO HTTP:%u STATE:%s\n",
(uint32)httpPort, isPaused ? "PAUSED" : "RUNNING");
}
// ---------------------------------------------------------------------------
// SSE broadcast
// ---------------------------------------------------------------------------
void WebDebugService::BroadcastSse(const char8 *eventType,
const StreamString &data) {
StreamString pkt;
pkt.Printf("event: %s\ndata: ", eventType);
pkt += data;
pkt += "\n\n";
sseMutex.FastLock();
for (uint32 i = 0u; i < MAX_SSE_CLIENTS; i++) {
if (sseClients[i] != NULL_PTR(BasicTCPSocket *)) {
uint32 sz = pkt.Size();
if (!sseClients[i]->Write(pkt.Buffer(), sz)) {
sseClients[i]->Close();
delete sseClients[i];
sseClients[i] = NULL_PTR(BasicTCPSocket *);
}
}
}
sseMutex.FastUnLock();
}
void WebDebugService::RemoveDeadSseClients() {
sseMutex.FastLock();
for (uint32 i = 0u; i < MAX_SSE_CLIENTS; i++) {
if (sseClients[i] != NULL_PTR(BasicTCPSocket *) &&
!sseClients[i]->IsConnected()) {
sseClients[i]->Close();
delete sseClients[i];
sseClients[i] = NULL_PTR(BasicTCPSocket *);
}
}
sseMutex.FastUnLock();
}
void WebDebugService::UpgradeToSse(BasicTCPSocket *client) {
const char8 *hdr =
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/event-stream\r\n"
"Cache-Control: no-cache\r\n"
"Connection: keep-alive\r\n"
"Access-Control-Allow-Origin: *\r\n"
"\r\n";
uint32 sz = StringHelper::Length(hdr);
(void)client->Write(hdr, sz);
// Send an initial status event
mutex.FastLock();
bool paused = isPaused;
StreamString gam = pausedAtGam;
uint32 rem = stepRemaining;
mutex.FastUnLock();
StreamString initEv;
initEv.Printf(
"{\"type\":\"status\",\"paused\":%s,\"gam\":\"%s\",\"remaining\":%u}",
paused ? "true" : "false", gam.Buffer(), rem);
StreamString initPkt;
initPkt += "event: message\ndata: ";
initPkt += initEv;
initPkt += "\n\n";
uint32 isz = initPkt.Size();
(void)client->Write(initPkt.Buffer(), isz);
// Replay log history so the browser sees messages emitted before it connected
logHistoryMutex.FastLock();
{
uint32 startIdx = (logHistoryFill >= LOG_HISTORY_SIZE)
? logHistoryWriteIdx : 0u;
uint32 count = logHistoryFill;
for (uint32 i = 0u; i < count; i++) {
uint32 idx = (startIdx + i) % LOG_HISTORY_SIZE;
if (logHistory[idx][0] != '\0') {
StreamString histPkt;
histPkt += "event: message\ndata: ";
histPkt += logHistory[idx];
histPkt += "\n\n";
uint32 hsz = histPkt.Size();
(void)client->Write(histPkt.Buffer(), hsz);
}
}
}
logHistoryMutex.FastUnLock();
// Register in the SSE client slot array
sseMutex.FastLock();
bool registered = false;
for (uint32 i = 0u; i < MAX_SSE_CLIENTS; i++) {
if (sseClients[i] == NULL_PTR(BasicTCPSocket *)) {
sseClients[i] = client;
registered = true;
break;
}
}
sseMutex.FastUnLock();
if (!registered) {
client->Close();
delete client;
}
}
// ---------------------------------------------------------------------------
// HTTP request parsing
// ---------------------------------------------------------------------------
bool WebDebugService::ParseRequest(BasicTCPSocket *client,
StreamString &method, StreamString &path,
StreamString &body) {
static const uint32 MAX_HEADER = 8192u;
char8 buf[MAX_HEADER + 1];
uint32 total = 0u;
uint32 headerEnd = 0u;
bool foundEnd = false;
while (total < MAX_HEADER && !foundEnd) {
uint32 sz = MAX_HEADER - total;
if (!client->Read(buf + total, sz) || sz == 0u) break;
total += sz;
buf[total] = '\0';
for (uint32 i = 0u; i + 3u < total; i++) {
if (buf[i] == '\r' && buf[i+1] == '\n' &&
buf[i+2] == '\r' && buf[i+3] == '\n') {
headerEnd = i + 4u;
foundEnd = true;
break;
}
}
}
if (!foundEnd) return false;
const char8 *line = buf;
const char8 *sp1 = StringHelper::SearchChar(line, ' ');
if (sp1 == NULL_PTR(const char8 *)) return false;
method = "";
{ uint32 mLen = (uint32)(sp1 - line); (void)method.Write(line, mLen); }
const char8 *sp2 = StringHelper::SearchChar(sp1 + 1, ' ');
if (sp2 == NULL_PTR(const char8 *)) return false;
path = "";
{ uint32 pLen = (uint32)(sp2 - sp1 - 1); (void)path.Write(sp1 + 1, pLen); }
uint32 contentLength = 0u;
const char8 *cl = StringHelper::SearchString(buf, "Content-Length:");
if (cl != NULL_PTR(const char8 *)) {
cl += 15u;
while (*cl == ' ') cl++;
AnyType cv(UnsignedInteger32Bit, 0u, &contentLength);
AnyType cs(CharString, 0u, cl);
(void)TypeConvert(cv, cs);
}
body = "";
if (contentLength > 0u) {
uint32 alreadyRead = (total > headerEnd) ? (total - headerEnd) : 0u;
if (alreadyRead > 0u) (void)body.Write(buf + headerEnd, alreadyRead);
static const uint32 MAX_BODY = 16384u;
if (contentLength > alreadyRead && contentLength <= MAX_BODY) {
char8 bodyBuf[MAX_BODY];
uint32 toRead = contentLength - alreadyRead;
uint32 sz = toRead;
if (client->Read(bodyBuf, sz) && sz > 0u) {
(void)body.Write(bodyBuf, sz);
}
}
}
return true;
}
void WebDebugService::SendHttp(BasicTCPSocket *client, uint32 code,
const char8 *ct, StreamString &body) {
StreamString resp;
resp.Printf("HTTP/1.1 %u %s\r\n"
"Content-Type: %s\r\n"
"Content-Length: %u\r\n"
"Cache-Control: no-cache\r\n"
"Access-Control-Allow-Origin: *\r\n"
"Connection: close\r\n"
"\r\n",
code, (code == 200u) ? "OK" : "Error",
ct, (uint32)body.Size());
resp += body;
uint32 sz = resp.Size();
(void)client->Write(resp.Buffer(), sz);
}
void WebDebugService::HandleRequest(BasicTCPSocket *client,
const StreamString &method,
const StreamString &path,
const StreamString &body) {
if (method == "GET" && (path == "/" || path == "/index.html")) {
StreamString html = WEB_UI_HTML;
SendHttp(client, 200u, "text/html; charset=utf-8", html);
} else if (method == "GET" && path == "/api/events") {
UpgradeToSse(client);
return; // socket ownership transferred
} else if (method == "POST" && path == "/api/command") {
StreamString out;
HandleCommand(body, out);
SendHttp(client, 200u, "text/plain; charset=utf-8", out);
} else if (method == "OPTIONS") {
const char8 *hdr =
"HTTP/1.1 204 No Content\r\n"
"Access-Control-Allow-Origin: *\r\n"
"Access-Control-Allow-Methods: GET, POST, OPTIONS\r\n"
"Access-Control-Allow-Headers: Content-Type\r\n"
"Connection: close\r\n"
"\r\n";
uint32 sz = StringHelper::Length(hdr);
(void)client->Write(hdr, sz);
} else {
StreamString body404 = "Not Found";
SendHttp(client, 404u, "text/plain", body404);
}
client->Close();
delete client;
}
// ---------------------------------------------------------------------------
// HTTP server thread
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType WebDebugService::HttpServer(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::TerminationStage)
return ErrorManagement::NoError;
if (info.GetStage() == ExecutionInfo::StartupStage) {
Sleep::MSec(500u); // wait for ORD to finish
return ErrorManagement::NoError;
}
BasicTCPSocket *client = tcpServer.WaitConnection(TimeoutType(100u));
if (client == NULL_PTR(BasicTCPSocket *))
return ErrorManagement::NoError;
StreamString method, path, body;
if (ParseRequest(client, method, path, body)) {
HandleRequest(client, method, path, body);
// For SSE, HandleRequest transferred ownership — don't touch client here
} else {
client->Close();
delete client;
}
return ErrorManagement::NoError;
}
// ---------------------------------------------------------------------------
// Streamer thread — drains log, polls monitors, drains trace buffer, heartbeat
// ---------------------------------------------------------------------------
ErrorManagement::ErrorType WebDebugService::Streamer(ExecutionInfo &info) {
if (info.GetStage() == ExecutionInfo::TerminationStage)
return ErrorManagement::NoError;
if (info.GetStage() == ExecutionInfo::StartupStage)
return ErrorManagement::NoError;
uint64 nowNs = (uint64)((float64)HighResolutionTimer::Counter() *
HighResolutionTimer::Period() * 1.0e9);
uint64 nowMs = (nowNs >= streamerT0Ns) ? (nowNs - streamerT0Ns) / 1000000u : 0u;
// -----------------------------------------------------------------
// Drain MARTe2 log pages → SSE log events
// -----------------------------------------------------------------
{
static const uint32 MAX_LOGS_PER_CYCLE = 32u;
uint32 logCount = 0u;
LoggerPage *logPage = Logger::Instance()->GetLogEntry();
while (logPage != NULL_PTR(LoggerPage *) && logCount < MAX_LOGS_PER_CYCLE) {
StreamString levelStr;
ErrorManagement::ErrorCodeToStream(
logPage->errorInfo.header.errorType, levelStr);
StreamString msgEsc;
WDS_EscapeJson(logPage->errorStrBuffer, msgEsc);
StreamString ev;
ev.Printf("{\"type\":\"log\",\"level\":\"%s\",\"msg\":\"%s\"}",
levelStr.Buffer(), msgEsc.Buffer());
Logger::Instance()->ReturnPage(logPage);
logHistoryMutex.FastLock();
{
uint32 evSz = ev.Size();
if (evSz >= LOG_ENTRY_MAX_SIZE) evSz = LOG_ENTRY_MAX_SIZE - 1u;
(void)StringHelper::CopyN(logHistory[logHistoryWriteIdx],
ev.Buffer(), evSz);
logHistory[logHistoryWriteIdx][evSz] = '\0';
logHistoryWriteIdx = (logHistoryWriteIdx + 1u) % LOG_HISTORY_SIZE;
if (logHistoryFill < LOG_HISTORY_SIZE) logHistoryFill++;
}
logHistoryMutex.FastUnLock();
BroadcastSse("message", ev);
logPage = Logger::Instance()->GetLogEntry();
logCount++;
}
}
// -----------------------------------------------------------------
// Poll monitored signals → SSE monitor events
// -----------------------------------------------------------------
mutex.FastLock();
for (uint32 i = 0u; i < monitoredSignals.GetNumberOfElements(); i++) {
if (nowMs >= (monitoredSignals[i].lastPollTime +
monitoredSignals[i].periodMs)) {
monitoredSignals[i].lastPollTime = nowMs;
uint32 tsMs = (uint32)nowMs;
void *addr = NULL_PTR(void *);
if (monitoredSignals[i].dataSource->GetSignalMemoryBuffer(
monitoredSignals[i].signalIdx, 0u, addr) &&
addr != NULL_PTR(void *)) {
uint32 sz = monitoredSignals[i].size;
if (sz > 1024u) sz = 1024u;
uint8 lb[1024]; memcpy(lb, addr, sz);
TypeDescriptor td = monitoredSignals[i].dataSource->GetSignalType(
monitoredSignals[i].signalIdx);
float64 val = WDS_ToFloat64(lb, td);
StreamString sigName;
for (uint32 j = 0u; j < aliases.GetNumberOfElements(); j++) {
if (signals[aliases[j].signalIndex]->internalID ==
monitoredSignals[i].internalID) {
sigName = aliases[j].name;
break;
}
}
if (sigName.Size() == 0u) sigName = monitoredSignals[i].path;
char8 valBuf[64] = { '\0' };
{
AnyType vdst(CharString, 0u, valBuf);
vdst.SetNumberOfElements(0u, 63u);
AnyType vsrc(Float64Bit, 0u, &val);
(void)TypeConvert(vdst, vsrc);
}
StreamString ev;
ev.Printf(
"{\"type\":\"monitor\",\"name\":\"%s\",\"ts\":%u,\"value\":%s}",
sigName.Buffer(), tsMs, valBuf);
mutex.FastUnLock();
BroadcastSse("message", ev);
mutex.FastLock();
}
}
}
mutex.FastUnLock();
// -----------------------------------------------------------------
// Drain ring buffer → SSE trace events
// -----------------------------------------------------------------
static const uint32 MAX_TRACE_CYCLE = 50u;
uint32 id, size;
uint64 sampleTs;
bool hasData = false;
uint32 traceCount = 0u;
while (traceCount < MAX_TRACE_CYCLE &&
traceBuffer.Pop(id, sampleTs, traceSampleBuffer, size, TRACE_SAMPLE_MAX)) {
hasData = true;
traceCount++;
StreamString sigName;
TypeDescriptor sigType = UnsignedInteger32Bit;
mutex.FastLock();
for (uint32 i = 0u; i < signals.GetNumberOfElements(); i++) {
if (signals[i]->internalID == id) {
sigName = signals[i]->name;
sigType = signals[i]->type;
break;
}
}
mutex.FastUnLock();
uint32 tsMs = (sampleTs >= streamerT0Ns)
? (uint32)((sampleTs - streamerT0Ns) / 1000000u) : 0u;
float64 val = WDS_ToFloat64(traceSampleBuffer, sigType);
char8 valBuf[64] = { '\0' };
{
AnyType vdst(CharString, 0u, valBuf); vdst.SetNumberOfElements(0u, 63u);
AnyType vsrc(Float64Bit, 0u, &val);
(void)TypeConvert(vdst, vsrc);
}
StreamString ev;
ev.Printf("{\"type\":\"trace\",\"name\":\"%s\",\"ts\":%u,\"value\":%s}",
sigName.Buffer(), tsMs, valBuf);
BroadcastSse("message", ev);
}
// -----------------------------------------------------------------
// Periodic status heartbeat (every 500 ms)
// -----------------------------------------------------------------
static uint64 lastStatusMs = 0u;
if (nowMs - lastStatusMs >= 500u) {
lastStatusMs = nowMs;
mutex.FastLock();
bool paused = isPaused;
StreamString gam = pausedAtGam;
uint32 rem = stepRemaining;
mutex.FastUnLock();
StreamString ev;
ev.Printf(
"{\"type\":\"status\",\"paused\":%s,\"gam\":\"%s\",\"remaining\":%u}",
paused ? "true" : "false", gam.Buffer(), rem);
BroadcastSse("message", ev);
}
RemoveDeadSseClients();
(void)hasData;
Sleep::MSec(10u);
return ErrorManagement::NoError;
}
} // namespace MARTe
Reference in New Issue View Git Blame Copy Permalink