#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.Size(); 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.Size(); 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 SAMPLE_BUF_SIZE = 1024u; static const uint32 MAX_TRACE_CYCLE = 50u; uint32 id, size; uint64 sampleTs; uint8 sampleData[SAMPLE_BUF_SIZE]; bool hasData = false; uint32 traceCount = 0u; while (traceCount < MAX_TRACE_CYCLE && traceBuffer.Pop(id, sampleTs, sampleData, size, SAMPLE_BUF_SIZE)) { hasData = true; traceCount++; if (size > SAMPLE_BUF_SIZE) continue; StreamString sigName; TypeDescriptor sigType = UnsignedInteger32Bit; mutex.FastLock(); for (uint32 i = 0u; i < signals.Size(); 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(sampleData, 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