/*****************************
File:       RemoteConnection.CPP
Language:   C++ (header)
Project:    H3DNetworkingUtils
The contents of this file are subject to the Mozilla Public License
Version 1.1 (the "License"); you may not use this file except in
compliance with the License. You may obtain a copy of the License at
http://www.mozilla.org/MPL/

Software distributed under the License is distributed on an "AS IS"
basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
License for the specific language governing rights and limitations
under the License.

The Original Code is H3DNetworkingUtils v1.0.

The Initial Developer of the Original Code is CSIRO.
Portions created by the Initial Developer are Copyright (C) 1009 CSIRO. All Rights Reserved.
Contributor(s):
    Chris Gunn  <Chris.Gunn@csiro.au> <ChrisJGunn@gmail.com>
***************************/
#include <stdio.h>

#include "H3DNetworkingUtils/RemoteConnection.h"
#include "H3DNetworkingUtils/fail.h"
#include "H3DNetworkingUtils/threadwrap.h"
#include "H3DNetworkingUtils/systypes.h"
#include "H3DNetworkingUtils/RemoteSync.h"
#include <iomanip>
#include "H3DNetworkingUtils/EndianCheck.h"
///#include <unistd.h>
#include "H3DNetworkingUtils/RemoteTimePacket.h"
#ifdef DEMO_LICENCE
#include "CU_DemoLicence.h"
#endif

using namespace H3D;
using namespace H3DNetworkingUtils;

///const int RemoteConnection::MAX_PACKET_SIZE = 80;  DEBUG   Need a variasble size!
const int RemoteConnection::MAX_PACKET_SIZE = 5000;

const int RemoteConnection::MAX_ID = 65535;

H3DNodeDatabase RemoteConnection::database(   
  "RemoteConnection", 
  NULL,
  typeid(RemoteConnection),
  &Group::database);

namespace RemoteConnectionInternals {   
  FIELDDB_ELEMENT( RemoteConnection, errorMessage,      OUTPUT_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, infoMessage,       OUTPUT_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, periodicSend,      INITIALIZE_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, periodicSendRate,  INITIALIZE_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, simulatedLatency,  INPUT_OUTPUT );
  FIELDDB_ELEMENT( RemoteConnection, remoteFields,      INITIALIZE_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, remoteSync,        INITIALIZE_ONLY );
  FIELDDB_ELEMENT( RemoteConnection, isConnected,       OUTPUT_ONLY );
}

// ----------------------------------------------------------------------------
RemoteConnection::RemoteConnection( Inst<SFString           >     _errorMessage,
                                    Inst<SFString           >     _infoMessage,
                                    Inst<SFFloat            >     _simulatedLatency,
                                    Inst<MFRemoteField      >     _remoteFields,
                                    Inst<SFBool             >     _periodicSend,
                                    Inst<SFFloat            >     _periodicSendRate,
                                    Inst<SFRemoteSync       >     _remoteSync,
                                    Inst<BufferedSField<SFBool> >   _isConnected) :
errorMessage         ( _errorMessage ),
infoMessage          ( _infoMessage ),
simulatedLatency     ( _simulatedLatency ),
remoteFields         ( _remoteFields ),
periodicSend         (_periodicSend),
periodicSendRate     (_periodicSendRate),
remoteSync           (_remoteSync),
isConnected          (_isConnected ),
send_loop_threadP    (0),
receive_loop_threadP (0),
last_send_time       (0),
last_try_time        (0),
sockP                ( 0 ),
close_requested      ( false ),
open_comms_cb_fn     ( 0 ),
open_comms_cb_data   ( 0 ),
close_comms_cb_fn    ( 0 ),
close_comms_cb_data  ( 0 ),
read_packet          (MAX_PACKET_SIZE),
write_packet         (MAX_PACKET_SIZE),
seq_number           (0),
use_seq_numbers      (false),
sim_latency          (false),
little_endian        (false),
comms_open_requested (false),
error_msg            (0),
info_msg             (0),
error_msg_waiting    (false),
info_msg_waiting     (false),
connected_sem        (0),
remote_client_addr   (0),
is_connected         (false),
shutting_down        (false),
send_loop_running    (false),
receive_loop_running (false),
initialized          (false) {

   type_name = "RemoteConnection";
   database.initFields( this );

#ifdef DEMO_LICENCE
  CU_DemoLicence demo_licence(200, 1093737903); // 200 days from 30/08/2004
  demo_licence.check();
#endif
  remoteFields->conP = this; 
  simulatedLatency->setValue(0);
  periodicSend->setValue(false);
  periodicSendRate->setValue(100);
  little_endian = EndianCheck::isLittleEndian();
  isConnected->initialiseValue(false);

}


// ----------------------------------------------------------------------------
RemoteConnection::~RemoteConnection() {
  std::cerr << "~RemoteConnection" << std::endl;
  setShuttingDown(true);
  // Wait for threads to finish
  ///if (receive_loop_threadP)            {
  ///  // We have to explicitely cancel the receive thread becasue it might be blocked at a receive() call.
  ///  receive_loop_threadP->cancel();
  ///  receive_loop_threadP->wait();
  ///  receive_loop_threadP = NULL;
  /// }

  while (sendLoopRunning() || receiveLoopRunning()) {
     Sleep(50);
  }

  if (sockP) {
    sockP->close();
    delete sockP;
    sockP = 0;
  }
  if (remote_client_addr) {
    delete remote_client_addr;
  }
 /// if (send_loop_threadP) {
 ///    send_loop_threadP->cancel();
 ///    send_loop_threadP->wait(); // Wait for it to die
 /// }
  cout << "end ~RemoteConnection" << endl;
}

// ----------------------------------------------------------------------------
void RemoteConnection::initialize() {
  Group::initialize();


  // Tell each of the remote fields what connection to use (i.e. this one):
/***
  TypedMFNode< RemoteField >::const_iterator i = remoteFields->begin();
  while ( i != remoteFields->end() ) {
    (*i)->conP = this;
    if (periodicSend->getValue()) {
      (*i)->setSendOnChange(false);
    }
    else {
      (*i)->setSendOnChange(true);
    }
    i++;
  }
 Done in enterNode***/

  if (periodicSend->getValue()) {
    if (periodicSendRate->getValue() <= 0) {
      period = 999999; // Allow for silly entry
    }
    else {
      period = 1 / periodicSendRate->getValue();
    }
  }

  H3DFloat lat = simulatedLatency->getValue();
  if (lat > 0.0) {
      max_sim_latency_buf_size = (unsigned int)(lat * (float)periodicSendRate->getValue());
      sim_latency = true;
//    }
  }

  initialized = true;

}


// ----------------------------------------------------------------------------
void RemoteConnection::doGraphicsLoopActions() {
  // Called in graphics thread.

  isConnected->checkForChange(id);
  if (isConnected->getValue() != is_connected.get()) {
     isConnected->setValue(is_connected.get());
  }

  if (commsOpenRequested()) {
    is_connecting = true;
    comms_open_requested = false;
    if (periodicSend->getValue()) {
      if (!send_loop_threadP) {
        send_loop_threadP = new Thread((ThreadFunction)sendLoopTH, this);
      }
    }
    close_requested = false;
    if (open_comms_cb_fn) {
      (*open_comms_cb_fn)(open_comms_cb_data);
    }
    if (sockP && sockP->isTCP()) {
      setConnected(true); // For UDP we need to wait until either connect is
      // called or data is received.
      isConnected->checkForChange(id);

      // Go through all fields, sending them once if they want to be sent on
      // connection.
      if (remote_client_addr) {
        delete remote_client_addr;
      }
      remote_client_addr = new char[strlen(sockP->remoteAddress()->text()) + 1];
      strcpy(remote_client_addr, sockP->remoteAddress()->text());
      sendFields(true);
      connected_sem.signal();
    }
  }

  if (is_connected.get()) {
    is_connecting = false;
    if (close_requested) {
      close();
      close_requested = false;
    }
  } else if (!is_connecting.get()) {
    doClosedActions();
  }

  msg_lock.lock();
  if (error_msg_waiting) {
    errorMessage->setValue(error_msg, id);
    cout << error_msg << endl;
    error_msg_waiting = false;
  }
  if (info_msg_waiting) {
    infoMessage->setValue(info_msg, id);
    cout << info_msg << endl;
    info_msg_waiting = false;
  }
  msg_lock.unlock();

}

// ----------------------------------------------------------------------------
void RemoteConnection::close(  ) {
  setConnected(false);
}

// ----------------------------------------------------------------------------
void RemoteConnection::showError(const char * msg) {
  // This function can be called by different threads, therefore lock it.
  msg_lock.lock();
  error_msg_waiting = true;
  delete error_msg;
  error_msg = new char[strlen(msg) + 1];
  strcpy(error_msg, msg);
  msg_lock.unlock();
}

// ----------------------------------------------------------------------------
void RemoteConnection::clearError() {
  // This function can be called by different threads, therefore lock it.
  msg_lock.lock();
  error_msg_waiting = true;
  error_msg = new char[1];
  strcpy(error_msg, "");
  msg_lock.unlock();
}

// ----------------------------------------------------------------------------
void RemoteConnection::showInfo(const char * msg) {
  msg_lock.lock();
  info_msg_waiting = true;
  delete info_msg;
  info_msg = new char[strlen(msg) + 1];
  strcpy(info_msg, msg);
  msg_lock.unlock();
}

// ----------------------------------------------------------------------------
void RemoteConnection::requestClose() {
  comms_lock.lock();
  if (is_connected.get() && !close_requested) {
    char tmp[80];
    sprintf(tmp, "TCP Client %s disconnected\n", remote_client_addr);
    showInfo(tmp);
    close_requested = true;
  }
  comms_lock.unlock();
}

// ----------------------------------------------------------------------------
void RemoteConnection::receiveLoop () {
  // Called in receiving thread

  // The following loop is required because I can't get sigignore(SIGCLD) to
  // work. If I let the receiving process exit, the parent process gets killed.
  // So I let it idly loop if it is closed, to awaken when closed becomes
  // false.
  // The function returns when a close is requested.

  setReceiveLoopRunning(true);
  while (!isShuttingDown()) {
  ///while (true) {
    if (initialized) {
      if (is_connected.get()) {
        if (close_requested) {
          // Do nothing .. waiting for graphics thread to close the connection.
          Sleep(500); // Win32 Sleep is millisleep
        }
        else {
          if (receivePacket()) {// Get from socket
            readPacket(); // Interpret it.
          }
        }
      }
      else { // not isConnected
        return;
      }
    }
    else {
      Sleep(500);// Win32 Sleep is millisleep
    }
  }
  setReceiveLoopRunning(false);
}

// ----------------------------------------------------------------------------
bool RemoteConnection::receivePacket() {
  // Called in receiving thread

  if (!is_connected.get()) {
    return false;
  }

  sock_lock.lock();
  while (!sockP->dataReady()) {
     if (isShuttingDown()) {
        return false;
     }
     Sleep(2);// Win32 Sleep is millisleep
  }
  bool ok = read_packet.receive(static_cast<TCPSock*>(sockP)); // This will block until data arrives.
  sock_lock.unlock();
  if (!ok) {
    cout << "receive failure" << endl;
    requestClose();
  }
  return ok;
  // There is a problem here if the sender is sending packets faster than we
  // can chew them up. This is hard to resolve - but we can use the dataReady()
  // function of sockwrap to tell if there is more data waiting.
  // Attend to this later if the problem occurs.
}

// ----------------------------------------------------------------------------
void RemoteConnection::readPacket() {
  // Called in receiving thread

  if (use_seq_numbers) {
    unsigned short seq_num = read_packet.getSeqNum();
    if (!seq_checker.inSequence(seq_num)) {
      cerr << 'X';
      return; // discard
    }
  }
  int fld_id = read_packet.getId();
  readField(fld_id);
}

// ----------------------------------------------------------------------------
void RemoteConnection::readField(int field_id) {
   // Called in receiving thread

   // Check to see which of the remote fields matches the supplied id.
   TypedMFNode< RemoteField >::const_iterator i = remoteFields->begin();
   RemoteField * rfP = NULL;
   while ( i != remoteFields->end() ) {
      rfP = static_cast<RemoteField*>(*i);
      if ( field_id == rfP->fieldId->getValue() ) {
         break;
      } else {
         rfP = NULL;
      }
      i++;
   }
   if ( rfP ) {
      // Tell the appropriate field to read what it wants from the packet.
      rfP->readField(); // Does readField + stats.
   } else {
      char tmp[120];
      sprintf(tmp, "Packet ID does not match any known field: %d ..ignore",
            field_id);
      showInfo(tmp);
   }
}

// ----------------------------------------------------------------------------
void RemoteConnection::sendPacket() {
  // Called in sending thread if periodic send, otherwise graphics thread

  if (sim_latency) {
    Packet * store_packetP = new Packet(write_packet);
    write_packet.clear();
    sim_latency_buffer.push(store_packetP);

    if (sim_latency_buffer.size() > max_sim_latency_buf_size) {
      Packet * pkP = sim_latency_buffer.front();
      sim_latency_buffer.pop();
      sendThisPacket(*pkP);
      delete pkP;
    }
  }
  else {
    sendThisPacket(write_packet);
  }
}

// ----------------------------------------------------------------------------
void RemoteConnection::sendThisPacket(Packet & packet) {
  // Called in sending thread if periodic send, otherwise graphics thread

  if (!isShuttingDown()) {
     bool success = packet.send(static_cast<TCPSock*>(sockP));
     if (!success) {
       cout << "send failure" << endl;
       requestClose();
     }
  }
}

// ----------------------------------------------------------------------------
void RemoteConnection::writeFieldHeader(int fld_id, bool with_timestamp) {
  // Called in sending thread if periodic send, otherwise graphics thread

  if (use_seq_numbers) {
    // If not using TCP we need to send a sequence number.
    write_packet.setSeqNum( seq_number++ );
  }
  if (fld_id > MAX_ID) {
    throw H3D::RemoteFieldIDTooLarge(fld_id, "Remote Field IDs must be < 65535");
  }
  write_packet.setId( fld_id );
  if (with_timestamp) {
    // Send time for latency info
     writeDouble( Scene::time->getValue() );
  }
}

// ----------------------------------------------------------------------------
// readDouble:
//    Extract 64 bit double from buffer_stream and return it.
//
double RemoteConnection::readDouble() {
  // Called in receiving thread

  // use a union so that we can easily access the bytes within a double
  union {
    double d;
    char sz[8];
  } network_double;

  read(&network_double.sz[0], 8);

  if (little_endian) {
    // reverse the order
    union {
      double d;
      char sz[8];
    } host_double;
    for (int i = 0; i < 8; i++) {
      host_double.sz[7-i] = network_double.sz[i];
    }
    return host_double.d;
  }

  return network_double.d;
}

// ----------------------------------------------------------------------------
void RemoteConnection::writeDouble(double const & val) {
  // Called in sending thread if periodic send, otherwise graphics thread

  if (!hasSock()) {
    return;
  }

  // use a union so that we can easily access the bytes within a double
  union {
    double d;
    char sz[8];
  } enc_double;

  enc_double.d = val;

  if (little_endian) {
    // reverse the order
    char network_double[8];
    for (int i = 0; i < 8; i++) {
      network_double[7-i] = enc_double.sz[i];
    }
    // send 8 bytes
    write(&network_double[0], 8);
  }
  else {
    // send 8 bytes
    write(&enc_double.sz[0], 8);
  }
}

// ----------------------------------------------------------------------------
void RemoteConnection::writeFloat(float const & val) {
  // use a union so that we can easily access the bytes within a float
  union {
    float f;
    char sz[4];
  } enc_float;


  // Convert to network format
  #ifdef WIN32
  union {
    int32 i;
    float32 f;
  } int_float;

  // I need this because i get a strange crash using the netwrap htonf
  // This code does the same thing.
  int_float.f = val;
  int_float.i = htonl(int_float.i);
  enc_float.f = int_float.f;
  #else
  enc_float.f = htonf(val);
  #endif

  // send 4 bytes
  if (hasSock()) {
    write(&enc_float.sz[0], 4);
  }
}

// ----------------------------------------------------------------------------
// readFloat:
//    Extract 32 bit float from buffer_stream and return it.
//
float RemoteConnection::readFloat() {
  // use a union so that we can easily access the bytes within a float
  union {
    float f;
    char sz[4];
  } decoded_float;

  // read 4 bytes
  read(&decoded_float.sz[0], 4);

  return ntohf(decoded_float.f);
}


// ----------------------------------------------------------------------------
void RemoteConnection::sendLoop() {
   // Only called (in sending thread) if periodic send is true

   // Note this only gets called if periodicSend is true.
   // By default, this class has periodic send false - If periodic send is set
   // to true, it is important to think of where the data is coming from.  If it is
   // coming from the graphic loop, it may not be available at a rate faster than
   // 30 hz so there is no point sending at a rate faster than this.

   setSendLoopRunning(true);
   while (!isShuttingDown()) {
   ///while (true) {
      if (!is_connected.get()) {
         #ifdef _WIN32
         Sleep(1000);  // Win32 Sleep is millisleep
         #else
         sleep(1);
         #endif
      } else {
         H3DTime t0 = Scene::time->getValue();
         double time_since_last_try = (double)(t0 - last_try_time);
         #ifdef _WIN32
         int time_to_kill = (int)(1000 * (period - time_since_last_try));
         if (time_to_kill > 0) {
            Sleep(time_to_kill);
         } else {
            Sleep(0); // Give other threads a go.
         }
         #else
         int time_to_kill = (int)(1000000 * (period - time_since_last_try));
         if (time_to_kill > 0) {
            usleep(time_to_kill);
         }else {
            usleep(0); // Give other threads a go.
         }
         #endif
         sendFields(false);
         last_send_time = t0;
      }
   }
   setSendLoopRunning(false);
}

// ----------------------------------------------------------------------------
void RemoteConnection::sendFields(bool first_connection) {
   // Called in sending thread if periodic send, otherwise in graphics thread
   // after the first connection.

   for (unsigned int i = 0; i < remoteFields->size() ; i++) {
      RemoteField * rfP = remoteFields->getValueByIndex(i);
      // We get here if we are doing periodic sending, or the first time we
      // connect with non-periodic sending. In the latter case, we may send some
      // fields if they want to be sent on connection.
      if (rfP->is_enabled.get()) {
         if (!first_connection || rfP->sendOnceOnConnect->getValue()) {
           lockWriteThread(); // Lock to avoid clash with reading thread which
           // sometime needs to write.
           rfP->writeField();
           sendPacket();
           unlockWriteThread();
         }
      } 
   }
}

// ----------------------------------------------------------------------------
void RemoteConnection:: setSendTimeout(InetSock * sockP, int millisecs) {
   // Set the socket to time out if it has a problem sending data.
   struct timeval tv;
   tv.tv_sec = 0;
   tv.tv_usec = millisecs * 1000;
   setsockopt((int)(sockP->sysRef()),
              SOL_SOCKET, SO_SNDTIMEO, (char*)(&tv),
              sizeof(tv));
}

//
// Collider::collide performs collisions by colliding with the
// children in turn.
//
// ----------------------------------------------------------------------------
void RemoteConnection::traverseSG( H3D::TraverseInfo & ti ) {
   H3D::Group::traverseSG(ti);
   // Do any actions required periodically at graphics loop rate.
   doGraphicsLoopActions();

   // call traverseSG on the remote fields.
   // The number of remote fields could change as a result of a traverseSG call
   // on one of the fields. Therefore we grab the number first, as an iterator
   // gets confused if we change the number during iteration.
   u_int num_flds = remoteFields->size();
   for (u_int i = 0; i < num_flds; i++) {
      RemoteField * fldP = remoteFields->getValueByIndex(i);
      fldP->traverseSG( ti );
   }
   if (remoteSync->getValue()) {
      remoteSync->getValue()->traverseSG(ti);
   }
}

// ----------------------------------------------------------------------------
void RemoteConnection::MFRemoteField::onAdd(Node * nodeP) {
   RemoteField * fldP = static_cast<RemoteField*>(nodeP);
   fldP->conP = conP;
   if (conP->periodicSend->getValue()) {
      fldP->setSendOnChange(false);
   } else {
      fldP->setSendOnChange(true);
   }
   H3D::TypedMFNode< RemoteField >::onAdd(nodeP);
}

// ----------------------------------------------------------------------------
void RemoteConnection::setShuttingDown(bool v) {
   connected_sem.signal(); // In case any thread is waiting on this semaphore.
   shutdown_lock.lock();
   shutting_down = v;
   setConnected(false);
   shutdown_lock.unlock();
}

// ----------------------------------------------------------------------------
bool RemoteConnection::isShuttingDown() {
   shutdown_lock.lock();
   bool v = shutting_down; 
   shutdown_lock.unlock();
   return v;
}

// ----------------------------------------------------------------------------
void RemoteConnection::setReceiveLoopRunning(bool v) {
   receive_loop_lock.lock();
   receive_loop_running = v;
   receive_loop_lock.unlock();
}

// ----------------------------------------------------------------------------
bool RemoteConnection::receiveLoopRunning() {
   receive_loop_lock.lock();
   bool v = receive_loop_running;
   receive_loop_lock.unlock();
   return v;
}

// ----------------------------------------------------------------------------
void RemoteConnection::setSendLoopRunning(bool v) {
   send_loop_lock.lock();
   send_loop_running = v;
   send_loop_lock.unlock();
}

// ----------------------------------------------------------------------------
bool RemoteConnection::sendLoopRunning() {
   send_loop_lock.lock();
   bool v = send_loop_running;
   send_loop_lock.unlock();
   return v;
}
