Imported CSS files

epics/css/sys-mng-opi/CSS/MARTe/Interfaces/TCPMessageHandler/TCPMessageHandler.cpp

@@ -0,0 +1,216 @@
+/*
+ * Copyright 2011 EFDA | European Fusion Development Agreement
+ *
+ * Licensed under the EUPL, Version 1.1 or - as soon they 
+   will be approved by the European Commission - subsequent  
+   versions of the EUPL (the "Licence"); 
+ * You may not use this work except in compliance with the 
+   Licence. 
+ * You may obtain a copy of the Licence at: 
+ *  
+ * http://ec.europa.eu/idabc/eupl
+ *
+ * Unless required by applicable law or agreed to in 
+   writing, software distributed under the Licence is 
+   distributed on an "AS IS" basis, 
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either 
+   express or implied. 
+ * See the Licence for the specific language governing 
+   permissions and limitations under the Licence. 
+ *
+ * $Id: MessageTriggeringTimeService.h 3 2012-01-15 16:26:07Z aneto $
+ *
+**/
+
+#include "TCPMessageHandler.h"
+#include "MessageDispatcher.h"
+
+void ConnectionHandlerFn(TCPMessageHandler &tcpmh){
+    while(tcpmh.keepAlive){
+        tcpmh.ConnectionHandler();
+        CStaticAssertErrorCondition(FatalError, "Lost server connection");
+        if(tcpmh.keepAlive){
+            CStaticAssertErrorCondition(FatalError, "Retrying in 10 seconds");
+            SleepSec(10.0);
+        }
+    }
+    //Just to signal that we have shutdown
+    tcpmh.keepAlive = True;
+}
+
+TCPMessageHandler::TCPMessageHandler(){
+    serverPort = -1;
+    serverTID  = 0;
+    cpuMask    = 0;
+    keepAlive  = False;
+    msgTimeout = TTInfiniteWait;
+}
+
+TCPMessageHandler::~TCPMessageHandler(){
+    keepAlive = False;
+    //Open a connection to the server to force the shutdown
+    FString   host = "localhost";
+    TCPSocket client;
+    //Open the socket
+    if(!client.Open()){
+        AssertErrorCondition(FatalError, "%s: failed to shutdown server. Waited for 1 second.", Name());
+        return;
+    }
+    //Connect to the server
+    if(!client.Connect(host.Buffer(), serverPort)){
+        CStaticAssertErrorCondition(FatalError, "%s: Failed to connect to %s:%d", Name(), host.Buffer(), serverPort);
+        client.Close();
+    }
+    //Write a line
+    FString line = "";
+    uint32  size = line.Size();
+    if(!client.Write(line.Buffer(), size)){
+        CStaticAssertErrorCondition(FatalError, "Failed to write to socket");
+    }
+    //Housekeeping
+    client.Close();
+    int32 exitCounter = 0;
+    while(!keepAlive){
+        exitCounter++;
+        SleepMsec(10);
+        if(exitCounter > 100){
+            AssertErrorCondition(FatalError, "%s: failed to shutdown server. Waited for 1 second.", Name());
+            break;
+        }
+    }
+    if(exitCounter > 100){
+        Threads::Kill(serverTID);
+    }
+    serverTID = 0;
+
+}
+
+bool TCPMessageHandler::ConnectionHandler(){
+    //Open the server connection
+    if(!server.Open()){
+        AssertErrorCondition(FatalError, "%s: Failed to open the server socket", Name());
+        return False;
+    }
+    //Set in server mode
+    if(!server.Listen(serverPort)){
+        CStaticAssertErrorCondition(FatalError, "Failed to create server running in port %d", serverPort);
+        server.Close();
+        return False;
+    }
+    //Wait for a connection
+    TCPSocket *client = NULL;
+
+    while(keepAlive){
+        client = server.WaitConnection();
+        if(client == NULL){
+            AssertErrorCondition(FatalError, "%s: Failed waiting for a connection in port %d", serverPort, Name());
+            server.Close();
+            return False;
+        }
+        //Set the client in blocking mode for the read
+        client->SetBlocking(True);
+        //Print information from the client
+        FString hostname;
+        client->Source().HostName(hostname);
+        AssertErrorCondition(Information, "%s: Accepted a connection from %s", Name(), hostname.Buffer());
+        ClientHandler(client);
+    }
+
+    server.Close();
+    return True;
+}
+
+void TCPMessageHandler::ClientHandler(TCPSocket *client){
+    //Read a line from the client socket
+    FString line;
+    bool ret = False;
+
+    line.SetSize(0);
+    while(client->GetLine(line)){
+        ret = HandleRequest(line, client);
+        line.SetSize(0);
+	if(!ret){
+		client->Printf("Failed\n");
+	}
+    }
+}
+
+bool TCPMessageHandler::HandleRequest(FString &req, TCPSocket *client){
+    FString destination;
+    FString code;
+    FString content;
+    req.Seek(0);
+    printf("%s\n", req.Buffer());
+    if(!req.GetToken(destination, "|")){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Could not read the message destination", Name());
+        return False;
+    }
+    if(!req.GetToken(code, "|")){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Could not read the message code", Name());
+        return False;
+    }
+    if(!req.GetToken(content, "|")){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Could not read the message content", Name());
+        return False;
+    }
+
+    AssertErrorCondition(Information, "%s: HandleRequest : D=%s,C=%s,CT=%s", Name(), destination.Buffer(), code.Buffer(), content.Buffer());
+    GCRTemplate<Message> msg(GCFT_Create);
+    if(!msg.IsValid()){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Failed to create message", Name());
+        return False;
+    }
+    GCRTemplate<MessageEnvelope> env(GCFT_Create);
+    if (!env.IsValid()){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Failed to creating envelope", Name());
+        return False;
+    }
+    msg->Init(atoi(code.Buffer()), content.Buffer());
+    env->PrepareMessageEnvelope(msg, destination.Buffer(), MDRF_ManualReply, this);
+
+    GCRTemplate<MessageEnvelope>   reply;
+    MessageHandler::SendMessageAndWait(env, reply, msgTimeout);
+    if(!reply.IsValid()){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Received an invalid reply", Name());
+        return False;
+    }
+    GCRTemplate<Message> replyMessage = reply->GetMessage(); 
+    if(!replyMessage.IsValid()){
+        AssertErrorCondition(FatalError, "%s: HandleRequest: Received an invalid reply message", Name());
+        return False;
+    }
+    AssertErrorCondition(Information, "%s: HandleRequest: Received a reply with code=%d and content=%s", Name(), replyMessage->GetMessageCode().Code(), replyMessage->Content());
+
+    client->Printf("Success with code=%d and content=%s\n", replyMessage->GetMessageCode().Code(), replyMessage->Content());
+
+    return True;
+}
+
+bool TCPMessageHandler::ObjectLoadSetup(ConfigurationDataBase &cdb, StreamInterface *err){
+    if(!GCNamedObject::ObjectLoadSetup(cdb, err)){
+        AssertErrorCondition(FatalError, "%s::ObjectLoadSetup: ObjectLoadSetup of GCNamedObject failed", Name());
+        return False;
+    }
+    
+    CDBExtended cdbe(cdb);
+    if(!cdbe.ReadInt32(serverPort, "ServerPort")){
+        AssertErrorCondition(FatalError, "%s::ObjectLoadSetup: PulseNumberMessageCode is compulsory when PulseNumberMessageDestinations are set", Name());
+    }
+    
+    //CPU mask for the thread
+    if(!cdbe.ReadInt32(cpuMask, "CPUMask", 0x1)){
+        AssertErrorCondition(Warning, "%s::ObjectLoadSetup: CPUMask was not specified. Using default: %d", Name(), cpuMask);
+    }
+
+    //The msg send timeout, default is infinite
+    int32 msgTimeoutMS = 0;
+    if(cdbe.ReadInt32(msgTimeoutMS, "MSGTimeout")){
+        msgTimeout = msgTimeoutMS;
+    }
+    keepAlive = True;
+    serverTID = Threads::BeginThread((void (__thread_decl *)(void *))&ConnectionHandlerFn,this, THREADS_DEFAULT_STACKSIZE, Name(), XH_NotHandled, cpuMask);
+    return True;
+}
+
+OBJECTLOADREGISTER(TCPMessageHandler, "$Id: TCPMessageHandler.cpp,v 1.2 2011/12/07 13:55:43 aneto Exp $")
+