source: orxonox.OLD/branches/network/src/lib/network/network_stream.cc @ 7571

/*
   orxonox - the future of 3D-vertical-scrollers

   Copyright (C) 2004 orx

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.

### File Specific:
   main-programmer: claudio
   co-programmer:
*/


/* this is for debug output. It just says, that all calls to PRINT() belong to the DEBUG_MODULE_NETWORK module
   For more information refere to https://www.orxonox.net/cgi-bin/trac.cgi/wiki/DebugOutput
*/
#define DEBUG_MODULE_NETWORK


#include "base_object.h"
#include "network_protocol.h"
#include "udp_socket.h"
#include "udp_server_socket.h"
#include "connection_monitor.h"
#include "synchronizeable.h"
#include "network_game_manager.h"
#include "shared_network_data.h"

#include "lib/util/loading/factory.h"

#include "debug.h"
#include "class_list.h"
#include <algorithm>

/* include your own header */
#include "network_stream.h"

/* probably unnecessary */
using namespace std;


#define PACKAGE_SIZE  256


NetworkStream::NetworkStream()
    : DataStream()
{
  this->init();
  /* initialize the references */
  this->type = NET_CLIENT;
  this->networkProtocol = new NetworkProtocol();
  this->connectionMonitor = new ConnectionMonitor();
}


NetworkStream::NetworkStream( std::string host, int port )
{
  this->type = NET_CLIENT;
  this->init();
  this->peers[0].socket = new UdpSocket( host, port );
  this->networkProtocol = new NetworkProtocol();
  this->connectionMonitor = new ConnectionMonitor();
  this->maxConnections = MAX_CONNECTIONS;
}


NetworkStream::NetworkStream( int port )
{
  this->type = NET_SERVER;
  this->init();
  this->serverSocket = new UdpServerSocket(port);
  this->networkProtocol = new NetworkProtocol();
  this->connectionMonitor = new ConnectionMonitor();
  this->bActive = true;
}


void NetworkStream::init()
{
  /* set the class id for the base object */
  this->setClassID(CL_NETWORK_STREAM, "NetworkStream");
  this->bActive = false;
  this->serverSocket = NULL;
  this->networkGameManager = NULL;
  myHostId = 0;
  currentState = 0;
}


NetworkStream::~NetworkStream()
{
  if ( this->serverSocket )
  {
    serverSocket->close();
    delete serverSocket;
  }

  for ( PeerList::iterator i = peers.begin(); i!=peers.end(); i++)
  {
    if ( i->second.socket )
    {
      i->second.socket->disconnectServer();
      delete i->second.socket;
      i->second.socket = NULL;
    }
    
    if ( i->second.handshake )
    {
      delete i->second.handshake;
      i->second.handshake = NULL;
    }
  }
  
  if ( serverSocket )
  {
    delete serverSocket;
    serverSocket = NULL;
  }

  delete connectionMonitor;
  delete networkProtocol;
}


void NetworkStream::createNetworkGameManager()
{
  this->networkGameManager = NetworkGameManager::getInstance();
  // setUniqueID( maxCon+2 ) because we need one id for every handshake
  // and one for handshake to reject client maxCon+1
  this->networkGameManager->setUniqueID( this->maxConnections + 2 );

}


void NetworkStream::startHandshake()
{
  Handshake* hs = new Handshake(false);
  hs->setUniqueID( 0 );
  assert( peers[0].handshake == NULL );
  peers[0].handshake = hs;
  //this->connectSynchronizeable(*hs);
  PRINTF(0)("NetworkStream: %s\n", hs->getName());
}


void NetworkStream::connectSynchronizeable(Synchronizeable& sync)
{
  this->synchronizeables.push_back(&sync);
  sync.setNetworkStream( this );

  this->bActive = true;
}


void NetworkStream::disconnectSynchronizeable(Synchronizeable& sync)
{
  // removing the Synchronizeable from the List.
  std::list<Synchronizeable*>::iterator disconnectSynchro = std::find(this->synchronizeables.begin(), this->synchronizeables.end(), &sync);
  if (disconnectSynchro != this->synchronizeables.end())
    this->synchronizeables.erase(disconnectSynchro);

  this->bActive = false;
}


void NetworkStream::processData()
{
  currentState++;
  
  if ( this->type == NET_SERVER )
  {
    if ( serverSocket )
      serverSocket->update();
    
    this->updateConnectionList();
  }
  else
  {
    if ( peers[0].socket && !peers[0].socket->isOk() )
    {
      PRINTF(1)("lost connection to server\n");

      peers[0].socket->disconnectServer();
      delete peers[0].socket;
      peers[0].socket = NULL;

      if ( peers[0].handshake )
        delete peers[0].handshake;
      peers[0].handshake = NULL;
    }
  }




  /* DOWNSTREAM */
#if 0


  int dataLength;
  int reciever;
  Header header;
  int counter;

  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
  {
    counter = 0;

    if ( (*it)!=NULL && (*it)->beSynchronized() /*&& (*it)->getOwner() == myHostId*/ )
    {
      do {
        counter++;

        reciever = 0;
#warning fix this
dataLength = 0;
//TODO fix
        //dataLength = (*it)->readBytes(downBuffer, DATA_STREAM_BUFFER_SIZE, &reciever);

        if ( dataLength<=0 ){
          reciever = 0;
          continue;
        }

        dataLength = networkProtocol->createHeader((byte*)downBuffer, dataLength, DATA_STREAM_BUFFER_SIZE, static_cast<const Synchronizeable&>(*(*it)));

        Header* header = (Header*)downBuffer;
        if ( header->synchronizeableID < this->maxConnections+2 )
        {
          //if ( !isServer() ) PRINTF(0)("RESET UNIQUEID FROM %d TO 0 maxCon=%d\n", header->synchronizeableID, this->maxConnections);
          header->synchronizeableID = 0;
        }
        else
        {
          //if ( !isServer() ) PRINTF(0)("UNIQUEID=%d\n", header->synchronizeableID);
        }

        if ( dataLength<=0 )
          continue;

        if ( reciever!=0 )
        {
          if ( reciever < 0)
          {
            for ( int i = 0; i<networkSockets.size(); i++)
            {
              if ( i!=abs(reciever) && networkSockets[i] != NULL )
              {
                PRINTF(0)("write %d bytes to socket %d uniqueid %d reciever %d\n", dataLength, i, (*it)->getUniqueID(), reciever);
                networkSockets[i]->writePacket(downBuffer, dataLength);
              }
            }
          }
          else
          {
            if ( networkSockets[reciever] != NULL )
            {
              PRINTF(5)("write %d bytes to socket %d\n", dataLength, reciever);
              networkSockets[reciever]->writePacket(downBuffer, dataLength);
            }
            else
            {
              PRINTF(1)("networkSockets[reciever] == NULL\n");
            }
          }
        }
        else
        {
          for ( int i = 0; i<networkSockets.size(); i++)
          {
            if ( networkSockets[i] != NULL )
            {
              PRINTF(5)("write %d bytes to socket %d\n", dataLength, i);
              networkSockets[i]->writePacket(downBuffer, dataLength);
            }
          }
        }

      } while( reciever!=0 );
    }
  }

  /* UPSTREAM */

  for ( int i = 0; i<networkSockets.size(); i++)
  {
    if ( networkSockets[i] )
    {
      do {
        dataLength = networkSockets[i]->readPacket(upBuffer, DATA_STREAM_BUFFER_SIZE);

        if ( dataLength<=0 )
          continue;

        header = networkProtocol->extractHeader(upBuffer, dataLength);
        dataLength -= sizeof(header);

        PRINTF(5)("read %d bytes from socket uniqueID = %d\n", dataLength, header.synchronizeableID);

        if ( dataLength != header.length )
        {
          PRINTF(1)("packetsize in header and real packetsize do not match! %d:%d\n", dataLength, header.length);
          continue;
        }

        if ( header.synchronizeableID == 0 )
        {
          header.synchronizeableID = i;
        }

        for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
        {
#warning fix this

          if ( *it && (*it)->getUniqueID()==header.synchronizeableID )
          {
            if ( (*it)->writeBytes(upBuffer+sizeof(header), dataLength, i) != header.length )
            {
              PRINTF(1)("%s did not read all the data id = %d!\n", (*it)->getClassName(), (*it)->getUniqueID());
              break;
            }
            continue;
          }

        }

      } while ( dataLength>0 );
    }

  }
#endif
}

void NetworkStream::updateConnectionList( )
{
  //check for new connections

  NetworkSocket* tempNetworkSocket = serverSocket->getNewSocket();

  if ( tempNetworkSocket )
  {
    int clientId;
    if ( freeSocketSlots.size() >0 )
    {
      clientId = freeSocketSlots.back();
      freeSocketSlots.pop_back();
      peers[clientId].socket = tempNetworkSocket;
      peers[clientId].handshake = new Handshake(true, clientId, this->networkGameManager->getUniqueID());
      peers[clientId].handshake->setUniqueID(clientId);
      peers[clientId].userId = clientId;
    } else
    {
      clientId = 0;
      
      for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
        if ( it->first >= clientId )
          clientId = it->first + 1;
      
      peers[clientId].socket = tempNetworkSocket;
      peers[clientId].handshake = new Handshake(true, clientId, this->networkGameManager->getUniqueID());
      peers[clientId].handshake->setUniqueID(clientId);
      peers[clientId].userId = clientId;
    }

    if ( clientId > this->maxConnections )
    {
      peers[clientId].handshake->doReject( "too many connections" );
      PRINTF(0)("Will reject client %d because there are to many connections!\n", clientId);
    }
    else

    PRINTF(0)("New Client: %d\n", clientId);

    //this->connectSynchronizeable(*handshakes[clientId]);
  }


  //check if connections are ok else remove them
  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
  {
    if ( it->second.socket && !it->second.socket->isOk() )
    {
      PRINTF(0)("Client is gone: %d\n", it->second.userId);

      it->second.socket->disconnectServer();
      delete it->second.socket;
      it->second.socket = NULL;

      if ( it->second.handshake )
        delete it->second.handshake;
      it->second.handshake = NULL;


      NetworkGameManager::getInstance()->signalLeftPlayer(it->second.userId);

      freeSocketSlots.push_back( it->second.userId );

    }
  }


}

void NetworkStream::debug()
{
  if( this->isServer())
    PRINT(0)(" Host ist Server with ID: %i\n", this->myHostId);
  else
    PRINT(0)(" Host ist Client with ID: %i\n", this->myHostId);

  PRINT(0)(" Got %i connected Synchronizeables, showing active Syncs:\n", this->synchronizeables.size());
  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
  {
    if( (*it)->beSynchronized() == true)
      PRINT(0)("  Synchronizeable of class: %s::%s, with unique ID: %i, Synchronize: %i\n", (*it)->getClassName(), (*it)->getName(),
               (*it)->getUniqueID(), (*it)->beSynchronized());
  }
  PRINT(0)(" Maximal Connections: %i\n", this->maxConnections);

}


int NetworkStream::getSyncCount()
{
  int n = 0;
  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
    if( (*it)->beSynchronized() == true)
      ++n;

  //return synchronizeables.size();
  return n;
}

/**
 * check if handshakes completed
 */
void NetworkStream::handleHandshakes( )
{
  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
  {
    if ( it->second.handshake )
    {
      if ( it->second.handshake->completed() )
      {
        if ( it->second.handshake->ok() )
        {
          if ( type != NET_SERVER )
          {
            SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
            myHostId = SharedNetworkData::getInstance()->getHostID();

            this->networkGameManager = NetworkGameManager::getInstance();
            this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
          }

          PRINT(0)("handshake finished id=%d\n", it->second.handshake->getNetworkGameManagerId());


          delete it->second.handshake;
          it->second.handshake = NULL;
        }
        else
        {
          PRINT(1)("handshake failed!\n");
          it->second.socket->disconnectServer();
        }
      }
    }
  }
}

/**
 * handle upstream network traffic
 */
void NetworkStream::handleUpstream( )
{
  byte buf[UDP_PACKET_SIZE];
  int offset;
  int n;
  
  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
  {
    offset = INTSIZE; //make already space for length 
    
    if ( peer->second.socket )
      continue;
    
    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;
    
    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;
    
    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;
    
    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
    {
      Synchronizeable & sync = **it;
      
      assert( offset + INTSIZE <= UDP_PACKET_SIZE );
      
      n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );
      assert( n == INTSIZE );
      offset += n;
      
      offset += sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, 0 );
    }
    
    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );
    
    assert( peer->second.socket->writePacket( buf, offset ) );
  }
}

/**
 * handle downstream network traffic
 */
void NetworkStream::handleDownstream( )
{
  byte buf[UDP_PACKET_SIZE];
  int offset = 0;
  
  int length = 0;
  int packetLength = 0;
  int uniqueId = 0;
  int state = 0;
  int ackedState = 0;
  int fromState = 0;
  
  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
  {
    packetLength = peer->second.socket->readPacket( buf, UDP_PACKET_SIZE );
    
    assert( Converter::byteArrayToInt( buf, &length ) == INTSIZE );
    assert( Converter::byteArrayToInt( buf + INTSIZE, &state ) == INTSIZE );
    assert( Converter::byteArrayToInt( buf + 2*INTSIZE, &fromState ) == INTSIZE );
    assert( Converter::byteArrayToInt( buf + 3*INTSIZE, &ackedState ) == INTSIZE );
    
    //if this is an old state drop it
    if ( state <= peer->second.lastRecvedState )
      continue;
    
    if ( packetLength != length )
    {
      PRINTF(1)("real packet length (%d) and transmitted packet length (%d) do not match!\n", packetLength, length);
      peer->second.socket->disconnectServer();
      continue;
    }
    
    offset = 4*INTSIZE;
    
    while ( offset < length )
    {
      assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
      offset += INTSIZE;
      
      Synchronizeable * sync = NULL;
      
      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
      {
        if ( (*it)->getUniqueID() == uniqueId )
        {
          sync = *it;
          break;
        }
      }
      
      if ( sync == NULL )
      {
        //TODO dont accept new object from all peers (probably only servers)
        int leafClassId;
        if ( INTSIZE > length - offset )
          break;
        
        Converter::byteArrayToInt( buf + offset, &leafClassId );
        
        BaseObject * b;
        /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
        /* Exception 1: NullParent */
        if( leafClassId == CL_NULL_PARENT)
        {
          PRINTF(1)("Can not create Class with ID %x!", (int)leafClassId);
          break;
        }
        else
          b = Factory::fabricate( (ClassID)leafClassId );

        if ( !b )
        {
          PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
          break;
        }
        
        if ( b->isA(CL_SYNCHRONIZEABLE) )
        {
          sync = dynamic_cast<Synchronizeable*>(b);
          sync->setUniqueID( uniqueId );
          sync->setSynchronized(true);
  
          PRINTF(0)("Fabricated %s with id %d\n", sync->getClassName(), sync->getUniqueID());
        }
        else
        {
          PRINTF(1)("Class with ID %x is not a synchronizeable!", (int)leafClassId);
          delete b;
          break;
        }
      }
      
      offset += sync->setStateDiff( peer->second.userId, buf+offset, length-offset, state, fromState );
    }
    
    if ( offset != length )
    {
      peer->second.socket->disconnectServer();
    }
    
    peer->second.lastAckedState = ackedState;
  }
}