WatcherProducer.cpp

Go to the documentation of this file.

#ifndef Gw_WatcherProducer
#include "WatcherProducer.h"
#endif

// root includes
#ifndef ROOT_TROOT
#include "TROOT.h"
#endif
#ifndef ROOT_TSystem
#include "TSystem.h"
#endif
#ifndef ROOT_TH3
#include "TH3.h"
#endif

// ADF include
#ifndef ADF_FrameIO
#include "FrameIO.h"
#endif

using namespace ADF;
using namespace Gw;

//#define APPLY_PATCH

#ifdef APPLY_PATCH

#include <stdio.h>      /* for printf() and fprintf() */
#include <stdlib.h>     /* for malloc */
#include <sys/socket.h> /* for socket(), connect(), send(), and recv() */
#include <arpa/inet.h>  /* for sockaddr_in and inet_addr() */
#include <string.h>     /* for memset() */
#include <unistd.h>     /* for close() */
#include <errno.h>
#include <netinet/in.h>
#include <netdb.h>

// command to know if a port is open :  netstat -a | egrep 'Proto|LISTEN'

namespace  {
        int sock = 0;
}
#endif

void WatcherProducer::process_reset (UInt_t */*error_code*/)    
{
        CloseSocket();
}

WatcherProducer::~WatcherProducer()
{
        UInt_t error_code;
        process_reset(&error_code);
}

Bool_t WatcherProducer::SetConnection(const Char_t *machine, UInt_t port)
{
        fMachine = machine;
        fPort = port;
        
        return true;
/*
        if ( fSocket ) 
                { delete fSocket; }
        fSocket = new TSocket(machine,port);
        
        return fSocket->IsValid(); */
}       
                
Bool_t WatcherProducer::OpenSocket()
{
        // just in case
#ifdef APPLY_PATCH
        struct sockaddr_in servAddr; /* server address */
        struct sockaddr_in      sin_me;
        int one;
        struct hostent *host_addr;
        unsigned tmp;
        
        one=1;
//      printf("toto 1\n");
        sock = socket(AF_INET, SOCK_STREAM, 0);
        if (sock == 0) {
                strerror(errno);
                return false;
        }
//      printf("toto 2\n");

        memset((char *)&sin_me, 0, sizeof(sin_me));
//      printf("toto 3\n");
        sin_me.sin_port = 0;    /* free choice */
        memset(&servAddr, 0, sizeof(servAddr));
//      printf("toto 4 %s \n",fMachine.Data());
        servAddr.sin_family = AF_INET;             /* Internet address family */
        host_addr = gethostbyname(fMachine.Data());
//      printf("toto 5\n");
        servAddr.sin_family = host_addr->h_addrtype;

        memcpy((char *)&tmp,host_addr->h_addr,host_addr->h_length);
//      memcpy((char *)&tmp,host_addr->h_addr,sizeof(unsigned));

//      printf("toto 6\n");
        servAddr.sin_addr.s_addr = tmp;
        servAddr.sin_port = htons(fPort); /* Server port */     
//        printf("toto 7\n");
        setsockopt(sock,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one));
        bind(sock, (struct sockaddr *)&sin_me, sizeof(sin_me)); 
//      printf("titi %d %d \n",servAddr.sin_port,fPort);

        Int_t val = 0;
        fcntl(sock, F_SETFL, O_NONBLOCK);
        if ( ( val = connect(sock, (struct sockaddr *) &servAddr, sizeof(servAddr)) ) < 0)
        {
                close(sock);

        printf("toto 3 %d %s \n",val,strerror(errno));
      return false;
        }       

#else
        CloseSocket();
        // printf(" --> TRYING TO OPEN %s %d \n",fMachine.Data(),fPort);
        if ( (fSocket = new TSocket(fMachine.Data(),fPort)) ) {
                return fSocket->IsValid(); 
        }
#endif  
        return true;
}

Bool_t WatcherProducer::CloseSocket()
{
#ifdef APPLY_PATCH
        if ( sock )
                close(sock);
#else   
        if ( fSocket )
                { delete fSocket; fSocket = 0x0; }
#endif  
        return true;
}

void WatcherProducer::SetTimeout(Long_t timeout)
{
        fTimeout = timeout;
}

UInt_t WatcherProducer::ProcessBlock (FrameBlock &block) 
{
        // 0 ok
        // 1 means socket is not good
        // 2 means the real size of the next buffer is 0
        // 3 means failed to get the size of the next buf from the socket
        // 4 means too large buffer
        // 5 means fails to get the whole buffer from the socket
        UInt_t error_code = 0u; Int_t bsize = 0; Int_t ret;
        
        // check the connection
        if ( !OpenSocket() )
                return 1u;

/*
 ret = fSocket->RecvRaw(&bsize, sizeof(UInt_t),kDontBlock);
 if ( gDebug )
        Log << debug << " Receive size of the buffer " << ret << dolog;
 
 if ( ret == sizeof(UInt_t) && bsize > 0u ) {
        // if possible get the full buffer and ask ProcessBlock to treat it
        fBuffer.Offset();
        if ( fBuffer.Reserve(bsize) ) {
                if ( gDebug )
                        Log << debug << " Ask for the buffer " << dolog;
 
                ret = fSocket->RecvRaw(fBuffer.Address(), bsize);
                if ( gDebug )
                        Log << debug << " Receive buffer with size " << ret << dolog;
                if ( ret == bsize ) {   
                        // used linked with fBuffer, so no need to allocated a large block
                        block.SetBlock(fBuffer.Address(),bsize,'-');
                        if ( gDebug ) 
                                Log << info << "A new block has been received " << bsize << dolog;
                }
                else error_code = 5u;
        }
        else error_code = 4u;
 }
 else { 
         if ( bsize == 0 ) 
                error_code = 2u;
         else 
                 error_code = 3u; 
 }
*/
        
/*
        if ( gDebug )
                Log << debug << " Wait " << fTimeout << " to see if there are data " << dolog;  
        
        switch ( fSocket->Select(TSocket::kRead,fTimeout) ) {
                case -1:
                        error_code = 1u;
                        break;
                case  0:
                        error_code = 2u;
                        break;                  
                default:
                        ret = fSocket->RecvRaw(&bsize, sizeof(UInt_t),kDontBlock);
                        if ( gDebug )
                                Log << debug << " Receive size of the buffer " << ret << dolog;
                        
                        if ( ret == sizeof(UInt_t) && bsize > 0 ) {
                                // if possible get the full buffer and ask ProcessBlock to treat it
                                fBuffer.Offset();
                                if ( fBuffer.Reserve(bsize) ) {
                                        if ( gDebug )
                                                Log << debug << " Ask for the buffer " << dolog;
                                        ret = fSocket->RecvRaw(fBuffer.Address(), bsize);
                                        if ( gDebug )
                                                Log << debug << " Receive buffer with size " << ret << dolog;
                                        if ( ret == bsize ) {   
                                                // used linked with fBuffer, so no need to allocated a large block
                                                block.SetBlock(fBuffer.Address(),bsize,'-');
                                                if ( gDebug ) 
                                                        Log << info << "A new block has been received " << bsize << dolog;
                                        }
                                        else error_code = 6u;
                                }
                                else error_code = 5u;
                        }
                        else { 
                                if ( bsize == 0 ) 
                                        error_code = 3u;
                                else 
                                        error_code = 4u; 
                        }
                        break;
        }
*/

#ifdef APPLY_PATCH
        ret = recv (sock, &bsize, sizeof (UInt_t), 0);
//      ret = fSocket->RecvRaw(&bsize, sizeof(UInt_t),kDontBlock);
        if ( gDebug )
                Log << debug << " Receive size of the buffer " << ret << dolog;
        
        if ( ret == sizeof(UInt_t) && bsize > 0 ) {
                // if possible get the full buffer and ask ProcessBlock to treat it
                fBuffer.Offset();
                if ( fBuffer.Reserve(bsize) ) {
                        if ( gDebug )
                                Log << debug << " Ask for the buffer " << dolog;
                        // new code !
                        int total_received_data = 0;
                        while (total_received_data < bsize)
                        {
                                ret = recv (sock, fBuffer.CurrentAddress(),bsize, 0);
//                              ret = fSocket->RecvRaw(fBuffer.CurrentAddress(), bsize);
                                total_received_data += ret;
                                if ( ret < 1 ) {
                                        break;
                                }
                                fBuffer.SetOffset(total_received_data);
                                //   buf += received_data ;
                        }
                        if ( total_received_data == bsize ) {
                                // used linked with fBuffer, so no need to allocated a large block
                                block.SetBlock(fBuffer.Address(),bsize,'-');
                                if ( gDebug )
                                        Log << info << "A new block has been received " << bsize << dolog;
                        }
                        else error_code = 5u;
                }
                else error_code = 4u;
        }
        else {
                if ( bsize == 0 )
                        error_code = 2u;
                else
                        error_code = 3u;
        }
#else
        ret = fSocket->RecvRaw(&bsize, sizeof(UInt_t),kDontBlock);
        if ( gDebug )
                Log << debug << " Receive size of the buffer " << ret << dolog;
        
        if ( ret == sizeof(UInt_t) && bsize > 0 ) {
                // if possible get the full buffer and ask ProcessBlock to treat it
                fBuffer.Offset();
                if ( fBuffer.Reserve(bsize) ) {
                        if ( gDebug )
                                Log << debug << " Ask for the buffer " << dolog;

                        switch ( fSocket->Select(TSocket::kRead,2) ) 
                        {
                                case 0:
                                        Log << info << "--> timeout " << dolog;
                                        error_code = 2u;
                                        break;
                                case -1:
                                        Log << info  << "--> Error " << dolog;
                                        error_code = 5u;
                                        break;
                                default:
                        //              fSocket->RecvRaw(&bsize, sizeof(Int_t));
                        //              cout << " Size expected " << bsize << endl;
                                        ret = fSocket->RecvRaw(fBuffer.Address(), bsize);
                                        if ( ret == bsize ) {
                                                block.SetBlock(fBuffer.Address(),bsize,'-');
                                        }
                                        else 
                                                Log << info << " Size received too small " << ret << " expected " << bsize << dolog;
                        //              cout << " Size received " <<  ret << endl;

                                break;
                        }

                        // new code !
/*
                        int total_received_data = 0;
                        while (total_received_data < bsize)
                        {
                                ret = fSocket->RecvRaw(fBuffer.CurrentAddress(), bsize);
                                total_received_data += ret;
                                if ( ret < 1 ) {
                                        break;
                                }
                                fBuffer.SetOffset(total_received_data);
                                //   buf += received_data ;
                        }
                        if ( total_received_data == bsize ) {
                                // used linked with fBuffer, so no need to allocated a large block
                                block.SetBlock(fBuffer.Address(),bsize,'-');
                                if ( gDebug )
                                        Log << info << "A new block has been received " << bsize << dolog;
                        }
                        else error_code = 5u;
*/
                }
                else error_code = 4u;
        }
        else {
                if ( bsize == 0 )
                        error_code = 2u;
                else
                        error_code = 3u;
        }

#endif

        
        // Narval way to work: the connection is open only to transfert one buffer
        CloseSocket();
        // return error
        return error_code;
}