Changeset 8864 for trunk/MagicSoft/Cosy/tcpip

Timestamp:

02/17/08 22:39:04 (17 years ago)

Author:

tbretz

Message:

*** empty log message ***

Location:

trunk/MagicSoft/Cosy/tcpip

Files:

• MTcpIpIO.cc (modified) (6 diffs)
• MTcpIpIO.h (modified) (2 diffs)

trunk/MagicSoft/Cosy/tcpip/MTcpIpIO.cc

@@ -2 +2 @@
 
 #include <unistd.h>    // usleep
+
+#include <errno.h>
 
 #include <TSocket.h>
@@ -153 +155 @@
 }
 
-void MTcpIpIO::ReadSocket(TSocket &rx)
-{
-    // Clear buffer!
-    char c;
-//    while (fRxSocket->RecvRaw(&c, 1)>0 && !HasStopFlag())
-    while (rx.RecvRaw(&c, 1)>0 && !IsThreadCanceled())
-        usleep(1);
-
+Bool_t MTcpIpIO::ReadSocket(TSocket &rx)
+{
     TString str;
-    //        while (!HasStopFlag())
+
     while (!IsThreadCanceled())
     {
@@ -175 +171 @@
         }
 
-        // Data received with zero length!
+        // Data received with zero length! (Connection lost)
         if (len==0)
-        {
-            // THIS MEANS CONNECTIION LOST!!!!
-            cout << "============> len==0 (CONNECTION LOST?)" << endl;
-            break; // This break is for TEST PURPOSE FIXME!!!
-            continue;
-        }
+            return kFALSE; // This break is for TEST PURPOSE FIXME!!!
 
         // Data received
@@ -202 +193 @@
         str = "";
     }
-}
-
-//void *MTcpIpIO::Thread()
+
+    return kTRUE;
+}
+
+Bool_t MTcpIpI::WaitForData(TSocket &sock)
+{
+    // No connection established?
+    if (!sock.IsValid())
+    {
+        gLog << warn << "TSocket invalid on port " << fPortRx << "." << endl;
+        return kFALSE;
+    }
+
+    fConnectionEstablished = kTRUE;
+
+    // Get connection on port fPortRx and redirected
+    while (!IsThreadCanceled())
+    {
+        // Check for pending data (every ms)
+        switch (sock.Select(TSocket::kRead, 1))
+        {
+        case kTRUE:  // Data pending... go on reading
+            if (!ReadSocket(sock))
+            {
+                gLog << warn << "WARNING - Connection lost to " << MTcpIpO::GetSocketAddress(sock) << endl;
+                return kFALSE;
+            }
+
+            // *FALLTHROUGH*
+        case kFALSE: // Time out, no data yet, go on waiting
+
+            // Go on waiting for new data
+            continue;
+
+        default:  // Error occurance
+            gLog << err << "TSocket::Select returned an error: " << strerror(errno) << endl;
+            return kFALSE;
+        }
+    }
+    return kTRUE; //???
+}
+
+void MTcpIpI::WaitForConnection(TServerSocket &server)
+{
+    gLog << all << " Wait for connection" << endl;
+
+    while (!IsThreadCanceled())
+    {
+        gLog << all << "Listening for new connection on port " << fPortRx << "..." << endl;
+
+        // Check for a connection request (reminder: we are in non-blocking mode)
+        TSocket *socket = 0;
+
+        while (!IsThreadCanceled())
+        {
+            socket = server.Accept();
+
+            // Accept returned an error
+            if (socket==0)
+            {
+                gLog << err << "Error: TServerSock::Accept on port " << fPortRx << ": " << strerror(errno) << endl;
+                // Since we don't know the type of the error we better shut down the socket
+                return;
+            }
+
+            // No connection request pending
+            if ((Long_t)socket<0)
+            {
+                MThread::Sleep(1000); // Wait a ms
+                continue;
+            }
+
+            // Connection established
+            break;
+        }
+
+        if ((Long_t)socket<=0)
+            return;
+
+        if (!WaitForData(*socket))
+            fConnectionEstablished = kFALSE;
+
+        delete socket;
+    }
+}
+
+Int_t MTcpIpI::Thread()
+{
+    gLog << inf << "- Starting server listening on port " << fPortRx << "..." << endl;
+
+    while (!IsThreadCanceled())
+    {
+        TServerSocket *server=new TServerSocket(fPortRx, kTRUE);
+        server->SetOption(kNoBlock, 1);
+
+        while (!IsThreadCanceled() && server->IsValid())
+            WaitForConnection(*server);
+
+        if (!server->IsValid())
+        {
+            gLog << err << "ServerSocket on port " << fPortRx << " invalid: ";
+            switch (server->GetErrorCode())
+            {
+            case  0: gLog << "No error." << endl; break;
+            case -1: gLog << "low level socket() call failed." << endl; break;
+            case -2: gLog << "low level bind() call failed." << endl; break;
+            case -3: gLog << "low level listen() call failed." << endl; break;
+            default: gLog << "Unknown." << endl; break;
+            }
+        }
+
+        delete server;
+
+        MThread::Sleep(5000000);
+    }
+
+    gLog << inf << "- Listening server stopped on port " << fPortRx << "." << endl;
+
+    return 0;
+}
+
+/*
 Int_t MTcpIpI::Thread()
 {
@@ -276 +386 @@
         gLog << all << "Connection established on port " << fPortRx << "." << endl;
 
-        fRxSocket->SetOption(kNoBlock, 1);
+
+        //fRxSocket->SetOption(kNoBlock, 1);
+
+        // Waqit for data
+        while (!IsThreadCanceled())
+        {
+            switch (fRxSocket->Select(kRead, 1))
+            {
+            case kTRUE:  // Data waiting to be read
+                break;
+            case kFALSE: // time out
+                usleep(10);
+                continue;
+            }
+
+            // ERROR
+            cout << "Error: TRxSocket::Select on port " << fPortRx << "." << endl;
+
+            delete fServSock;
+            delete fRxSocket;
+            fServSock = NULL;
+            fRxSocket = NULL;
+            break;
+        }¨
+
+        if (!fServSock)
+            continue;
+
+        if (IsThreadCanceled())
+        {
+            delete fServSock;
+            delete fRxSocket;
+            fServSock = NULL;
+            fRxSocket = NULL;
+            continue;
+        }
 
         // ------ IDENTICAL UP TO HERE ------
 
+        // Read and evaluate data
         ReadSocket(*fRxSocket);
 
@@ -292 +438 @@
     return 0;
 //    return NULL;
-}
+}*/

trunk/MagicSoft/Cosy/tcpip/MTcpIpIO.h

@@ -18 +18 @@
 {
 private:
-    Int_t fPortRx;
+    Int_t  fPortRx;
+    Bool_t fConnectionEstablished;
+
+    Bool_t WaitForData(TSocket &sock);
+    void   WaitForConnection(TServerSocket &server);
 
     Int_t Thread();
 
-    virtual void ReadSocket(TSocket &rx) = 0;
+    virtual Bool_t ReadSocket(TSocket &rx) = 0;
 
 public:
-    MTcpIpI(Int_t rx) : MThread(Form("MTcpIpI::%d", rx)), fPortRx(rx) { /*RunThread();*/ }
+    MTcpIpI(Int_t rx) : MThread(Form("MTcpIpI::%d", rx)), fPortRx(rx), fConnectionEstablished(kFALSE) { /*RunThread();*/ }
     ~MTcpIpI() { CancelThread(); }
+
+    Bool_t IsConnectionEstablished() const { return fConnectionEstablished; }
 };
 
@@ -52 +58 @@
 {
 private:
-    void ReadSocket(TSocket &rx);
+    Bool_t ReadSocket(TSocket &rx);
 
 public: