source: trunk/MagicSoft/Mars/mbase/MReadSocket.cc@ 2436

/* ======================================================================== *\
!
! *
! * This file is part of MARS, the MAGIC Analysis and Reconstruction
! * Software. It is distributed to you in the hope that it can be a useful
! * and timesaving tool in analysing Data of imaging Cerenkov telescopes.
! * It is distributed WITHOUT ANY WARRANTY.
! *
! * Permission to use, copy, modify and distribute this software and its
! * documentation for any purpose is hereby granted without fee,
! * provided that the above copyright notice appear in all copies and
! * that both that copyright notice and this permission notice appear
! * in supporting documentation. It is provided "as is" without express
! * or implied warranty.
! *
!
!
!   Author(s): Thomas Bretz  12/2000 <mailto:tbretz@uni-sw.gwdg.de>
!
!   Copyright: MAGIC Software Development, 2000-2001
!
!
\* ======================================================================== */


//////////////////////////////////////////////////////////////////////////////
//
// MReadSocket
//
//////////////////////////////////////////////////////////////////////////////
#include "MReadSocket.h"

/*
 #ifdef _REENTRANT
 #include <pthread.h>
 #endif
*/

#include <unistd.h>          // usleep

#include <TMath.h>           // TMath::Min
#include <TTime.h>           // TTime
#include <TDatime.h>         // TDatime
#include <TSystem.h>         // gSystem
#include <TSocket.h>         // TSocket
#include <TServerSocket.h>   // TServerSocket

ClassImp(MReadSocket);

using namespace std;

MReadSocket::MReadSocket(int port, int mtu) : istream(this), fMtu(mtu), fTimeout(2500), fServSock(NULL), fRxSocket(NULL)
{
    fBuffer = new char[mtu];

    setg(fBuffer, fBuffer, fBuffer+1);

    cout << "Starting server socket on port 7000..." << endl;

    while (1)
    {
        fServSock=new TServerSocket(port, kTRUE);
        if (fServSock->IsValid())
            break;

        cout << "ServerSocket not valid: ";
        switch (fServSock->GetErrorCode())
        {
        case 0: cout << "No error." << endl; break;
        case -1: cout << "low level socket() call failed." << endl; break;
        case -2: cout << "low level bind() call failed." << endl; break;
        case -3: cout << "low level listen() call failed." << endl; break;
        default: cout << "Unknown." << endl; break;
        }

        delete fServSock;
        fServSock=NULL;
        break;
    }
    if (!fServSock)
    {
        cout << "MReadSocket: TServerSocket - Connection timed out." << endl;
        setstate(ios::failbit);
        return;
    }

    fServSock->SetOption(kNoBlock, 1);

    while (1)
    {
        const TTime timeout = gSystem->Now() + TTime(5000);

        TDatime now;
        cout << now.AsString() << ": Waiting for conntection on port 7000..." << endl;
        fRxSocket = NULL;
        while ((Long_t)fRxSocket<=0 && gSystem->Now()<timeout)
        {
            fRxSocket = fServSock->Accept();
            if (fRxSocket==0)
                cout << "Error: TServerSock::Accept" << endl;
            usleep(1);
        }

        if ((Long_t)fRxSocket<=0)
            continue;

        if (fRxSocket->IsValid())
            break;

        cout << "TSocket: Connection not valid..." << endl;
        delete fRxSocket;
    }

    if ((Long_t)fRxSocket<=0)
    {
        cout << "MReadSocket: TServerSocket::Accept - Connection timed out." << endl;
        fRxSocket=NULL;
        setstate(ios::failbit);
        return;
    }

    cout << "Connection established..." << endl;

    fRxSocket->SetOption(kNoBlock, 1);

    underflow();
}

// --------------------------------------------------------------------------
//
//  Destructor, destroying the gui mutex.
//
MReadSocket::~MReadSocket()
{
    if (fRxSocket)
        delete fRxSocket;
    if (fServSock)
        delete fServSock;

    delete fBuffer;

    cout << "Connection on Port 7000 closed." << endl;
}

// --------------------------------------------------------------------------
//
// This is called to flush the buffer of the streaming devices
//
int MReadSocket::sync()
{
    cout << "sync" << endl;
    return 0;
}

int MReadSocket::underflow()
{
    //
    // This simple trick should do its job, because the
    // TCP/IP stream is buffered already
    //
    const TTime timeout = fTimeout+gSystem->Now();

    Int_t l, len=-1;
    while (len<0 && gSystem->Now()<timeout)
    {
        fRxSocket->GetOption(kBytesToRead, l);
        if (l==0)
        {
            usleep(1);
            continue;
        }
        len = fRxSocket->RecvRaw(fBuffer, TMath::Min(fMtu, l));
    }

    if (len<0)
    {
        cout << "MReadSocket: TSocket::RecvRaw - Connection timed out." << endl;
        setstate(ios::failbit);
    }

    setg(fBuffer, fBuffer, fBuffer+len);
    return 0;
}