diff --git a/Source/Components/Interfaces/DebugService/DebugService.cpp b/Source/Components/Interfaces/DebugService/DebugService.cpp index 1fefebc..27af8b6 100644 --- a/Source/Components/Interfaces/DebugService/DebugService.cpp +++ b/Source/Components/Interfaces/DebugService/DebugService.cpp @@ -7,6 +7,7 @@ #include "GAM.h" #include "GlobalObjectsDatabase.h" #include "HighResolutionTimer.h" +#include "Message.h" #include "ObjectBuilder.h" #include "ObjectRegistryDatabase.h" #include "StreamString.h" @@ -110,7 +111,6 @@ DebugService::~DebugService() { for (uint32 i = 0; i < signals.Size(); i++) { delete signals[i]; } - this->Purge(); } bool DebugService::Initialise(StructuredDataI &data) { @@ -189,34 +189,6 @@ bool DebugService::Initialise(StructuredDataI &data) { return false; if (streamerService.Start() != ErrorManagement::NoError) return false; - - if (logPort > 0) { - Reference tcpLogger( - "TcpLogger", GlobalObjectsDatabase::Instance()->GetStandardHeap()); - if (tcpLogger.IsValid()) { - ConfigurationDatabase loggerConfig; - loggerConfig.Write("Port", (uint32)logPort); - if (tcpLogger->Initialise(loggerConfig)) { - this->Insert(tcpLogger); - - Reference loggerService( - "LoggerService", - GlobalObjectsDatabase::Instance()->GetStandardHeap()); - if (loggerService.IsValid()) { - ConfigurationDatabase serviceConfig; - serviceConfig.Write("CPUs", (uint32)1); - ReferenceContainer *lc = - dynamic_cast(loggerService.operator->()); - if (lc != NULL_PTR(ReferenceContainer *)) { - lc->Insert(tcpLogger); - } - if (loggerService->Initialise(serviceConfig)) { - this->Insert(loggerService); - } - } - } - } - } } return true; } @@ -397,6 +369,11 @@ ErrorManagement::ErrorType DebugService::Execute(ExecutionInfo &info) { return ErrorManagement::FatalError; } +ErrorManagement::ErrorType DebugService::HandleMessage(ReferenceT &data) { + printf(" DebugService received custom message: Function=%s\n", (const char8*)data->GetFunction()); + return ErrorManagement::NoError; +} + ErrorManagement::ErrorType DebugService::Server(ExecutionInfo &info) { if (info.GetStage() == ExecutionInfo::TerminationStage) return ErrorManagement::NoError; @@ -600,7 +577,135 @@ void DebugService::HandleCommand(StreamString cmd, BasicTCPSocket *client) { } } else if (token == "DISCOVER") Discover(client); - else if (token == "SERVICE_INFO") { + else if (token == "MSG") { + StreamString dest, func, waitStr; + if (cmd.GetToken(dest, delims, term) && + cmd.GetToken(func, delims, term) && + cmd.GetToken(waitStr, delims, term)) { + bool wait = (waitStr == "1"); + + const char8 *pStart = cmd.Buffer() + cmd.Position(); + StreamString rawPayload = pStart; + + // Decode escaped newlines (\n) + StreamString payload; + rawPayload.Seek(0u); + char8 c; + while (rawPayload.Size() > rawPayload.Position()) { + uint32 readS = 1; + if (rawPayload.Read(&c, readS)) { + if (c == '\\') { + char8 next; + if (rawPayload.Read(&next, readS)) { + if (next == 'n') { + payload += '\n'; + } else { + payload += c; + payload += next; + } + } else { + payload += c; + } + } else { + payload += c; + } + } + } + + ReferenceT msg( + "Message", GlobalObjectsDatabase::Instance()->GetStandardHeap()); + + ConfigurationDatabase msgConfig; + msgConfig.Write("Destination", dest.Buffer()); + msgConfig.Write("Function", func.Buffer()); + if (wait) { + msgConfig.Write("Mode", "ExpectsReply"); + } + + if (payload.Size() > 0u) { + payload.Seek(0u); + StreamString line; + while (payload.GetToken(line, "\n", term)) { + if (line.Size() > 0u) { + const char8 *eq = StringHelper::SearchChar(line.Buffer(), '='); + if (eq != NULL_PTR(const char8 *)) { + StreamString key, val; + uint32 eqPos = (uint32)(eq - line.Buffer()); + (void)line.Seek(0u); + + char8* keyBuf = new char8[eqPos + 1]; + uint32 keyReadSize = eqPos; + if (line.Read(keyBuf, keyReadSize)) { + keyBuf[eqPos] = '\0'; + key = keyBuf; + } + delete[] keyBuf; + + (void)line.Seek(eqPos + 1u); + uint32 valLen = line.Size() - eqPos - 1u; + char8* valBuf = new char8[valLen + 1]; + uint32 valReadSize = valLen; + if (line.Read(valBuf, valReadSize)) { + valBuf[valLen] = '\0'; + val = valBuf; + } + delete[] valBuf; + + if (key.Size() > 0u) { + if (msgConfig.CreateRelative("Payload")) { + (void)msgConfig.Write(key.Buffer(), val.Buffer()); + (void)msgConfig.MoveToAncestor(1u); + } + } + + } + } + line = ""; + } + } + + ErrorManagement::ErrorType err = ErrorManagement::ParametersError; + if (msg->Initialise(msgConfig)) { + // Find destination object in the global database + Reference destObj = ObjectRegistryDatabase::Instance()->Find(dest.Buffer()); + if (destObj.IsValid()) { + Object* sender = this; + // Double check if we are in the registry to be a valid sender + StreamString myPath; + if (!GetFullObjectName(*this, myPath)) { + sender = NULL_PTR(Object*); + } + + if (wait) { + err = MessageI::WaitForReply(msg, TTInfiniteWait); + } else { + err = MessageI::SendMessage(msg, sender); + } + } else { + + printf(" MSG: Destination object %s not found in ORD\n", dest.Buffer()); + } + + if (err != ErrorManagement::NoError) { + printf(" MSG: MessageI dispatch failed.\n"); + } + } else { + printf(" MSG: Message initialization failed\n"); + } + + if (client) { + + if (err == ErrorManagement::NoError) { + uint32 okSize = 7; + (void)client->Write("OK MSG\n", okSize); + } else { + uint32 errSize = 10; + (void)client->Write("ERROR MSG\n", errSize); + } + } + } + } + else if (token == "SERVICE_INFO") { if (client) { StreamString resp; resp.Printf("OK SERVICE_INFO TCP_CTRL:%u UDP_STREAM:%u TCP_LOG:%u STATE:%s\n", diff --git a/Source/Components/Interfaces/DebugService/DebugService.h b/Source/Components/Interfaces/DebugService/DebugService.h index faabddb..5cc6172 100644 --- a/Source/Components/Interfaces/DebugService/DebugService.h +++ b/Source/Components/Interfaces/DebugService/DebugService.h @@ -58,6 +58,8 @@ public: virtual ErrorManagement::ErrorType Execute(ExecutionInfo &info); + virtual ErrorManagement::ErrorType HandleMessage(ReferenceT &data); + bool IsPaused() const { return isPaused; } void SetPaused(bool paused) { isPaused = paused; } diff --git a/Test/Configurations/debug_test.cfg b/Test/Configurations/debug_test.cfg index f7ba9f8..139d4bc 100644 --- a/Test/Configurations/debug_test.cfg +++ b/Test/Configurations/debug_test.cfg @@ -164,3 +164,12 @@ LogPort = 8082 StreamIP = "127.0.0.1" } + ++LoggerService = { + Class = LoggerService + CPUs = 0x1 + +DebugConsumer = { + Class = TcpLogger + Port = 8082 + } +} diff --git a/Test/Integration/IntegrationTests.cpp b/Test/Integration/IntegrationTests.cpp index 7da9d4c..930e86f 100644 --- a/Test/Integration/IntegrationTests.cpp +++ b/Test/Integration/IntegrationTests.cpp @@ -86,6 +86,10 @@ int main() { printf("\n--- Test 6: TREE Command Enhancement ---\n"); TestTreeCommand(); Sleep::MSec(1000); + + printf("\n--- Test 7: Custom MARTe Message (MSG) ---\n"); + TestMessageCommand(); + Sleep::MSec(1000); printf("\nAll Integration Tests Finished.\n"); diff --git a/Test/Integration/Makefile.inc b/Test/Integration/Makefile.inc index bfa0918..32bf5da 100644 --- a/Test/Integration/Makefile.inc +++ b/Test/Integration/Makefile.inc @@ -1,4 +1,4 @@ -OBJSX = SchedulerTest.x TraceTest.x ValidationTest.x ConfigCommandTest.x TreeCommandTest.x TestCommon.x +OBJSX = SchedulerTest.x TraceTest.x ValidationTest.x ConfigCommandTest.x TreeCommandTest.x MessageCommandTest.x TestCommon.x PACKAGE = Test/Integration diff --git a/Test/Integration/MessageCommandTest.cpp b/Test/Integration/MessageCommandTest.cpp new file mode 100644 index 0000000..8d5ba76 --- /dev/null +++ b/Test/Integration/MessageCommandTest.cpp @@ -0,0 +1,72 @@ +#include "TestCommon.h" +#include "ObjectRegistryDatabase.h" +#include "DebugService.h" +#include "StandardParser.h" +#include "GlobalObjectsDatabase.h" +#include + +using namespace MARTe; + +namespace MARTe { + +void TestMessageCommand() { + printf("--- Test: Custom MARTe Message (MSG) ---\n"); + + ObjectRegistryDatabase::Instance()->Purge(); + Sleep::MSec(1000); + + ConfigurationDatabase cdb; + const char8 * const msg_test_config = + "DebugService = {" + " Class = DebugService " + " ControlPort = 8120 " + " UdpPort = 8121 " + " StreamIP = \"127.0.0.1\" " + "}"; + + StreamString ss = msg_test_config; + ss.Seek(0); + StandardParser parser(ss, cdb); + if (!parser.Parse()) { + printf("ERROR: Failed to parse config\n"); + return; + } + + // Initialize Service + ReferenceT service("DebugService", GlobalObjectsDatabase::Instance()->GetStandardHeap()); + service->SetName("DebugService"); + + cdb.MoveToRoot(); + if (cdb.MoveRelative("DebugService")) { + if (!service->Initialise(cdb)) { + printf("ERROR: Failed to initialize DebugService\n"); + return; + } + } + ObjectRegistryDatabase::Instance()->Insert(service); + + if (ObjectRegistryDatabase::Instance()->Find("DebugService").IsValid()) { + printf("DebugService successfully registered in ORD.\n"); + } else { + printf("ERROR: DebugService NOT found in ORD.\n"); + } + + printf("Service initialized on port 8120.\n"); + Sleep::MSec(500); + + StreamString reply; + if (SendCommandGAM(8120, "MSG DebugService UnknownFunc 0 Key1=Val1\\nKey2=Val2\n", reply)) { + printf("MSG response received: %s", reply.Buffer()); + if (StringHelper::SearchString(reply.Buffer(), "OK MSG") != NULL_PTR(const char8 *)) { + printf("SUCCESS: Asynchronous message dispatched correctly.\n"); + } else { + printf("FAILURE: MSG command returned error.\n"); + } + } else { + printf("ERROR: MSG command communication failed\n"); + } + + ObjectRegistryDatabase::Instance()->Purge(); +} + +} // namespace MARTe diff --git a/Test/Integration/TestCommon.h b/Test/Integration/TestCommon.h index 6c86ab7..0077bfa 100644 --- a/Test/Integration/TestCommon.h +++ b/Test/Integration/TestCommon.h @@ -7,6 +7,7 @@ namespace MARTe { extern const char8 * const debug_test_config; bool SendCommandGAM(uint16 port, const char8* cmd, StreamString &reply); + void TestMessageCommand(); } #endif diff --git a/Test/UnitTests/UnitTests.cpp b/Test/UnitTests/UnitTests.cpp index d7db67d..280d4ec 100644 --- a/Test/UnitTests/UnitTests.cpp +++ b/Test/UnitTests/UnitTests.cpp @@ -55,6 +55,7 @@ public: service.HandleCommand("RESUME", NULL_PTR(BasicTCPSocket*)); service.HandleCommand("LS /", NULL_PTR(BasicTCPSocket*)); service.HandleCommand("INFO X.Y.Z", NULL_PTR(BasicTCPSocket*)); + service.HandleCommand("MSG DebugService DummyFunc 0 K=V", NULL_PTR(BasicTCPSocket*)); // 3. Broker Active Status volatile bool active = false; diff --git a/Tools/gui_client/src/main.rs b/Tools/gui_client/src/main.rs index b7cac63..18fb21c 100644 --- a/Tools/gui_client/src/main.rs +++ b/Tools/gui_client/src/main.rs @@ -186,6 +186,13 @@ struct MonitorDialog { period_ms: String, } +struct MessageDialog { + destination: String, + function: String, + payload: String, + expect_reply: bool, +} + struct LogFilters { show_debug: bool, show_info: bool, @@ -232,6 +239,7 @@ struct MarteDebugApp { telem_match_count: HashMap, forcing_dialog: Option, monitoring_dialog: Option, + message_dialog: Option, style_editor: Option<(usize, usize)>, tx_cmd: Sender, rx_events: Receiver, @@ -312,6 +320,7 @@ impl MarteDebugApp { telem_match_count: HashMap::new(), forcing_dialog: None, monitoring_dialog: None, + message_dialog: None, style_editor: None, tx_cmd, rx_events, @@ -400,6 +409,34 @@ impl MarteDebugApp { } } + fn get_all_objects(&self) -> Vec { + let mut objects = Vec::new(); + if let Some(tree) = &self.app_tree { + fn collect(item: &TreeItem, path: String, objects: &mut Vec) { + let current_path = if path.is_empty() { + if item.name == "Root" { + "".to_string() + } else { + item.name.clone() + } + } else { + format!("{}.{}", path, item.name) + }; + if !current_path.is_empty() && !item.class.contains("Signal") { + objects.push(current_path.clone()); + } + if let Some(children) = &item.children { + for child in children { + collect(child, current_path.clone(), objects); + } + } + } + collect(tree, "".to_string(), &mut objects); + } + objects.sort(); + objects + } + fn render_tree(&mut self, ui: &mut egui::Ui, item: &TreeItem, path: String) { let current_path = if path.is_empty() { if item.name == "Root" { @@ -1202,6 +1239,65 @@ impl eframe::App for MarteDebugApp { } } + if self.message_dialog.is_some() { + let mut dialog = self.message_dialog.take().unwrap(); + let mut close = false; + let objects = self.get_all_objects(); + egui::Window::new("Send MARTe Message").show(ctx, |ui| { + egui::Grid::new("msg_grid").num_columns(2).show(ui, |ui| { + ui.label("Destination:"); + egui::ComboBox::from_id_salt("dest_combo") + .selected_text(&dialog.destination) + .width(200.0) + .show_ui(ui, |ui| { + for obj in objects { + ui.selectable_value(&mut dialog.destination, obj.clone(), obj); + } + }); + ui.end_row(); + + ui.label("Function:"); + ui.text_edit_singleline(&mut dialog.function); + ui.end_row(); + + ui.label("Payload:"); + ui.vertical(|ui| { + ui.text_edit_multiline(&mut dialog.payload); + ui.label(egui::RichText::new("Format: Key = Value (one per line)").small().weak()); + }); + ui.end_row(); + + ui.label("Wait Reply:"); + ui.checkbox(&mut dialog.expect_reply, ""); + ui.end_row(); + }); + + ui.horizontal(|ui| { + if ui.button("🚀 Send").clicked() { + let wait = if dialog.expect_reply { "1" } else { "0" }; + // Replace actual newlines with literal '\n' for the server-side tokenizer if needed, + // or ensure the server handles the raw multi-line stream if the protocol allows it. + // Given HandleCommand reads line-by-line, we must send it carefully. + // Actually, our Server loop reads up to \n. + // So we should encode newlines in payload if we want to send them in one go. + let encoded_payload = dialog.payload.replace('\n', "\\n"); + let cmd = format!( + "MSG {} {} {} {}", + dialog.destination, dialog.function, wait, encoded_payload + ); + let _ = self.tx_cmd.send(cmd); + close = true; + } + if ui.button("Cancel").clicked() { + close = true; + } + }); + }); + if !close { + self.message_dialog = Some(dialog); + } + } + if let Some((p_idx, s_idx)) = self.style_editor { let mut close = false; egui::Window::new("Signal Style").show(ctx, |ui| { @@ -1407,6 +1503,15 @@ impl eframe::App for MarteDebugApp { } }); ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| { + if ui.button("✉ Send Msg").clicked() { + self.message_dialog = Some(MessageDialog { + destination: "".to_string(), + function: "".to_string(), + payload: "".to_string(), + expect_reply: false, + }); + } + ui.separator(); ui.label(format!( "UDP: OK[{}] DROP[{}]", self.udp_packets, self.udp_dropped