// ///////////////////  ***********************  \\\\\\\\\\\\\\\\\\\\ //
//
//  DsComputer, part of DSI (Distributed Scene Inventor)
//
//  DSI is an extension to popular Open Inventor and enpowers it
//  by distributed scene management
//
//  Copyright (C) 2001-2002 by PC John (pcjohn@email.cz).
//  All rights reserved.
//
//  This file is free software; you can redistribute it and/or
//  modify it under the terms of the GNU Lesser General Public License
//  version 2.1 as published by the Free Software Foundation. See the
//  file LICENSE.LGPL at the root directory of the distribution for
//  more details.
//
// \\\\\\\\\\\\\\\\\\\  ***********************  //////////////////// //


#include <Inventor/SoInput.h>

#include <Inventor/cve/SoComputer.h>
#include <Inventor/cve/SoNetwork.h>
#include <Inventor/cve/SoNetworkP.h>
#include <Inventor/cve/TNet.h>
#include <Inventor/cve/SbTransaction.h>
#include <Inventor/cve/SoDistributionGroup.h>



/*! Returns the state of the network connection. 
 *  If the computer asked for disconnection,
 *  the state will be changed to DISCONNECTING after receiving all data sent by the computer.
 */
SoComputer::State SoComputer::getState()
{
  if (netc <= 0)
    return (netc==0) ? DISCONNECTED : BROKEN_CON;

  TNState prevState = tnPeekState(netc);

  switch (tnGetState(netc)) {
  case TN_CONNECTING:     return CONNECTING;
  case TN_CONNECTED:      if (prevState == TN_CONNECTING && !tsToSendAfterConnect.isZero()) {
                            SbNMNullMessage *msg = new SbNMNullMessage;
                            msg->timeStamp = tsToSendAfterConnect;
                            msg->sendTo(this);
                            tsToSendAfterConnect = SbTimeStamp::zero();
                          }
                          return CONNECTED;
  case TN_DISCONNECTING:  return DISCONNECTING;
  case TN_BROKEN:         return BROKEN_CON;

  case TN_DISCONNECTED:   getLocalAddr(); // to make sure the address is cached before connection is closed
                          tnClose(netc); 
                          netc = 0; 
                          return DISCONNECTED;

  case TN_LISTENING:
  case TN_FREE:
  default: assert(FALSE); return DISCONNECTED;
  }
}


SoComputer::State SoComputer::peekState() const
{
  if (netc <= 0)
    return (netc==0) ? DISCONNECTED : BROKEN_CON;

  switch (tnPeekState(netc)) {
  case TN_CONNECTING:     return CONNECTING;
  case TN_CONNECTED:      return CONNECTED;
  case TN_DISCONNECTING:  return DISCONNECTING;
  case TN_BROKEN:         return BROKEN_CON;
  case TN_DISCONNECTED:   return DISCONNECTED;
  case TN_LISTENING:
  case TN_FREE:
  default: assert(FALSE); return DISCONNECTED;
  }
}


//! Constructor.
SoComputer::SoComputer() : netc(0), addr(0,0), localAddrState(EMPTY), recvStreamBuf(NULL), 
                           recvStreamPos(0), largestTimeStamp(SbTimeStamp::zero()),
                           tsToSendAfterConnect(SbTimeStamp::zero()), activeState(NON_ACTIVE)
{
  SoNetworkP::addComputer(this);
}


//! Constructor.
SoComputer::SoComputer(SbNetConnection netc) : netc(0), addr(0,0), localAddrState(EMPTY),
                       recvStreamBuf(NULL), recvStreamPos(0), largestTimeStamp(SbTimeStamp::zero()),
                       tsToSendAfterConnect(SbTimeStamp::zero()), activeState(NON_ACTIVE)
{
  SoNetworkP::addComputer(this);
  this->connectByNetC(netc);
}


/*! Destructor. Note: Destructing of computer in other state than DISCONNECTED 
 *  may lead to data loss. Particulary, unsend data from send buffers are lost,
 *  unacknowledged sent data may also be undelivered, and all unreceived data are discarded.
 */
SoComputer::~SoComputer()
{
  this->dropConnection();
  free(recvStreamBuf);
  SoNetworkP::removeComputer(this);
}


/*! Receives data from computer and returns amount of them.
 *  The received data are copied into the buffer. If no data arrived,
 *  zero is returned. Zero is returned also, if the computer 
 *  is not in CONNECTED.
 */
int SoComputer::recvData(void *buf, int size)
{
  if (netc > 0) 
    return tnRecv(netc, buf, size);
  else 
    return 0;
}


//! Sends data to the computer. Actual data sending happens in CONNECTED state only.
void SoComputer::sendData(const void *buf, int size)
{
  if (netc > 0)
    tnSend(netc, buf, size);
  else {
    dataIntegrityFailure(); // to indicates improper use of this function and possible data inconsistency
    return;
  }
}


//! Sends data to the group of computers. The function may use multicast for sending.
void SoComputer::sendData(SoDistributionGroup *dg, const void *buf, int size, SbBool sendToNotActive)
{
  const int c = dg->computer.getNum();
  int i;
  for (i=0; i<c; i++)
    if (sendToNotActive || dg->active[i] == TRUE)
      dg->computer[i]->sendData(buf, size);
}


void SoComputer::sendData(SoDistributionGroup *dg, const void *buf, int size, const SoMFComputer &list)
{
  const int c = list.getNum();
  int i;
  for (i=0; i<c; i++)
    list[i]->sendData(buf, size);
}


void SoComputer::sendMessage(SbNetMessage *msg)
{
  assert(FALSE);
}


SbBool SoComputer::messageReceived()
{
  // notes on the stream:
  // - recvStreamPos index includes 4 bytes of recvStreamSize 
  //   and 4 bytes of stream identifier ("TRA","MSG"...)
  // - recvStreamSize includes 8 bytes of previous two points
  // - if recvStreamPos < 4, then recvStreamSize contains identifier "TRA","MSG",...

  // receive "recvStreamSize"
  if (recvStreamPos < 8) {
    uint8_t buf[8];
    int count = this->recvData(&buf[recvStreamPos], 8-recvStreamPos);
    for (; recvStreamPos<4 && count>0; recvStreamPos++, count--)
      recvStreamType[recvStreamPos] = buf[recvStreamPos];
    for (; recvStreamPos<8 && count>0; recvStreamPos++, count--)
      ((uint8_t*)&recvStreamSize)[recvStreamPos-4] = buf[recvStreamPos];
      
    if (recvStreamPos < 8)
      return FALSE;

    // asserts for detecting broken data stream
    assert(strcmp(&recvStreamType[0], "TRA") == 0 || strcmp(&recvStreamType[0], "MSG") == 0 ||
           strcmp(&recvStreamType[0], "TYP") == 0 && "Broken network stream.");
    assert(recvStreamSize >= 8);
    
    recvStreamBuf = malloc(recvStreamSize-8);
  }

  assert(recvStreamPos >= 8 && "Something is wrong.");

  if (recvStreamPos==recvStreamSize)  return TRUE;
  
  // receive stream
  recvStreamPos += this->recvData(&(((uint8_t*)recvStreamBuf)[recvStreamPos-8]), recvStreamSize-recvStreamPos);

  // return   
  return (recvStreamPos==recvStreamSize);
}


SbNetMessage* SoComputer::receiveMessage(SoDistributionGroup *dg)
{
  // make sure a transaction is present
  if (!messageReceived())
    return NULL;

  // read the transaction
  SoInput in;
  in.setBuffer(recvStreamBuf, recvStreamSize-8);
  SbNetMessage *msg = SbNetMessage::readMessage(&recvStreamType[0], &in, this);
  if (msg == NULL) {
    // the failure was already handled, let's just return here
    return NULL;
  }

  // update largest timestamp
  if (msg->isTransaction())
    updateLargestTimeStamp(((SbTransaction*)msg)->getTimeStamp());

  // clean up
  recvStreamPos = 0;
  free(recvStreamBuf);
  recvStreamBuf = NULL;

  return msg;
}


SbBool SoComputer::process()
{
  // FIXME: only default distribution group is supported for now
  SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();

  while (messageReceived()) {
    SbNetMessage *msg = this->receiveMessage(dg);
    if (msg == NULL) break;
    if (msg->isTransaction())
      dg->onTransactionArrival((SbTransaction*)msg);
    else {
      msg->process();
      delete msg;
    }
  }

  return getState() == CONNECTED;
}


const SbTimeStamp& SoComputer::getLargestTimeStamp()
{
  return largestTimeStamp;
}


void SoComputer::updateLargestTimeStamp(const SbTimeStamp &timeStamp)
{
  if (largestTimeStamp < timeStamp) {
    largestTimeStamp = timeStamp;

    // update timeStamp generator
    SbTimeStamp::adjustGeneratorFloor(timeStamp);
  }
}


void SoComputer::updateLinkTimesOfNonActiveComputers(const SbTimeStamp &ts, SoDistributionGroup *dg)
{
  SoComputerList list;
  int i,c = dg->computer.getNum();
  for (i=0; i<c; i++)
    if (dg->active[i] == FALSE) {
      SoComputer *comp = dg->computer[i];
      if (comp->peekState() == CONNECTING) {
        // handle connecting computers
        if (comp->tsToSendAfterConnect.isZero() || comp->tsToSendAfterConnect < ts)
          comp->tsToSendAfterConnect = ts;
      } else
        // send link time directly to connected computers
        list.append(comp);
    }

  // send the message
  SbNMNullMessage *msg = new SbNMNullMessage;
  msg->timeStamp = ts;
  msg->sendToComputers(list);
}


//! Returns the information about ip and port of the connection on the peer.
SbInetAddr SoComputer::getAddr() const
{
  return addr;
}


//! Returns the information about ip and port of the connection on this computer.
SbInetAddr SoComputer::getLocalAddr() const
{
  switch (localAddrState) {
  case CACHED:  return localAddr;
  case INVALID: if (netc > 0)
                  if (tnPeekState(netc) != TN_CONNECTING) {
                    SoComputer *comp = (SoComputer*)this; // to put out const
                    comp->localAddrState = CACHED;
                    comp->localAddr = SbInetAddr(tnGet(TN_LOCAL_IP, netc), tnGet(TN_LOCAL_PORT, netc));
                    return comp->localAddr;
                  }
  case EMPTY:   return SbInetAddr(0,0);
  default: assert(FALSE); return SbInetAddr(0,0);
  };
}


//! Returns network connection.
SbNetConnection SoComputer::getNetC() const
{
  if (netc > 0) return netc;
  else return 0;
}


//! Returns connection's socket.
int SoComputer::getSocket() const
{
  int c = this->getNetC();
  return tnGet(TN_SOCKET, c);
}


SbBool SoComputer::isThisComputer() const
{
  return FALSE;
}


SbComputerIdentity SoComputer::getComputerIdentity(SoDistributionGroup *dg) const
{
  SoDistributionGroup *dg2 = (dg ? dg : SoNetwork::getDefaultDistributionGroup());
  return dg2->getComputerIdentity(this);
}


SoDistributionGroup* SoComputer::getCurrentDistributionGroup() const
{
  // FIXME: dg identity should be transmitted over network. Just now let's use default dg. PCJohn 2006-05-24
  return SoNetwork::getDefaultDistributionGroup();
}


/*! Connects the computer to the address given by addr parameter.
 *  Any port and any computer interface can be used for the connection.
 */
void SoComputer::connect(const SbInetAddr &addr)
{
  this->connect(addr, SbInetAddr(0,0));
}


/*! Connects the computer to the address given by addr parameter.
 *  The port and computer interface are specified by localAddr parameter.
 */
void SoComputer::connect(const SbInetAddr &addr, const SbInetAddr &localAddr)
{
  this->dropConnection();
  int c = tnConnectFromIP(addr.getIP(), addr.getPort(), localAddr.getIP(), localAddr.getPort());

  if (c != -1) {
    netc = c;
    this->addr = addr;
  } else
    this->addr = SbInetAddr(0,0);

  this->localAddrState = INVALID;
  largestTimeStamp = SbTimeStamp::zero();
}


//! Associate the computer with the network connection.
void SoComputer::connectByNetC(SbNetConnection nc)
{
  this->dropConnection();
  if (nc != 0)
    switch (tnGetState(nc)) {
    case TN_DISCONNECTED:
    case TN_BROKEN:       break;
    case TN_LISTENING:
    case TN_FREE:         assert(FALSE); break;
    default:
      netc = nc;
      addr = SbInetAddr(tnGet(TN_PEER_IP, nc), tnGet(TN_PEER_PORT, nc));
      localAddrState = INVALID;
      return;
    }

  addr = SbInetAddr(0,0);
  localAddrState = INVALID;
  largestTimeStamp = SbTimeStamp::zero();
}


//! Send disconnect request through connection.
void SoComputer::disconnect()
{
  if (netc <= 0) {
    netc = 0;
    return;
  }

  tnDisconnect(netc);
}


/*! Immediatelly terminates network connection. Note: It may result in some data lost,
 *  e.g. undelivered data from the peer, or unsend data from sending buffers.
 */
void SoComputer::dropConnection()
{
  if (netc > 0)
    tnClose(netc);
  netc = -1;
}


void SoComputer::dataIntegrityFailure()
{
  dropConnection();
  netc = -1;
}


SoComputer::ActiveState SoComputer::getActiveState() const
{
  return activeState;
}


void SoComputer::activate()
{
  activeState = ACTIVE;
}


void SoComputer::deactivate()
{
  activeState = NON_ACTIVE;
}
