source: Sophya/trunk/AddOn/TAcq/sockrawstream.cc@ 3659

#include <stdlib.h>
#include <string.h>

#include "sockrawstream.h"
#include "sopnamsp.h"
#include "pexceptions.h"


/*!
   \class SOPHYA::RawInOutSocketStream
   \ingroup SysTools
   This class implements the interface defined by RawInOutStream 
   over a socket (read and write operations).
   It is mainly intended to be used by PPF In/Out streams.
*/

// Taille de base pour les messages echanges = Longueur header = long. Trailer
// Longueur de trailer pour marquer la fin 
#define   RIOS_BBSIZE  64

/* --Methode-- */
RawInOutSocketStream::RawInOutSocketStream(Socket &skt, size_t npaqbuff)
  : RawInOutStream() , _skt(skt)
{

// ---- preparation de la chaine d'identification / taille du buffer 
  const char * hdrmarq = "SIPSS-RawInOutSocketStream-Header";
//                        0123456789012345678901234567890123456789
  char msg[RIOS_BBSIZE];
  for(size_t k=30; k<RIOS_BBSIZE; k++) msg[k]='\0';
  strcpy(msg, hdrmarq);
  if (npaqbuff < 1) npaqbuff = 1;
  size_t sbsz =  npaqbuff*RIOS_BBSIZE;
  sprintf(msg+48,"%ld",(long)sbsz);
// ----- Envoi du marqueur de debut et negociation de taille du buffer 
  Send(msg, (size_t)RIOS_BBSIZE);
  Receive(msg, (size_t)RIOS_BBSIZE);
  if (strcmp(msg, hdrmarq)!=0)  
    throw SocketException("RawInOutSocketStream::RawInOutSocketStream()- ExchangeMsg check failed");

  size_t rbsz = atol(msg+48);
  size_t bufsize = (sbsz >= rbsz) ? sbsz : rbsz;

//DBG  cout << "RIOSS::RawInOutSocketStream()*DBG* sbsz=" << sbsz << " rbsz=" << rbsz << " -->bufsize=" << bufsize << endl;
 
  _rdbuff.buff = new char[bufsize];
  _rdbuff.sz = bufsize;
  _rdbuff.cpos = bufsize;

  _wrbuff.buff = new char[bufsize];
  _wrbuff.sz = bufsize;
  _wrbuff.cpos = 0;
  
}

/* --Methode-- */
RawInOutSocketStream::~RawInOutSocketStream()
{
//  cout << " ---- DESTRUCTEUR ---- ~RawInOutSocketStream() " << endl;
  //----- ecriture/envoi du buffer si pas vide 
  if (_wrbuff.cpos > 0) {
    for (size_t k= _wrbuff.cpos; k<_wrbuff.sz; k++)  _wrbuff.buff[k] = '\0';
    _wrbuff.cpos = _wrbuff.sz;
    SendBuffer();
  }
  // Envoi du marquer de fin (trailer) 
  const char * trlmarq = "TRAILER-SIPSS-RawInOutSocketStream-Trailer";
//                        0123456789012345678901234567890123456789
  char msg[RIOS_BBSIZE];
  for(size_t k=30; k<RIOS_BBSIZE; k++) msg[k]='\0';
  strcpy(msg, trlmarq);
  Send(msg, (size_t)RIOS_BBSIZE);

  bool oktrailer=false;
  size_t nrdt = 0;
  size_t maxtry = 16*_rdbuff.sz / RIOS_BBSIZE;
  while(!oktrailer && (nrdt<maxtry)) {
    Receive(msg, (size_t)RIOS_BBSIZE);
    nrdt++;
    if (strcmp(msg, trlmarq)==0)  oktrailer = true;
  }
//DBG  cout << "RIOSS::~RawInOutSocketStream()*DBG* nrdt=" << nrdt << " Trailer:" << ((oktrailer)?" OK":" ERROR") << endl;

  delete [] _rdbuff.buff;
  delete [] _wrbuff.buff;
  if (!oktrailer) 
    throw SocketException("RawInOutSocketStream::~RawInOutSocketStream(): EndOfStream marker not found");
}

/* --Methode-- */
size_t RawInOutSocketStream::CopyToSendBuffer(const char* s, size_t n)
{

  size_t len = _wrbuff.sz-_wrbuff.cpos;
  if (len > n) len = n;
  if ( (_wrbuff.cpos == 0) && (len == _wrbuff.sz) ) {
 //DBG   cout << "RawInOutSocketStream::CopyToSend/DBG-1 - n=" 
//DBG    << n << " len=" << len << endl;
    Send(s, len);
  }
  else {
 //DBG   cout << "RawInOutSocketStream::CopyToSend/DBG-2 - n=" 
//DBG    << n << " len=" << len << " cpos=" << _wrbuff.cpos << endl;
    memcpy(_wrbuff.buff+_wrbuff.cpos, s, len);
    //DBG    cout << " memcpy OK - cpos= " << _wrbuff.cpos << endl;
    _wrbuff.cpos += len;
    if (_wrbuff.cpos == _wrbuff.sz)  SendBuffer();
  }
  return len;
}

/* --Methode-- */
size_t RawInOutSocketStream::CopyFromRecvBuffer(char* s, size_t n)
{
  if ( (_rdbuff.cpos == _rdbuff.sz) && ( n >= _rdbuff.sz)) {
    size_t len = _rdbuff.sz;
  //DBG  cout << "RawInOutSocketStream::CopyFromRecv/DBG-1 - n=" << n 
//DBG    << " len=" << len << endl;
    Receive(s, len);
    return len;
  }
  else {
//DBG    cout << "RawInOutSocketStream::CopyFromRecv/DBG-2 - n=" << n << endl;
    if (_rdbuff.cpos == _rdbuff.sz)  ReceiveBuffer();
    size_t len = _rdbuff.sz-_rdbuff.cpos;
    if (len > n) len = n;
    memcpy(s, _rdbuff.buff+_rdbuff.cpos, len);
    _rdbuff.cpos += len;
    return len;
  }
}

/* --Methode-- */
void RawInOutSocketStream::SendBuffer()
{
//DBG  cout << "RawInOutSocketStream::SendBuffer/DBG - cpos=" 
 //DBG      <<  _wrbuff.cpos << " sz=" << _wrbuff.sz << endl;
  //  if (_wrbuff.cpos != _wrbuff.sz)  return;
  Send(_wrbuff.buff, _wrbuff.sz);
  _wrbuff.cpos = 0;
}

/* --Methode-- */
void RawInOutSocketStream::ReceiveBuffer()
{
  //  if (_rdbuff.cpos != _rdbuff.sz)  return;
 //DBG cout << "RawInOutSocketStream::ReceiveBuffer/DBG - cpos=" 
 //DBG      <<  _rdbuff.cpos << " sz=" << _rdbuff.sz << endl;
  Receive(_rdbuff.buff, _rdbuff.sz);
  _rdbuff.cpos = 0;
  
}

/* --Methode-- */
size_t RawInOutSocketStream::Send(const char* s, size_t n)
{
  size_t  nst = 0;
  while (nst < n) {
    size_t ns = _skt.Send(s+nst, n-nst);
    if (ns < 1)  break;
    nst += ns;
  }
  if ( nst < n) 
    throw IOExc("RawInOutSocketStream::Send()/write() Error nwrite < n");
  return nst;
}

/* --Methode-- */
size_t RawInOutSocketStream::Receive(char* s, size_t n)
{
  size_t  nst = 0;
  int ntry = 0;
  while (nst < n) {
    size_t ns = _skt.Receive(s+nst, n-nst);
    ntry++;
    if (ns < 1)  break;
    nst += ns;
  }
  if ( nst < n) {
    cout << "  RawInOutSocketStream::Receive() / Pb ! ntry=" << ntry
         << " nst=" << nst << " n=" << n << endl;
    throw IOExc("RawInOutSocketStream::Receive/read() Error nread < n");
  }
  return nst;
}

int_8 RawInOutSocketStream::tellg()
{
  return _totnrd;
}

/* --Methode-- */
RawInOutStream& RawInOutSocketStream::read(char* s, uint_8 n)
{
//DBG  cout << "RawInOutSocketStream::read()/DBG - n=" << n << endl;
  size_t  nst = 0;
  while (nst < n) {
    size_t ns = CopyFromRecvBuffer(s+nst, n-nst);
    if (ns < 1)  break;
    nst += ns;
  }
  if ( nst < n) 
    throw IOExc("RawInOutSocketStream::read() Error nread < n");
  _totnrd += n;
  return *this;
}

/* --Methode-- */
int_8 RawInOutSocketStream::tellp()
{
  return _totnwr;
}

/* --Methode-- */
RawInOutStream& RawInOutSocketStream::write(const char* s, uint_8 n)
{
//DBG  cout << "RawInOutSocketStream::write()/DBG - n=" << n << endl;
  size_t  nst = 0;
  while (nst < n) {
    size_t ns = CopyToSendBuffer(s+nst, n-nst);
    if (ns < 1)  break;
    nst += ns;
  }
 //DBG cout << "RawInOutSocketStream::write()/DBG ---> nst=" << nst << endl;
  
  if ( nst < n) 
    throw IOExc("RawInOutSocketStream::write() Error nwrite < n");
  _totnwr += n;
  return *this;
}