Added custom Message functionality
This commit is contained in:
Martino Ferrari
@@ -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<ReferenceContainer *>(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<Message> &data) {
|
||||
printf("<debug> 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<Message> 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("<debug> MSG: Destination object %s not found in ORD\n", dest.Buffer());
|
||||
}
|
||||
|
||||
if (err != ErrorManagement::NoError) {
|
||||
printf("<debug> MSG: MessageI dispatch failed.\n");
|
||||
}
|
||||
} else {
|
||||
printf("<debug> 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",
|
||||
|
||||
Reference in New Issue
Block a user