Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

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

Last change on this file since 9573 was 9571, checked in by patrick, 18 years ago

disconnecting instead of joining..

File size: 35.4 KB
Line 
1/*
2   orxonox - the future of 3D-vertical-scrollers
3
4   Copyright (C) 2004 orx
5
6   This program is free software; you can redistribute it and/or modify
7   it under the terms of the GNU General Public License as published by
8   the Free Software Foundation; either version 2, or (at your option)
9   any later version.
10
11### File Specific:
12   main-programmer: Christoph Renner rennerc@ee.ethz.ch
13   co-programmer:   Patrick Boenzli  patrick@orxonox.ethz.ch
14
15     June 2006: finishing work on the network stream for pps presentation (rennerc@ee.ethz.ch)
16     July 2006: some code rearangement and integration of the proxy server mechanism (patrick@orxonox.ethz.ch)
17*/
18
19
20#define DEBUG_MODULE_NETWORK
21
22#include "proxy/proxy_control.h"
23
24#include "base_object.h"
25#include "network_protocol.h"
26#include "udp_socket.h"
27#include "udp_server_socket.h"
28#include "monitor/connection_monitor.h"
29#include "monitor/network_monitor.h"
30#include "synchronizeable.h"
31#include "ip.h"
32#include "network_game_manager.h"
33#include "shared_network_data.h"
34#include "message_manager.h"
35#include "preferences.h"
36#include "zip.h"
37
38#include "src/lib/util/loading/resource_manager.h"
39
40#include "network_log.h"
41
42#include "player_stats.h"
43
44#include "lib/util/loading/factory.h"
45
46#include "debug.h"
47#include "class_list.h"
48#include <algorithm>
49
50
51#include "network_stream.h"
52
53
54#include "converter.h"
55
56
57#define PACKAGE_SIZE  256
58
59
60/**
61 * empty constructor
62 */
63NetworkStream::NetworkStream()
64    : DataStream()
65{
66  this->init();
67  /* initialize the references */
68  this->pInfo->nodeType = NET_UNASSIGNED;
69}
70
71
72NetworkStream::NetworkStream( int nodeType)
73{
74  this->init();
75
76  this->pInfo->nodeType = nodeType;
77
78  switch( nodeType)
79  {
80    case NET_MASTER_SERVER:
81      // init the shared network data
82      SharedNetworkData::getInstance()->setHostID(NET_ID_MASTER_SERVER);
83      break;
84    case NET_PROXY_SERVER_ACTIVE:
85      // init the shared network data
86      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
87      break;
88    case NET_PROXY_SERVER_PASSIVE:
89      // init the shared network data
90      SharedNetworkData::getInstance()->setHostID(NET_ID_PROXY_SERVER_01);
91      break;
92    case NET_CLIENT:
93      SharedNetworkData::getInstance()->setHostID(NET_ID_UNASSIGNED);
94      break;
95  }
96
97  SharedNetworkData::getInstance()->setDefaultSyncStream(this);
98
99  // get the local ip address
100  IPaddress ip;
101  SDLNet_ResolveHost( &ip, NULL, 0);
102  this->pInfo->ip = ip;
103}
104
105
106
107/**
108 * generic init functions
109 */
110void NetworkStream::init()
111{
112  /* set the class id for the base object */
113  this->setClassID(CL_NETWORK_STREAM, "NetworkStream");
114  this->clientSocket = NULL;
115  this->proxySocket = NULL;
116  this->networkGameManager = NULL;
117  this->networkMonitor = NULL;
118
119  this->pInfo = new PeerInfo();
120  this->pInfo->userId = NET_UID_UNASSIGNED;
121  this->pInfo->lastAckedState = 0;
122  this->pInfo->lastRecvedState = 0;
123
124  this->bRedirect = false;
125
126  this->currentState = 0;
127
128  remainingBytesToWriteToDict = Preferences::getInstance()->getInt( "compression", "writedict", 0 );
129
130  assert( Zip::getInstance()->loadDictionary( "testdict" ) >= 0 );
131  this->dictClient = Zip::getInstance()->loadDictionary( "dict2pl_client" );
132  assert( this->dictClient >= 0 );
133  this->dictServer = Zip::getInstance()->loadDictionary( "dict2p_server" );
134  assert( this->dictServer >= 0 );
135}
136
137
138/**
139 * deconstructor
140 */
141NetworkStream::~NetworkStream()
142{
143  if ( this->clientSocket )
144  {
145    clientSocket->close();
146    delete clientSocket;
147    clientSocket = NULL;
148  }
149  if ( this->proxySocket)
150  {
151    proxySocket->close();
152    delete proxySocket;
153    proxySocket = NULL;
154  }
155  for ( PeerList::iterator i = peers.begin(); i!=peers.end(); i++)
156  {
157    if ( i->second.socket )
158    {
159      i->second.socket->disconnectServer();
160      delete i->second.socket;
161      i->second.socket = NULL;
162    }
163
164    if ( i->second.handshake )
165    {
166      delete i->second.handshake;
167      i->second.handshake = NULL;
168    }
169
170    if ( i->second.connectionMonitor )
171    {
172      delete i->second.connectionMonitor;
173      i->second.connectionMonitor = NULL;
174    }
175  }
176  for ( SynchronizeableList::const_iterator it = getSyncBegin(); it != getSyncEnd(); it ++ )
177    (*it)->setNetworkStream( NULL );
178
179  if( this->pInfo)
180    delete this->pInfo;
181
182  if( this->networkMonitor)
183    delete this->networkMonitor;
184}
185
186
187/**
188 * establish a connection to a remote master server
189 * @param host: host name
190 * @param port: the port number
191 */
192void NetworkStream::connectToMasterServer(std::string host, int port)
193{
194  int node = NET_ID_MASTER_SERVER;
195  // this create the new node in the peers map
196  this->peers[node].socket = new UdpSocket( host, port );
197  this->peers[node].userId = NET_ID_MASTER_SERVER;
198
199  this->peers[node].nodeType = NET_MASTER_SERVER;
200  this->peers[node].connectionMonitor = new ConnectionMonitor( NET_ID_MASTER_SERVER );
201  this->peers[node].ip = this->peers[node].socket->getRemoteAddress();
202}
203
204
205/**
206 * establish a connection to a remote proxy server
207 * @param host: host name
208 * @param port: the port number
209 */
210void NetworkStream::connectToProxyServer(int proxyId, std::string host, int port)
211{
212  PRINTF(0)("connect to proxy %s, this is proxyId %i\n", host.c_str(), proxyId);
213
214  // this creates the new proxyId in the peers map
215  this->peers[proxyId].socket = new UdpSocket( host, port );
216  this->peers[proxyId].userId = proxyId;
217
218  this->peers[proxyId].nodeType = NET_PROXY_SERVER_ACTIVE;
219  this->peers[proxyId].connectionMonitor = new ConnectionMonitor( proxyId );
220  this->peers[proxyId].ip = this->peers[proxyId].socket->getRemoteAddress();
221}
222
223
224/**
225 * create a server
226 * @param port: interface port for all clients
227 */
228void NetworkStream::createServer(int clientPort, int proxyPort)
229{
230  PRINTF(0)(" Creating new Server: listening for clients on port %i and for proxies on port %i", clientPort, proxyPort);
231  this->clientSocket= new UdpServerSocket(clientPort);
232  this->proxySocket = new UdpServerSocket(proxyPort);
233}
234
235
236/**
237 * creates a new instance of the network game manager
238 */
239void NetworkStream::createNetworkGameManager()
240{
241  this->networkGameManager = NetworkGameManager::getInstance();
242
243  this->networkGameManager->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
244  MessageManager::getInstance()->setUniqueID( SharedNetworkData::getInstance()->getNewUniqueID() );
245}
246
247
248/**
249 * starts the network handshake
250 * handsakes are always initialized from the client side first. this starts the handshake and therefore is only
251 * executed as client
252 * @param userId: start handshake for this user id (optional, default == 0)
253 */
254void NetworkStream::startHandshake(int userId)
255{
256  Handshake* hs = new Handshake(this->pInfo->nodeType);
257  // fake the unique id
258  hs->setUniqueID( NET_UID_HANDSHAKE );
259  assert( peers[userId].handshake == NULL );
260  peers[userId].handshake = hs;
261
262  // set the preferred nick name
263  hs->setPreferedNickName( Preferences::getInstance()->getString( "multiplayer", "nickname", "Player" ) );
264
265  PRINTF(0)("NetworkStream: Handshake created: %s\n", hs->getCName());
266}
267
268
269/**
270 * this functions connects a synchronizeable to the networkstream, therefore synchronizeing
271 * it all over the network and creating it on the other platforms (if and only if it is a
272 * server
273 * @param sync: the synchronizeable to add
274 */
275void NetworkStream::connectSynchronizeable(Synchronizeable& sync)
276{
277  this->synchronizeables.push_back(&sync);
278  sync.setNetworkStream( this );
279}
280
281
282/**
283 * removes the synchronizeable from the list of synchronized entities
284 * @param sync: the syncronizeable to remove
285 */
286void NetworkStream::disconnectSynchronizeable(Synchronizeable& sync)
287{
288  // removing the Synchronizeable from the List.
289  std::list<Synchronizeable*>::iterator disconnectSynchro = std::find(this->synchronizeables.begin(), this->synchronizeables.end(), &sync);
290  if (disconnectSynchro != this->synchronizeables.end())
291    this->synchronizeables.erase(disconnectSynchro);
292
293  oldSynchronizeables[sync.getUniqueID()] = SDL_GetTicks();
294}
295
296
297/**
298 * this is called to process data from the network socket to the synchronizeable and vice versa
299 */
300void NetworkStream::processData()
301{
302  // create the network monitor after all the init work and before there is any connection handlings
303  if( this->networkMonitor == NULL)
304    this->networkMonitor = new NetworkMonitor(this);
305
306
307  int tick = SDL_GetTicks();
308
309  this->currentState++;
310  // there was a wrap around
311  if( this->currentState < 0)
312  {
313    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");
314  }
315
316  if ( SharedNetworkData::getInstance()->isMasterServer())
317  {
318    // execute everytthing the master server shoudl do
319    if ( this->clientSocket )
320      this->clientSocket->update();
321    if( this->proxySocket)
322      this->proxySocket->update();
323
324    this->updateConnectionList();
325  }
326  else if( SharedNetworkData::getInstance()->isProxyServerActive())
327  {
328    //execute everything the proxy server should do
329    if ( this->clientSocket )
330      this->clientSocket->update();
331    if( this->proxySocket)
332      this->proxySocket->update();
333
334    this->updateConnectionList();
335  }
336
337#warning make this more modular: every proxy/master server connection should be watched for termination
338  if( !SharedNetworkData::getInstance()->isMasterServer())
339  {
340    // check if the connection is ok else terminate and remove
341    if ( !peers.empty() && peers[NET_ID_MASTER_SERVER].socket &&
342          ( !peers[NET_ID_MASTER_SERVER].socket->isOk() ||
343          peers[NET_ID_MASTER_SERVER].connectionMonitor->hasTimedOut() ) )
344    {
345      this->handleDisconnect( NET_ID_MASTER_SERVER);
346      PRINTF(1)("lost connection to server\n");
347    }
348    // check if there is a redirection command
349    if( this->bRedirect)
350    {
351      this->handleReconnect( NET_ID_MASTER_SERVER);
352    }
353  }
354
355  this->cleanUpOldSyncList();
356  this->handleHandshakes();
357
358  // update the network monitor
359  this->networkMonitor->process();
360
361  // order of up/downstream is important!!!!
362  // don't change it
363  this->handleDownstream( tick );
364  this->handleUpstream( tick );
365}
366
367
368/**
369 * @brief handles incoming connections
370 *
371 * if we are a NET_MASTER_SERVER or NET_PROXY_SERVER_ACTIVE update the connection list to accept new connections (clients)
372 * start and initialize the handsake for the new clients
373 */
374void NetworkStream::updateConnectionList( )
375{
376  //check for new connections
377
378  NetworkSocket* tempNetworkSocket = NULL;
379  int userId;
380
381  if( this->clientSocket != NULL)
382  {
383    tempNetworkSocket = this->clientSocket->getNewSocket();
384
385    // we got new NET_CLIENT connecting
386    if ( tempNetworkSocket )
387    {
388      // determine the network node id
389      if ( freeSocketSlots.size() > 0 )
390      {
391        // this should never be called
392        assert(false);
393        userId = freeSocketSlots.back();
394        freeSocketSlots.pop_back();
395      }
396      else
397      {
398        //userId = 1;
399        // each server (proxy and master) have an address space for new network nodes of 1000 nodes
400        userId = SharedNetworkData::getInstance()->getHostID() * 1000 + 1;
401
402        for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
403          if ( it->first >= userId )
404            userId = it->first + 1;
405
406        // make sure that this server only uses an address space of 1000
407        assert( userId < (SharedNetworkData::getInstance()->getHostID() + 1) * 1000);
408      }
409      // this creates a new entry in the peers list
410      peers[userId].socket = tempNetworkSocket;
411      peers[userId].nodeType = NET_CLIENT;
412
413      // handle the newly connected client
414      this->handleConnect(userId);
415
416      PRINTF(0)("New Client: %d\n", userId);
417    }
418  }
419
420
421  if( this->proxySocket != NULL)
422  {
423    tempNetworkSocket = this->proxySocket->getNewSocket();
424
425    // we got new NET_PROXY_SERVER_ACTIVE connecting
426    if ( tempNetworkSocket )
427    {
428      // determine the network node id
429      if ( freeSocketSlots.size() > 0 )
430      {
431        userId = freeSocketSlots.back();
432        freeSocketSlots.pop_back();
433      }
434      else
435      {
436        userId = 1;
437
438        for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
439          if ( it->first >= userId )
440            userId = it->first + 1;
441      }
442
443      // this creates a new entry in the peers list
444      peers[userId].socket = tempNetworkSocket;
445      peers[userId].nodeType = NET_PROXY_SERVER_ACTIVE;
446
447      // handle the newly connected proxy server
448      this->handleConnect(userId);
449
450      PRINTF(0)("New proxy connected: %d\n", userId);
451    }
452  }
453
454
455
456  //check if connections are ok else remove them
457  for ( PeerList::iterator it = peers.begin(); it != peers.end(); )
458  {
459    if (
460          it->second.socket &&
461          (
462            !it->second.socket->isOk()  ||
463            it->second.connectionMonitor->hasTimedOut()
464          )
465       )
466    {
467      std::string reason = "disconnected";
468      if ( it->second.connectionMonitor->hasTimedOut() )
469        reason = "timeout";
470      PRINTF(0)("Client is gone: %d (%s)\n", it->second.userId, reason.c_str());
471
472
473      this->handleDisconnect( it->second.userId);
474
475      if( SharedNetworkData::getInstance()->isProxyServerActive())
476        ProxyControl::getInstance()->signalLeaveClient(it->second.userId);
477
478      it++;
479      continue;
480    }
481
482    it++;
483  }
484
485
486}
487
488
489/**
490 * this handles new connections
491 * @param userId: the id of the new user node
492 */
493void NetworkStream::handleConnect( int userId)
494{
495  // create new handshake and init its variables
496  peers[userId].handshake = new Handshake(this->pInfo->nodeType, userId, this->networkGameManager->getUniqueID(), MessageManager::getInstance()->getUniqueID());
497  peers[userId].handshake->setUniqueID(userId);
498
499  peers[userId].connectionMonitor = new ConnectionMonitor( userId );
500  peers[userId].userId = userId;
501
502  PRINTF(0)("num sync: %d\n", synchronizeables.size());
503
504  // get the proxy server informations and write them to the handshake, if any (proxy)
505  assert( this->networkMonitor != NULL);
506  PeerInfo* pi = this->networkMonitor->getFirstChoiceProxy();
507  if( pi != NULL)
508  {
509    peers[userId].handshake->setProxy1Address( pi->ip);
510  }
511  pi = this->networkMonitor->getSecondChoiceProxy();
512  if( pi != NULL)
513    peers[userId].handshake->setProxy2Address( pi->ip);
514
515  // check if the connecting client should reconnect to a proxy server
516  if( SharedNetworkData::getInstance()->isMasterServer())
517    peers[userId].handshake->setRedirect(/*this->networkMonitor->isReconnectNextClient()*/false);
518
519  // the connecting node of course is a client
520  peers[userId].ip = peers[userId].socket->getRemoteAddress();
521}
522
523
524
525/**
526 * some debug output
527 */
528void NetworkStream::debug()
529{
530  if( SharedNetworkData::getInstance()->isMasterServer()) {
531    PRINT(0)(" Host ist Master Server with ID: %i\n", this->pInfo->userId);
532  }
533  else if( SharedNetworkData::getInstance()->isProxyServerActive()) {
534    PRINT(0)(" Host ist Proxy Server with ID: %i\n", this->pInfo->userId);
535  }
536  else {
537    PRINT(0)(" Host ist Client with ID: %i\n", this->pInfo->userId);
538  }
539
540  PRINTF(0)("Current number of connections is: %i\n", this->peers.size());
541  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
542  {
543    PRINTF(0)("peers[%i] with uniqueId %i and address: %s\n", it->first, it->second.userId, it->second.ip.ipString().c_str());
544  }
545
546
547  PRINT(0)(" Got %i connected Synchronizeables, showing active Syncs:\n", this->synchronizeables.size());
548  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
549  {
550    if( (*it)->beSynchronized() == true)
551      PRINT(0)("  Synchronizeable of class: %s::%s, with unique ID: %i, Synchronize: %i\n", (*it)->getClassCName(), (*it)->getCName(),
552               (*it)->getUniqueID(), (*it)->beSynchronized());
553  }
554  PRINT(0)(" Maximal Connections: %i\n", SharedNetworkData::getInstance()->getMaxPlayer() );
555
556}
557
558
559/**
560 * @returns the number of synchronizeables registered to this stream
561 */
562int NetworkStream::getSyncCount()
563{
564  int n = 0;
565  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
566    if( (*it)->beSynchronized() == true)
567      ++n;
568
569  //return synchronizeables.size();
570  return n;
571}
572
573
574/**
575 * check if handshakes completed. if so create the network game manager else remove it again
576 */
577void NetworkStream::handleHandshakes( )
578{
579  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
580  {
581    if ( it->second.handshake )
582    {
583      // handshake finished
584      if ( it->second.handshake->completed() )
585      {
586        //handshake is correct
587        if ( it->second.handshake->ok() )
588        {
589          // write the first informations into the node so they can be read from there for case differentiation
590          it->second.nodeType = it->second.handshake->getRemoteNodeType();
591
592          // the counter part didn't mark it free for deletion yet
593          if ( !it->second.handshake->allowDel() )
594          {
595            // make sure this is a connection:
596            // - client       <==> master server
597            // - proxy server <==> master server
598            if(  SharedNetworkData::getInstance()->isClient() ||
599                 SharedNetworkData::getInstance()->isProxyServerActive() &&
600                 SharedNetworkData::getInstance()->isUserMasterServer(it->second.userId))
601            {
602              PRINTF(0)("Handshake: i am in client role\n");
603
604              SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
605              this->pInfo->userId = SharedNetworkData::getInstance()->getHostID();
606
607#warning the ip address is not set here because it results in a segfault when connecting to a proxy server => trace this later
608//               it->second.ip = it->second.socket->getRemoteAddress();
609
610              // it->second.nodeType = it->second.handshake->getRemoteNodeType();
611              // it->second.ip = it->second.socket->getRemoteAddress();
612              // add the new server to the nodes list (it can be a NET_MASTER_SERVER or NET_PROXY_SERVER)
613              this->networkMonitor->addNode(&it->second);
614              // get proxy 1 address and add it
615              this->networkMonitor->addNode(it->second.handshake->getProxy1Address(), NET_PROXY_SERVER_ACTIVE);
616              // get proxy 2 address and add it
617              this->networkMonitor->addNode(it->second.handshake->getProxy2Address(), NET_PROXY_SERVER_ACTIVE);
618
619              // now check if the server accepted the connection
620              if( SharedNetworkData::getInstance()->isClient() && it->second.handshake->redirect() )
621              {
622                this->bRedirect = true;
623              }
624
625              // create the new network game manager and init it
626              this->networkGameManager = NetworkGameManager::getInstance();
627              this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
628              // init the new message manager
629              MessageManager::getInstance()->setUniqueID( it->second.handshake->getMessageManagerId() );
630            }
631
632            PRINT(0)("handshake finished id=%d\n", it->second.handshake->getNetworkGameManagerId());
633            it->second.handshake->del();
634
635          }
636          else
637          {
638            // handshake finished registring new player
639            if ( it->second.handshake->canDel() )
640            {
641
642              if (  SharedNetworkData::getInstance()->isMasterServer() )
643              {
644                it->second.ip = it->second.socket->getRemoteAddress();
645
646                this->networkMonitor->addNode(&it->second);
647
648                this->handleNewClient( it->second.userId );
649
650                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
651                {
652                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
653                }
654              }
655              else if ( SharedNetworkData::getInstance()->isProxyServerActive() && it->second.isClient() )
656              {
657                PRINTF(0)("Handshake: Proxy in server role: connecting %i\n", it->second.userId);
658
659                it->second.ip = it->second.socket->getRemoteAddress();
660
661                this->networkMonitor->addNode(&it->second);
662
663                // work with the ProxyControl to init the new client
664                ProxyControl::getInstance()->signalNewClient( it->second.userId);
665
666#warning dont know if this works: nick name handling
667                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
668                {
669                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
670                }
671              }
672
673              PRINT(0)("handshake finished delete it\n");
674              delete it->second.handshake;
675              it->second.handshake = NULL;
676            }
677          }
678
679        }
680        else
681        {
682          PRINT(1)("handshake failed!\n");
683          it->second.socket->disconnectServer();
684        }
685      }
686    }
687  }
688}
689
690
691/**
692 * this functions handles a reconnect event received from the a NET_MASTER_SERVER or NET_PROXY_SERVER
693 */
694void NetworkStream::handleReconnect(int userId)
695{
696  this->bRedirect = false;
697#warning this peer will be created if it does not yet exist: dangerous
698  PeerInfo* pInfo = &this->peers[userId];
699
700  PRINTF(0)("===============================================\n");
701  PRINTF(0)("Client is redirected to the other proxy servers\n");
702  PRINTF(0)("  user id: %i\n", userId);
703  PRINTF(0)("  connecting to: %s\n", this->networkMonitor->getFirstChoiceProxy()->ip.ipString().c_str());
704  PRINTF(0)("===============================================\n");
705
706  // flush the old synchronization states, since the numbering could be completely different
707  pInfo->lastAckedState = 0;
708  pInfo->lastRecvedState = 0;
709
710  // temp save the ip address here
711  IP proxyIP = pInfo->handshake->getProxy1Address();
712
713  // disconnect from the current server and reconnect to proxy server
714  this->handleDisconnect( userId);
715  this->connectToProxyServer(NET_ID_PROXY_SERVER_01, proxyIP.ipString(), 9999);
716  #warning the ports are not yet integrated correctly in the ip class
717
718  // and restart the handshake
719  this->startHandshake( userId);
720}
721
722
723/**
724 * handles the disconnect event
725 * @param userId id of the user to remove
726 */
727void NetworkStream::handleDisconnect( int userId )
728{
729  peers[userId].socket->disconnectServer();
730  delete peers[userId].socket;
731  peers[userId].socket = NULL;
732
733  if ( peers[userId].handshake )
734    delete peers[userId].handshake;
735  peers[userId].handshake = NULL;
736
737  if ( peers[userId].connectionMonitor )
738    delete peers[userId].connectionMonitor;
739  peers[userId].connectionMonitor = NULL;
740
741
742  for ( SynchronizeableList::iterator it2 = synchronizeables.begin(); it2 != synchronizeables.end(); it2++ )  {
743    (*it2)->cleanUpUser( userId );
744  }
745
746  if( SharedNetworkData::getInstance()->isMasterServer())
747    NetworkGameManager::getInstance()->signalLeftPlayer(userId);
748
749  this->freeSocketSlots.push_back( userId );
750
751  peers.erase( userId);
752}
753
754
755
756/**
757 * handle upstream network traffic
758 * @param tick: seconds elapsed since last update
759 */
760void NetworkStream::handleUpstream( int tick )
761{
762  int offset;
763  int n;
764
765  for ( PeerList::reverse_iterator peer = peers.rbegin(); peer != peers.rend(); peer++ )
766  {
767    offset = INTSIZE; // reserve enough space for the packet length
768
769    // continue with the next peer if this peer has no socket assigned (therefore no network)
770    if ( !peer->second.socket )
771      continue;
772
773    // header informations: current state
774    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
775    assert( n == INTSIZE );
776    offset += n;
777
778    // header informations: last acked state
779    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
780    assert( n == INTSIZE );
781    offset += n;
782
783    // header informations: last recved state
784    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
785    assert( n == INTSIZE );
786    offset += n;
787
788    // now write all synchronizeables in the packet
789    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
790    {
791
792      int oldOffset = offset;
793      Synchronizeable & sync = **it;
794
795
796      // do not include synchronizeables with uninit id and syncs that don't want to be synchronized
797      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
798        continue;
799
800      // if handshake not finished only sync handshake
801      if ( peer->second.handshake && sync.getLeafClassID() != CL_HANDSHAKE )
802        continue;
803
804      // if we are a server (both master and proxy servers) and this is not our handshake
805      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
806             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient())
807             && sync.getLeafClassID() == CL_HANDSHAKE && sync.getUniqueID() != peer->second.userId )
808        continue;
809
810      /* list of synchronizeables that will never be synchronized over the network: */
811      // do not sync null parent
812      if ( sync.getLeafClassID() == CL_NULL_PARENT )
813        continue;
814
815
816      assert( sync.getLeafClassID() != 0);
817
818      assert( offset + INTSIZE <= UDP_PACKET_SIZE );
819
820      // server fakes uniqueid == 0 for handshake synchronizeable
821      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
822             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient() ) &&
823             ( sync.getUniqueID() >= 1000 || sync.getUniqueID() <= SharedNetworkData::getInstance()->getMaxPlayer() + 1)
824             /*<= SharedNetworkData::getInstance()->getMaxPlayer() + 1*/) // plus one to handle one client more than the max to redirect it
825        n = Converter::intToByteArray( 0, buf + offset, UDP_PACKET_SIZE - offset );
826      else
827        n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );
828
829
830      assert( n == INTSIZE );
831      offset += n;
832
833      // make space for packet size
834      offset += INTSIZE;
835
836      n = sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, -1000 );
837      offset += n;
838
839      assert( Converter::intToByteArray( n, buf + offset - n - INTSIZE, INTSIZE ) == INTSIZE );
840
841      // check if all data bytes == 0 -> remove data and the synchronizeable from the sync process since there is no update
842      // TODO not all synchronizeables like this maybe add Synchronizeable::canRemoveZeroDiff()
843      bool allZero = true;
844      for ( int i = 0; i < n; i++ )
845      {
846         if ( buf[i+oldOffset+2*INTSIZE] != 0 )
847           allZero = false;
848      }
849      // if there is no new data in this synchronizeable reset the data offset to the last state -> dont synchronizes
850      // data that hast not changed
851      if ( allZero )
852      {
853        offset = oldOffset;
854      }
855    } // all synchronizeables written
856
857
858
859    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
860    {
861      Synchronizeable & sync = **it;
862
863      // again exclude all unwanted syncs
864      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED)
865        continue;
866
867      sync.handleSentState( peer->second.userId, currentState, peer->second.lastAckedState );
868    }
869
870
871    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );
872
873    // now compress the data with the zip library
874    int compLength = 0;
875    if ( SharedNetworkData::getInstance()->isMasterServer() ||
876         SharedNetworkData::getInstance()->isProxyServerActive())
877      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictServer );
878    else
879      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictClient );
880
881    if ( compLength <= 0 )
882    {
883      PRINTF(1)("compression failed!\n");
884      continue;
885    }
886
887    assert( peer->second.socket->writePacket( compBuf, compLength ) );
888
889    if ( this->remainingBytesToWriteToDict > 0 )
890      writeToNewDict( buf, offset, true );
891
892    peer->second.connectionMonitor->processUnzippedOutgoingPacket( tick, buf, offset, currentState );
893    peer->second.connectionMonitor->processZippedOutgoingPacket( tick, compBuf, compLength, currentState );
894
895  }
896}
897
898/**
899 * handle downstream network traffic
900 */
901void NetworkStream::handleDownstream( int tick )
902{
903  int offset = 0;
904
905  int length = 0;
906  int packetLength = 0;
907  int compLength = 0;
908  int uniqueId = 0;
909  int state = 0;
910  int ackedState = 0;
911  int fromState = 0;
912  int syncDataLength = 0;
913
914  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
915  {
916
917    if ( !peer->second.socket )
918      continue;
919
920    while ( 0 < (compLength = peer->second.socket->readPacket( compBuf, UDP_PACKET_SIZE )) )
921    {
922      peer->second.connectionMonitor->processZippedIncomingPacket( tick, compBuf, compLength );
923
924      packetLength = Zip::getInstance()->unZip( compBuf, compLength, buf, UDP_PACKET_SIZE );
925
926      if ( packetLength < 4*INTSIZE )
927      {
928        if ( packetLength != 0 )
929          PRINTF(1)("got too small packet: %d\n", packetLength);
930        continue;
931      }
932
933      if ( this->remainingBytesToWriteToDict > 0 )
934        writeToNewDict( buf, packetLength, false );
935
936      assert( Converter::byteArrayToInt( buf, &length ) == INTSIZE );
937      assert( Converter::byteArrayToInt( buf + INTSIZE, &state ) == INTSIZE );
938      assert( Converter::byteArrayToInt( buf + 2*INTSIZE, &fromState ) == INTSIZE );
939      assert( Converter::byteArrayToInt( buf + 3*INTSIZE, &ackedState ) == INTSIZE );
940      offset = 4*INTSIZE;
941
942      peer->second.connectionMonitor->processUnzippedIncomingPacket( tick, buf, packetLength, state, ackedState );
943
944
945      //if this is an old state drop it
946      if ( state <= peer->second.lastRecvedState )
947        continue;
948
949      if ( packetLength != length )
950      {
951        PRINTF(1)("real packet length (%d) and transmitted packet length (%d) do not match!\n", packetLength, length);
952        peer->second.socket->disconnectServer();
953        continue;
954      }
955
956      while ( offset + 2 * INTSIZE < length )
957      {
958        // read the unique id of the sync
959        assert( offset > 0 );
960        assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
961        offset += INTSIZE;
962
963        // read the data length
964        assert( Converter::byteArrayToInt( buf + offset, &syncDataLength ) == INTSIZE );
965        offset += INTSIZE;
966
967        assert( syncDataLength > 0 );
968        assert( syncDataLength < 10000 );
969
970        Synchronizeable * sync = NULL;
971
972        // look for the synchronizeable in question
973        for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
974        {
975          // client thinks his handshake has a special id: hostId * 1000 (host id of this server)
976          if ( (*it)->getUniqueID() == uniqueId ||
977                 ( uniqueId == 0  && (*it)->getUniqueID() == peer->second.userId ) )
978          {
979            sync = *it;
980            break;
981          }
982        }
983
984        // this synchronizeable does not yet exist! create it
985        if ( sync == NULL )
986        {
987          PRINTF(0)("could not find sync with id %d. try to create it\n", uniqueId);
988
989          // if it is an old synchronizeable already removed, ignore it
990          if ( oldSynchronizeables.find( uniqueId ) != oldSynchronizeables.end() )
991          {
992            offset += syncDataLength;
993            continue;
994          }
995
996          // if the node we got this unknown sync we ignore it if:
997          //  - the remote host is a client
998          //  - the remote host is a proxy server and we are master server
999          // (since it has no rights to create a new sync)
1000          if ( peers[peer->second.userId].isClient() ||
1001               (peers[peer->second.userId].isProxyServerActive() && SharedNetworkData::getInstance()->isMasterServer()))
1002          {
1003            offset += syncDataLength;
1004            continue;
1005          }
1006
1007          int leafClassId;
1008          if ( INTSIZE > length - offset )
1009          {
1010            offset += syncDataLength;
1011            continue;
1012          }
1013
1014          Converter::byteArrayToInt( buf + offset, &leafClassId );
1015
1016          assert( leafClassId != 0 );
1017
1018
1019          BaseObject * b = NULL;
1020          /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
1021          /* Exception 1: NullParent */
1022          if( leafClassId == CL_NULL_PARENT || leafClassId == CL_SYNCHRONIZEABLE || leafClassId == CL_NETWORK_GAME_MANAGER )
1023          {
1024            PRINTF(1)("Don't create Object with ID %x, ignored!\n", (int)leafClassId);
1025            offset += syncDataLength;
1026            continue;
1027          }
1028          else
1029            b = Factory::fabricate( (ClassID)leafClassId );
1030
1031          if ( !b )
1032          {
1033            PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
1034            offset += syncDataLength;
1035            continue;
1036          }
1037
1038          if ( b->isA(CL_SYNCHRONIZEABLE) )
1039          {
1040            sync = dynamic_cast<Synchronizeable*>(b);
1041            sync->setUniqueID( uniqueId );
1042            sync->setSynchronized(true);
1043
1044            PRINTF(0)("Fabricated %s with id %d\n", sync->getClassCName(), sync->getUniqueID());
1045          }
1046          else
1047          {
1048            PRINTF(1)("Class with ID %x is not a synchronizeable!\n", (int)leafClassId);
1049            delete b;
1050            offset += syncDataLength;
1051            continue;
1052          }
1053        }
1054
1055
1056        int n = sync->setStateDiff( peer->second.userId, buf+offset, syncDataLength, state, fromState );
1057        offset += n;
1058
1059      }
1060
1061      if ( offset != length )
1062      {
1063        PRINTF(0)("offset (%d) != length (%d)\n", offset, length);
1064        peer->second.socket->disconnectServer();
1065      }
1066
1067
1068      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
1069      {
1070        Synchronizeable & sync = **it;
1071
1072        if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
1073          continue;
1074
1075        sync.handleRecvState( peer->second.userId, state, fromState );
1076      }
1077
1078      assert( peer->second.lastAckedState <= ackedState );
1079      peer->second.lastAckedState = ackedState;
1080
1081      assert( peer->second.lastRecvedState < state );
1082      peer->second.lastRecvedState = state;
1083
1084    }
1085
1086  }
1087
1088}
1089
1090/**
1091 * is executed when a handshake has finished
1092 */
1093void NetworkStream::handleNewClient( int userId )
1094{
1095  // init and assign the message manager
1096  MessageManager::getInstance()->initUser( userId );
1097  // do all game relevant stuff here
1098  networkGameManager->signalNewPlayer( userId );
1099}
1100
1101
1102/**
1103 * removes old items from oldSynchronizeables
1104 */
1105void NetworkStream::cleanUpOldSyncList( )
1106{
1107  int now = SDL_GetTicks();
1108
1109  for ( std::map<int,int>::iterator it = oldSynchronizeables.begin(); it != oldSynchronizeables.end();  )
1110  {
1111    if ( it->second < now - 10*1000 )
1112    {
1113      std::map<int,int>::iterator delIt = it;
1114      it++;
1115      oldSynchronizeables.erase( delIt );
1116      continue;
1117    }
1118    it++;
1119  }
1120}
1121
1122/**
1123 * writes data to DATA/dicts/newdict
1124 * @param data pointer to data
1125 * @param length length
1126 */
1127void NetworkStream::writeToNewDict( byte * data, int length, bool upstream )
1128{
1129  if ( remainingBytesToWriteToDict <= 0 )
1130    return;
1131
1132  if ( length > remainingBytesToWriteToDict )
1133    length = remainingBytesToWriteToDict;
1134
1135  std::string fileName = ResourceManager::getInstance()->getDataDir();
1136  fileName += "/dicts/newdict";
1137
1138  if ( upstream )
1139    fileName += "_upstream";
1140  else
1141    fileName += "_downstream";
1142
1143  FILE * f = fopen( fileName.c_str(), "a" );
1144
1145  if ( !f )
1146  {
1147    PRINTF(2)("could not open %s\n", fileName.c_str());
1148    remainingBytesToWriteToDict = 0;
1149    return;
1150  }
1151
1152  if ( fwrite( data, 1, length, f ) != length )
1153  {
1154    PRINTF(2)("could not write to file\n");
1155    fclose( f );
1156    return;
1157  }
1158
1159  fclose( f );
1160
1161  remainingBytesToWriteToDict -= length;
1162}
1163
1164
1165
1166
1167
1168
Note: See TracBrowser for help on using the repository browser.