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

/*
   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: Christoph Renner rennerc@ee.ethz.ch
   co-programmer:   Patrick Boenzli  boenzlip@orxonox.ethz.ch

     June 2006: finishing work on the network stream for pps presentation (rennerc@ee.ethz.ch)
     July 2006: some code rearangement and integration of the proxy server mechanism (boenzlip@ee.ethz.ch)
*/


#define DEBUG_MODULE_NETWORK


#include "base_object.h"
#include "network_protocol.h"
#include "udp_socket.h"
#include "udp_server_socket.h"
#include "monitor/connection_monitor.h"
#include "monitor/network_monitor.h"
#include "synchronizeable.h"
#include "ip.h"
#include "network_game_manager.h"
#include "shared_network_data.h"
#include "message_manager.h"
#include "preferences.h"
#include "zip.h"

#include "src/lib/util/loading/resource_manager.h"

#include "network_log.h"

#include "player_stats.h"

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

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


#include "network_stream.h"


#include "converter.h"


#define PACKAGE_SIZE  256


/**
 * empty constructor
 */
NetworkStream::NetworkStream()
    : DataStream()
{
  this->init();
  /* initialize the references */
  this->pInfo->nodeType = NET_CLIENT;
}


NetworkStream::NetworkStream( int nodeType)
{
  this->init();

  this->pInfo->nodeType = nodeType;

  switch( nodeType)
  {
    case NET_MASTER_SERVER:
      // init the shared network data
      SharedNetworkData::getInstance()->setHostID(NET_ID_MASTER_SERVER);
      break;

    case NET_PROXY_SERVER_ACTIVE:
      // init the shared network data
      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
      break;
    case NET_PROXY_SERVER_PASSIVE:
      // init the shared network data
      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
      break;
    case NET_CLIENT:
      SharedNetworkData::getInstance()->setHostID(NET_ID_UNASSIGNED);
      break;
  }

  SharedNetworkData::getInstance()->setDefaultSyncStream(this);

  // get the local ip address
  IPaddress ip;
  SDLNet_ResolveHost( &ip, NULL, 0);
  this->pInfo->ip = ip;
}



/**
 * generic init functions
 */
void NetworkStream::init()
{
  /* set the class id for the base object */
  this->setClassID(CL_NETWORK_STREAM, "NetworkStream");
  this->serverSocket = NULL;
  this->networkGameManager = NULL;
  this->networkMonitor = NULL;

  this->pInfo = new PeerInfo();
  this->pInfo->userId = 0;
  this->pInfo->lastAckedState = 0;
  this->pInfo->lastRecvedState = 0;

  this->bRedirect = false;

  this->currentState = 0;

  remainingBytesToWriteToDict = Preferences::getInstance()->getInt( "compression", "writedict", 0 );

  assert( Zip::getInstance()->loadDictionary( "testdict" ) >= 0 );
  this->dictClient = Zip::getInstance()->loadDictionary( "dict2pl_client" );
  assert( this->dictClient >= 0 );
  this->dictServer = Zip::getInstance()->loadDictionary( "dict2p_server" );
  assert( this->dictServer >= 0 );
}


/**
 * deconstructor
 */
NetworkStream::~NetworkStream()
{
  if ( this->serverSocket )
  {
    serverSocket->close();
    delete serverSocket;
    serverSocket = NULL;
  }
  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 ( i->second.connectionMonitor )
    {
      delete i->second.connectionMonitor;
      i->second.connectionMonitor = NULL;
    }
  }
  for ( SynchronizeableList::const_iterator it = getSyncBegin(); it != getSyncEnd(); it ++ )
    (*it)->setNetworkStream( NULL );

  if( this->pInfo)
    delete this->pInfo;

  if( this->networkMonitor)
    delete this->networkMonitor;
}


/**
 * establish a connection to a remote master server
 * @param host: host name
 * @param port: the port number
 */
void NetworkStream::connectToMasterServer(std::string host, int port)
{
  int node = NET_ID_MASTER_SERVER;
  // this create the new node in the peers map
  this->peers[node].socket = new UdpSocket( host, port );
  this->peers[node].userId = NET_ID_MASTER_SERVER;

  this->peers[node].nodeType = NET_MASTER_SERVER;
  this->peers[node].connectionMonitor = new ConnectionMonitor( NET_ID_MASTER_SERVER );
  this->peers[node].ip = this->peers[node].socket->getRemoteAddress();
}


/**
 * establish a connection to a remote proxy server
 * @param host: host name
 * @param port: the port number
 */
void NetworkStream::connectToProxyServer(int proxyId,std::string host, int port)
{
  PRINTF(0)("connect to proxy %s, this is proxyId %i\n", host.c_str(), proxyId);

  // this creates the new proxyId in the peers map
  this->peers[proxyId].socket = new UdpSocket( host, port );
  this->peers[proxyId].userId = proxyId;

  this->peers[proxyId].nodeType = NET_PROXY_SERVER_ACTIVE;
  this->peers[proxyId].connectionMonitor = new ConnectionMonitor( proxyId );
  this->peers[proxyId].ip = this->peers[proxyId].socket->getRemoteAddress();
}


/**
 * create a server
 * @param port: interface port for all clients
 */
void NetworkStream::createServer(int port)
{
  this->serverSocket = new UdpServerSocket(port);
}


/**
 * creates a new instance of the network game manager
 */
void NetworkStream::createNetworkGameManager()
{
  this->networkGameManager = NetworkGameManager::getInstance();

  this->networkGameManager->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
  MessageManager::getInstance()->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
}


/**
 * starts the network handshake
 * handsakes are always initialized from the client side first. this starts the handshake and therefore is only
 * executed as client
 * @param userId: start handshake for this user id (optional, default == 0)
 */
void NetworkStream::startHandshake(int userId)
{
  Handshake* hs = new Handshake(this->pInfo->nodeType);
  // fake the unique id
  hs->setUniqueID( NET_UID_HANDSHAKE );
  assert( peers[userId].handshake == NULL );
  peers[userId].handshake = hs;

  // set the preferred nick name
  hs->setPreferedNickName( Preferences::getInstance()->getString( "multiplayer", "nickname", "Player" ) );

  PRINTF(0)("NetworkStream: Handshake created: %s\n", hs->getCName());
}


/**
 * this functions connects a synchronizeable to the networkstream, therefore synchronizeing
 * it all over the network and creating it on the other platforms (if and only if it is a
 * server
 * @param sync: the synchronizeable to add
 */
void NetworkStream::connectSynchronizeable(Synchronizeable& sync)
{
  this->synchronizeables.push_back(&sync);
  sync.setNetworkStream( this );
}


/**
 * removes the synchronizeable from the list of synchronized entities
 * @param sync: the syncronizeable to remove
 */
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);

  oldSynchronizeables[sync.getUniqueID()] = SDL_GetTicks();
}


/**
 * this is called to process data from the network socket to the synchronizeable and vice versa
 */
void NetworkStream::processData()
{
  // create the network monitor after all the init work and before there is any connection handlings
  if( this->networkMonitor == NULL)
    this->networkMonitor = new NetworkMonitor(this);


  int tick = SDL_GetTicks();

  this->currentState++;
  // there was a wrap around
  if( this->currentState < 0)
  {
    PRINTF(1)("A wrap around in the state variable as occured. The server was running so long? Pls restart server or write a mail to the supporters!\n");
  }

  if ( this->pInfo->isMasterServer())
  {
    // execute everytthing the master server shoudl do
    if ( serverSocket )
      serverSocket->update();

    this->updateConnectionList();
  }
  else if( this->pInfo->isProxyServerActive())
  {
    // execute everything the proxy server should do
    if ( serverSocket )
      serverSocket->update();

    this->updateConnectionList();
  }
  else
  {
    // check if the connection is ok else terminate and remove
#warning make this more modular: every proxy/master server connection should be watched for termination
    if ( !peers.empty() && peers[NET_ID_MASTER_SERVER].socket &&
          ( !peers[NET_ID_MASTER_SERVER].socket->isOk() ||
          peers[NET_ID_MASTER_SERVER].connectionMonitor->hasTimedOut() ) )
    {
      this->handleDisconnect( NET_ID_MASTER_SERVER);
      PRINTF(1)("lost connection to server\n");
    }
    // check if there is a redirection command
    if( this->bRedirect)
    {
      this->handleReconnect( NET_ID_MASTER_SERVER);
    }
  }

  this->cleanUpOldSyncList();
  this->handleHandshakes();

  // update the network monitor
  this->networkMonitor->process();

  // order of up/downstream is important!!!!
  // don't change it
  this->handleDownstream( tick );
  this->handleUpstream( tick );
}


/**
 * if we are a NET_MASTER_SERVER or NET_PROXY_SERVER_ACTIVE update the connection list to accept new
 * connections (clients) also start the handsake for the new clients
 */
void NetworkStream::updateConnectionList( )
{
  //check for new connections

  NetworkSocket* tempNetworkSocket = serverSocket->getNewSocket();

  // we got new network node
  if ( tempNetworkSocket )
  {
    int clientId;
    // determine the network node id
    if ( freeSocketSlots.size() > 0 )
    {
      clientId = freeSocketSlots.back();
      freeSocketSlots.pop_back();
    }
    else
    {
      clientId = 1;

      for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
        if ( it->first >= clientId )
          clientId = it->first + 1;
    }
    // this creates a new entry in the peers list
    peers[clientId].socket = tempNetworkSocket;


    // create new handshake and init its variables
    peers[clientId].handshake = new Handshake(this->pInfo->nodeType, clientId, this->networkGameManager->getUniqueID(), MessageManager::getInstance()->getUniqueID());
    peers[clientId].handshake->setUniqueID(clientId);

    peers[clientId].connectionMonitor = new ConnectionMonitor( clientId );
    peers[clientId].userId = clientId;

    PRINTF(0)("num sync: %d\n", synchronizeables.size());

    // get the proxy server informations and write them to the handshake, if any (proxy)
    assert( this->networkMonitor != NULL);
    PeerInfo* pi = this->networkMonitor->getFirstChoiceProxy();
    if( pi != NULL)
    {
      peers[clientId].handshake->setProxy1Address( pi->ip);
    }
    pi = this->networkMonitor->getSecondChoiceProxy();
    if( pi != NULL)
      peers[clientId].handshake->setProxy2Address( pi->ip);

    // check if the connecting client should reconnect to a proxy server
    peers[clientId].handshake->setRedirect(this->networkMonitor->isReconnectNextClient());

    // the connecting node of course is a client
    peers[clientId].nodeType = NET_CLIENT;
    peers[clientId].ip = peers[clientId].socket->getRemoteAddress();


    // check if there are too many clients connected (DEPRECATED: new: the masterserver sends a list of proxy servers)
//     if ( clientId > SharedNetworkData::getInstance()->getMaxPlayer() )
//     {
// //       peers[clientId].handshake->setRedirect(true);
// //
// //       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);
//     }
    PRINTF(0)("New Client: %d\n", clientId);


  }



  //check if connections are ok else remove them
  for ( PeerList::iterator it = peers.begin(); it != peers.end(); )
  {
    if (
          it->second.socket &&
          (
            !it->second.socket->isOk()  ||
            it->second.connectionMonitor->hasTimedOut()
          )
       )
    {
      std::string reason = "disconnected";
      if ( it->second.connectionMonitor->hasTimedOut() )
        reason = "timeout";
      PRINTF(0)("Client is gone: %d (%s)\n", it->second.userId, reason.c_str());

      this->handleDisconnect( it->second.userId);

      it++;
      continue;
    }

    it++;
  }


}


void NetworkStream::debug()
{
  if( SharedNetworkData::getInstance()->isMasterServer()) {
    PRINT(0)(" Host ist Master Server with ID: %i\n", this->pInfo->userId);
  }
  else if( SharedNetworkData::getInstance()->isProxyServerActive()) {
    PRINT(0)(" Host ist Proxy Server with ID: %i\n", this->pInfo->userId);
  }
  else {
    PRINT(0)(" Host ist Client with ID: %i\n", this->pInfo->userId);
  }

  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)->getClassCName(), (*it)->getCName(),
               (*it)->getUniqueID(), (*it)->beSynchronized());
  }
  PRINT(0)(" Maximal Connections: %i\n", SharedNetworkData::getInstance()->getMaxPlayer() );

}


/**
 * @returns the number of synchronizeables registered to this stream
 */
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. if so create the network game manager else remove it again
 */
void NetworkStream::handleHandshakes( )
{
  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
  {
    if ( it->second.handshake )
    {
      // handshake finished
      if ( it->second.handshake->completed() )
      {
        //handshake is correct
        if ( it->second.handshake->ok() )
        {
          // the counter part didn't mark it free for deletion yet
          if ( !it->second.handshake->allowDel() )
          {
            // make sure this is a client
            if( this->pInfo->isClient())
            {
              SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
              this->pInfo->userId = SharedNetworkData::getInstance()->getHostID();

              it->second.nodeType = it->second.handshake->getRemoteNodeType();
              it->second.ip = it->second.socket->getRemoteAddress();
              // add the new server to the nodes list (it can be a NET_MASTER_SERVER or NET_PROXY_SERVER)
              this->networkMonitor->addNode(&it->second);
              // get proxy 1 address and add it
              this->networkMonitor->addNode(it->second.handshake->getProxy1Address(), NET_PROXY_SERVER_ACTIVE);
              // get proxy 2 address and add it
              this->networkMonitor->addNode(it->second.handshake->getProxy2Address(), NET_PROXY_SERVER_ACTIVE);

              // now check if the server accepted the connection
              if( it->second.handshake->redirect() )
              {
                this->bRedirect = true;
              }

              // create the new network game manager and init it
              this->networkGameManager = NetworkGameManager::getInstance();
              this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
              // init the new message manager
              MessageManager::getInstance()->setUniqueID( it->second.handshake->getMessageManagerId() );
            }

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

          }
          else
          {
            // handshake finished registring new player
            if ( it->second.handshake->canDel() )
            {

              if ( this->pInfo->isMasterServer() )
              {
                it->second.nodeType = it->second.handshake->getRemoteNodeType();
                it->second.ip = it->second.socket->getRemoteAddress();

                this->networkMonitor->addNode(&it->second);

                this->handleNewClient( it->second.userId );

                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
                {
                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
                }
              }
              else if ( this->pInfo->isProxyServerActive() )
              {
                it->second.nodeType = it->second.handshake->getRemoteNodeType();
                it->second.ip = it->second.socket->getRemoteAddress();

                this->networkMonitor->addNode(&it->second);

                this->handleNewClient( it->second.userId );

                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
                {
                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
                }
              }

              PRINT(0)("handshake finished delete it\n");
              delete it->second.handshake;
              it->second.handshake = NULL;
            }
          }

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


/**
 * this functions handles a reconnect event received from the a NET_MASTER_SERVER or NET_PROXY_SERVER
 */
void NetworkStream::handleReconnect(int userId)
{
  this->bRedirect = false;
  PeerInfo* pInfo = &this->peers[userId];

  PRINTF(0)("===============================================\n");
  PRINTF(0)("Client is redirected to the other proxy servers\n");
  PRINTF(0)("  user id: %i\n", userId);
  PRINTF(0)("  connecting to: %s\n", this->networkMonitor->getFirstChoiceProxy()->ip.ipString().c_str());
  PRINTF(0)("===============================================\n");

  // flush the old synchronization states, since the numbering could be completely different
  pInfo->lastAckedState = 0;
  pInfo->lastRecvedState = 0;

  // temp save the ip address here
  IP proxyIP = pInfo->handshake->getProxy1Address();

  // disconnect from the current server and reconnect to proxy server
  this->handleDisconnect( userId);
  this->connectToProxyServer(NET_ID_PROXY_SERVER_01, proxyIP.ipString(), 9999);
  #warning the ports are not yet integrated correctly in the ip class

  // and restart the handshake
  this->startHandshake( userId);
}


/**
 * handles the disconnect event
 * @param userId id of the user to remove
 */
void NetworkStream::handleDisconnect( int userId )
{
  peers[userId].socket->disconnectServer();
  delete peers[userId].socket;
  peers[userId].socket = NULL;

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

  if ( peers[userId].connectionMonitor )
    delete peers[userId].connectionMonitor;
  peers[userId].connectionMonitor = NULL;


  for ( SynchronizeableList::iterator it2 = synchronizeables.begin(); it2 != synchronizeables.end(); it2++ )  {
    (*it2)->cleanUpUser( userId );
  }

  if( SharedNetworkData::getInstance()->isMasterServer())
    NetworkGameManager::getInstance()->signalLeftPlayer(userId);

  this->freeSocketSlots.push_back( userId );

  peers.erase( userId);
}



/**
 * handle upstream network traffic
 * @param tick: seconds elapsed since last update
 */
void NetworkStream::handleUpstream( int tick )
{
  int offset;
  int n;

  for ( PeerList::reverse_iterator peer = peers.rbegin(); peer != peers.rend(); peer++ )
  {
    offset = INTSIZE; // reserve enough space for the packet length

    // continue with the next peer if this peer has no socket assigned (therefore no network)
    if ( !peer->second.socket )
      continue;

    // header informations: current state
    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;

    // header informations: last acked state
    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;

    // header informations: last recved state
    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
    assert( n == INTSIZE );
    offset += n;

    // now write all synchronizeables in the packet
    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
    {

      int oldOffset = offset;
      Synchronizeable & sync = **it;


      // do not include synchronizeables with uninit id and syncs that don't want to be synchronized
      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
        continue;

      // if handshake not finished only sync handshake
      if ( peer->second.handshake && sync.getLeafClassID() != CL_HANDSHAKE )
        continue;

      // if we are a server (both master and proxy servers) and this is not our handshake
      if ( ( SharedNetworkData::getInstance()->isMasterServer() || SharedNetworkData::getInstance()->isProxyServerActive() ) && sync.getLeafClassID() == CL_HANDSHAKE && sync.getUniqueID() != peer->second.userId )
        continue;

      /* list of synchronizeables that will never be synchronized over the network: */
      // do not sync null parent
      if ( sync.getLeafClassID() == CL_NULL_PARENT )
        continue;


      assert( sync.getLeafClassID() != 0);

      assert( offset + INTSIZE <= UDP_PACKET_SIZE );

      // server fakes uniqueid == 0 for handshake
      if ( ( SharedNetworkData::getInstance()->isMasterServer() || SharedNetworkData::getInstance()->isProxyServerActive() ) &&
             sync.getUniqueID() <= SharedNetworkData::getInstance()->getMaxPlayer() + 1) // plus one to handle one client more than the max to redirect it
        n = Converter::intToByteArray( 0, buf + offset, UDP_PACKET_SIZE - offset );
      else
        n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );


      assert( n == INTSIZE );
      offset += n;

      // make space for packet size
      offset += INTSIZE;

      n = sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, -1000 );
      offset += n;

      assert( Converter::intToByteArray( n, buf + offset - n - INTSIZE, INTSIZE ) == INTSIZE );

      // check if all data bytes == 0 -> remove data and the synchronizeable from the sync process since there is no update
      // TODO not all synchronizeables like this maybe add Synchronizeable::canRemoveZeroDiff()
      bool allZero = true;
      for ( int i = 0; i < n; i++ )
      {
         if ( buf[i+oldOffset+2*INTSIZE] != 0 )
           allZero = false;
      }
      // if there is no new data in this synchronizeable reset the data offset to the last state -> dont synchronizes
      // data that hast not changed
      if ( allZero )
      {
        offset = oldOffset;
      }
    } // all synchronizeables written



    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
    {
      Synchronizeable & sync = **it;

      // again exclude all unwanted syncs
      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED)
        continue;

      sync.handleSentState( peer->second.userId, currentState, peer->second.lastAckedState );
    }


    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );

    // now compress the data with the zip library
    int compLength = 0;
    if ( SharedNetworkData::getInstance()->isMasterServer() || SharedNetworkData::getInstance()->isProxyServerActive())
      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictServer );
    else
      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictClient );

    if ( compLength <= 0 )
    {
      PRINTF(1)("compression failed!\n");
      continue;
    }

    assert( peer->second.socket->writePacket( compBuf, compLength ) );

    if ( this->remainingBytesToWriteToDict > 0 )
      writeToNewDict( buf, offset, true );

    peer->second.connectionMonitor->processUnzippedOutgoingPacket( tick, buf, offset, currentState );
    peer->second.connectionMonitor->processZippedOutgoingPacket( tick, compBuf, compLength, currentState );

  }
}

/**
 * handle downstream network traffic
 */
void NetworkStream::handleDownstream( int tick )
{
  int offset = 0;

  int length = 0;
  int packetLength = 0;
  int compLength = 0;
  int uniqueId = 0;
  int state = 0;
  int ackedState = 0;
  int fromState = 0;
  int syncDataLength = 0;

  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
  {

    if ( !peer->second.socket )
      continue;

    while ( 0 < (compLength = peer->second.socket->readPacket( compBuf, UDP_PACKET_SIZE )) )
    {
      peer->second.connectionMonitor->processZippedIncomingPacket( tick, compBuf, compLength );

      packetLength = Zip::getInstance()->unZip( compBuf, compLength, buf, UDP_PACKET_SIZE );

      if ( packetLength < 4*INTSIZE )
      {
        if ( packetLength != 0 )
          PRINTF(1)("got too small packet: %d\n", packetLength);
        continue;
      }

      if ( this->remainingBytesToWriteToDict > 0 )
        writeToNewDict( buf, packetLength, false );

      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 );
      offset = 4*INTSIZE;

      peer->second.connectionMonitor->processUnzippedIncomingPacket( tick, buf, packetLength, state, ackedState );


      //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;
      }

      while ( offset + 2 * INTSIZE < length )
      {
        assert( offset > 0 );
        assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
        offset += INTSIZE;

        assert( Converter::byteArrayToInt( buf + offset, &syncDataLength ) == INTSIZE );
        offset += INTSIZE;

        assert( syncDataLength > 0 );
        assert( syncDataLength < 10000 );

        Synchronizeable * sync = NULL;

        // look for the synchronizeable in question
        for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
        {
          // client thinks his handshake has id 0!!!!!
          if ( (*it)->getUniqueID() == uniqueId || ( uniqueId == 0 && (*it)->getUniqueID() == peer->second.userId ) )
          {
            sync = *it;
            break;
          }
        }

        // this synchronizeable does not yet exist! create it
        if ( sync == NULL )
        {
          PRINTF(0)("could not find sync with id %d. try to create it\n", uniqueId);

          // if it is an old synchronizeable already removed, ignore it
          if ( oldSynchronizeables.find( uniqueId ) != oldSynchronizeables.end() )
          {
            offset += syncDataLength;
            continue;
          }

          // if the node we got this unknown sync from is a client we ignore it (since it has no rights to create a new sync)
          if ( peers[peer->second.userId].isClient() )
          {
            offset += syncDataLength;
            continue;
          }

          int leafClassId;
          if ( INTSIZE > length - offset )
          {
            offset += syncDataLength;
            continue;
          }

          Converter::byteArrayToInt( buf + offset, &leafClassId );

          assert( leafClassId != 0 );


          BaseObject * b = NULL;
          /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
          /* Exception 1: NullParent */
          if( leafClassId == CL_NULL_PARENT || leafClassId == CL_SYNCHRONIZEABLE || leafClassId == CL_NETWORK_GAME_MANAGER )
          {
            PRINTF(1)("Don't create Object with ID %x, ignored!\n", (int)leafClassId);
            offset += syncDataLength;
            continue;
          }
          else
            b = Factory::fabricate( (ClassID)leafClassId );

          if ( !b )
          {
            PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
            offset += syncDataLength;
            continue;
          }

          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->getClassCName(), sync->getUniqueID());
          }
          else
          {
            PRINTF(1)("Class with ID %x is not a synchronizeable!\n", (int)leafClassId);
            delete b;
            offset += syncDataLength;
            continue;
          }
        }


        int n = sync->setStateDiff( peer->second.userId, buf+offset, syncDataLength, state, fromState );
        offset += n;

      }

      if ( offset != length )
      {
        PRINTF(0)("offset (%d) != length (%d)\n", offset, length);
        peer->second.socket->disconnectServer();
      }


      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
      {
        Synchronizeable & sync = **it;

        if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
          continue;

        sync.handleRecvState( peer->second.userId, state, fromState );
      }

      assert( peer->second.lastAckedState <= ackedState );
      peer->second.lastAckedState = ackedState;

      assert( peer->second.lastRecvedState < state );
      peer->second.lastRecvedState = state;

    }

  }

}

/**
 * is executed when a handshake has finished
 */
void NetworkStream::handleNewClient( int userId )
{
  // init and assign the message manager
  MessageManager::getInstance()->initUser( userId );
  // do all game relevant stuff here
  networkGameManager->signalNewPlayer( userId );
}


/**
 * removes old items from oldSynchronizeables
 */
void NetworkStream::cleanUpOldSyncList( )
{
  int now = SDL_GetTicks();

  for ( std::map<int,int>::iterator it = oldSynchronizeables.begin(); it != oldSynchronizeables.end();  )
  {
    if ( it->second < now - 10*1000 )
    {
      std::map<int,int>::iterator delIt = it;
      it++;
      oldSynchronizeables.erase( delIt );
      continue;
    }
    it++;
  }
}

/**
 * writes data to DATA/dicts/newdict
 * @param data pointer to data
 * @param length length
 */
void NetworkStream::writeToNewDict( byte * data, int length, bool upstream )
{
  if ( remainingBytesToWriteToDict <= 0 )
    return;

  if ( length > remainingBytesToWriteToDict )
    length = remainingBytesToWriteToDict;

  std::string fileName = ResourceManager::getInstance()->getDataDir();
  fileName += "/dicts/newdict";

  if ( upstream )
    fileName += "_upstream";
  else
    fileName += "_downstream";

  FILE * f = fopen( fileName.c_str(), "a" );

  if ( !f )
  {
    PRINTF(2)("could not open %s\n", fileName.c_str());
    remainingBytesToWriteToDict = 0;
    return;
  }

  if ( fwrite( data, 1, length, f ) != length )
  {
    PRINTF(2)("could not write to file\n");
    fclose( f );
    return;
  }

  fclose( f );

  remainingBytesToWriteToDict -= length;
}