Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

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

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

found the bug, master-server proxy architecture is working correctly. yeaaa

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  PRINT(0)(" Current number of connections is: %i\n", this->peers.size());
541  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
542  {
543    PRINT(0)("  peers[%i] with uniqueId %i and address: %s\n", it->first, it->second.userId, it->second.ip.ipString().c_str());
544  }
545  PRINT(0)("\n\n");
546
547
548  PRINT(0)(" Got %i connected Synchronizeables, showing active Syncs:\n", this->synchronizeables.size());
549  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
550  {
551    if( (*it)->beSynchronized() == true)
552      PRINT(0)("  Synchronizeable of class: %s::%s, with unique ID: %i, Synchronize: %i\n", (*it)->getClassCName(), (*it)->getCName(),
553               (*it)->getUniqueID(), (*it)->beSynchronized());
554  }
555  PRINT(0)(" Maximal Connections: %i\n", SharedNetworkData::getInstance()->getMaxPlayer() );
556
557}
558
559
560/**
561 * @returns the number of synchronizeables registered to this stream
562 */
563int NetworkStream::getSyncCount()
564{
565  int n = 0;
566  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
567    if( (*it)->beSynchronized() == true)
568      ++n;
569
570  //return synchronizeables.size();
571  return n;
572}
573
574
575/**
576 * check if handshakes completed. if so create the network game manager else remove it again
577 */
578void NetworkStream::handleHandshakes( )
579{
580  for ( PeerList::iterator it = peers.begin(); it != peers.end(); it++ )
581  {
582    if ( it->second.handshake )
583    {
584      // handshake finished
585      if ( it->second.handshake->completed() )
586      {
587        //handshake is correct
588        if ( it->second.handshake->ok() )
589        {
590          // write the first informations into the node so they can be read from there for case differentiation
591          it->second.nodeType = it->second.handshake->getRemoteNodeType();
592
593          // the counter part didn't mark it free for deletion yet
594          if ( !it->second.handshake->allowDel() )
595          {
596            // make sure this is a connection:
597            // - client       <==> master server
598            // - proxy server <==> master server
599            if(  SharedNetworkData::getInstance()->isClient() ||
600                 SharedNetworkData::getInstance()->isProxyServerActive() &&
601                 SharedNetworkData::getInstance()->isUserMasterServer(it->second.userId))
602            {
603              PRINTF(0)("Handshake: i am in client role\n");
604
605              SharedNetworkData::getInstance()->setHostID( it->second.handshake->getHostId() );
606              this->pInfo->userId = SharedNetworkData::getInstance()->getHostID();
607
608#warning the ip address is not set here because it results in a segfault when connecting to a proxy server => trace this later
609//               it->second.ip = it->second.socket->getRemoteAddress();
610
611              // it->second.nodeType = it->second.handshake->getRemoteNodeType();
612              // it->second.ip = it->second.socket->getRemoteAddress();
613              // add the new server to the nodes list (it can be a NET_MASTER_SERVER or NET_PROXY_SERVER)
614              this->networkMonitor->addNode(&it->second);
615              // get proxy 1 address and add it
616              this->networkMonitor->addNode(it->second.handshake->getProxy1Address(), NET_PROXY_SERVER_ACTIVE);
617              // get proxy 2 address and add it
618              this->networkMonitor->addNode(it->second.handshake->getProxy2Address(), NET_PROXY_SERVER_ACTIVE);
619
620              // now check if the server accepted the connection
621              if( SharedNetworkData::getInstance()->isClient() && it->second.handshake->redirect() )
622              {
623                this->bRedirect = true;
624              }
625
626              // create the new network game manager and init it
627              this->networkGameManager = NetworkGameManager::getInstance();
628              this->networkGameManager->setUniqueID( it->second.handshake->getNetworkGameManagerId() );
629              // init the new message manager
630              MessageManager::getInstance()->setUniqueID( it->second.handshake->getMessageManagerId() );
631            }
632
633            PRINT(0)("handshake finished id=%d\n", it->second.handshake->getNetworkGameManagerId());
634            it->second.handshake->del();
635
636          }
637          else
638          {
639            // handshake finished registring new player
640            if ( it->second.handshake->canDel() )
641            {
642
643              if (  SharedNetworkData::getInstance()->isMasterServer() )
644              {
645                it->second.ip = it->second.socket->getRemoteAddress();
646
647                this->networkMonitor->addNode(&it->second);
648
649                this->handleNewClient( it->second.userId );
650
651                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
652                {
653                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
654                }
655              }
656              else if ( SharedNetworkData::getInstance()->isProxyServerActive() && it->second.isClient() )
657              {
658                PRINTF(0)("Handshake: Proxy in server role: connecting %i\n", it->second.userId);
659
660                it->second.ip = it->second.socket->getRemoteAddress();
661
662                this->networkMonitor->addNode(&it->second);
663
664                // work with the ProxyControl to init the new client
665                ProxyControl::getInstance()->signalNewClient( it->second.userId);
666
667#warning dont know if this works: nick name handling
668                if ( PlayerStats::getStats( it->second.userId ) && it->second.handshake->getPreferedNickName() != "" )
669                {
670                  PlayerStats::getStats( it->second.userId )->setNickName( it->second.handshake->getPreferedNickName() );
671                }
672              }
673
674              PRINT(0)("handshake finished delete it\n");
675              delete it->second.handshake;
676              it->second.handshake = NULL;
677            }
678          }
679
680        }
681        else
682        {
683          PRINT(1)("handshake failed!\n");
684          it->second.socket->disconnectServer();
685        }
686      }
687    }
688  }
689}
690
691
692/**
693 * this functions handles a reconnect event received from the a NET_MASTER_SERVER or NET_PROXY_SERVER
694 */
695void NetworkStream::handleReconnect(int userId)
696{
697  this->bRedirect = false;
698#warning this peer will be created if it does not yet exist: dangerous
699  PeerInfo* pInfo = &this->peers[userId];
700
701  PRINTF(0)("===============================================\n");
702  PRINTF(0)("Client is redirected to the other proxy servers\n");
703  PRINTF(0)("  user id: %i\n", userId);
704  PRINTF(0)("  connecting to: %s\n", this->networkMonitor->getFirstChoiceProxy()->ip.ipString().c_str());
705  PRINTF(0)("===============================================\n");
706
707  // flush the old synchronization states, since the numbering could be completely different
708  pInfo->lastAckedState = 0;
709  pInfo->lastRecvedState = 0;
710
711  // temp save the ip address here
712  IP proxyIP = pInfo->handshake->getProxy1Address();
713
714  // disconnect from the current server and reconnect to proxy server
715  this->handleDisconnect( userId);
716  this->connectToProxyServer(NET_ID_PROXY_SERVER_01, proxyIP.ipString(), 9999);
717  #warning the ports are not yet integrated correctly in the ip class
718
719  // and restart the handshake
720  this->startHandshake( userId);
721}
722
723
724/**
725 * handles the disconnect event
726 * @param userId id of the user to remove
727 */
728void NetworkStream::handleDisconnect( int userId )
729{
730  peers[userId].socket->disconnectServer();
731  delete peers[userId].socket;
732  peers[userId].socket = NULL;
733
734  if ( peers[userId].handshake )
735    delete peers[userId].handshake;
736  peers[userId].handshake = NULL;
737
738  if ( peers[userId].connectionMonitor )
739    delete peers[userId].connectionMonitor;
740  peers[userId].connectionMonitor = NULL;
741
742
743  for ( SynchronizeableList::iterator it2 = synchronizeables.begin(); it2 != synchronizeables.end(); it2++ )  {
744    (*it2)->cleanUpUser( userId );
745  }
746
747  if( SharedNetworkData::getInstance()->isMasterServer())
748    NetworkGameManager::getInstance()->signalLeftPlayer(userId);
749
750  this->freeSocketSlots.push_back( userId );
751
752  peers.erase( userId);
753}
754
755
756
757/**
758 * handle upstream network traffic
759 * @param tick: seconds elapsed since last update
760 */
761void NetworkStream::handleUpstream( int tick )
762{
763  int offset;
764  int n;
765
766  for ( PeerList::reverse_iterator peer = peers.rbegin(); peer != peers.rend(); peer++ )
767  {
768    offset = INTSIZE; // reserve enough space for the packet length
769
770    // continue with the next peer if this peer has no socket assigned (therefore no network)
771    if ( !peer->second.socket )
772      continue;
773
774    // header informations: current state
775    n = Converter::intToByteArray( currentState, buf + offset, UDP_PACKET_SIZE - offset );
776    assert( n == INTSIZE );
777    offset += n;
778
779    // header informations: last acked state
780    n = Converter::intToByteArray( peer->second.lastAckedState, buf + offset, UDP_PACKET_SIZE - offset );
781    assert( n == INTSIZE );
782    offset += n;
783
784    // header informations: last recved state
785    n = Converter::intToByteArray( peer->second.lastRecvedState, buf + offset, UDP_PACKET_SIZE - offset );
786    assert( n == INTSIZE );
787    offset += n;
788
789    // now write all synchronizeables in the packet
790    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
791    {
792
793      int oldOffset = offset;
794      Synchronizeable & sync = **it;
795
796
797      // do not include synchronizeables with uninit id and syncs that don't want to be synchronized
798      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
799        continue;
800
801      // if handshake not finished only sync handshake
802      if ( peer->second.handshake && sync.getLeafClassID() != CL_HANDSHAKE )
803        continue;
804
805      // if we are a server (both master and proxy servers) and this is not our handshake
806      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
807             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient())
808             && sync.getLeafClassID() == CL_HANDSHAKE && sync.getUniqueID() != peer->second.userId )
809        continue;
810
811      /* list of synchronizeables that will never be synchronized over the network: */
812      // do not sync null parent
813      if ( sync.getLeafClassID() == CL_NULL_PARENT )
814        continue;
815
816
817      assert( sync.getLeafClassID() != 0);
818
819      assert( offset + INTSIZE <= UDP_PACKET_SIZE );
820
821      // server fakes uniqueid == 0 for handshake synchronizeable
822      if ( ( SharedNetworkData::getInstance()->isMasterServer() ||
823             SharedNetworkData::getInstance()->isProxyServerActive() &&  peer->second.isClient() ) &&
824             ( sync.getUniqueID() >= 1000 || sync.getUniqueID() <= SharedNetworkData::getInstance()->getMaxPlayer() + 1)
825             /*<= SharedNetworkData::getInstance()->getMaxPlayer() + 1*/) // plus one to handle one client more than the max to redirect it
826        n = Converter::intToByteArray( 0, buf + offset, UDP_PACKET_SIZE - offset );
827      else
828        n = Converter::intToByteArray( sync.getUniqueID(), buf + offset, UDP_PACKET_SIZE - offset );
829
830
831      assert( n == INTSIZE );
832      offset += n;
833
834      // make space for packet size
835      offset += INTSIZE;
836
837      n = sync.getStateDiff( peer->second.userId, buf + offset, UDP_PACKET_SIZE-offset, currentState, peer->second.lastAckedState, -1000 );
838      offset += n;
839
840      assert( Converter::intToByteArray( n, buf + offset - n - INTSIZE, INTSIZE ) == INTSIZE );
841
842      // check if all data bytes == 0 -> remove data and the synchronizeable from the sync process since there is no update
843      // TODO not all synchronizeables like this maybe add Synchronizeable::canRemoveZeroDiff()
844      bool allZero = true;
845      for ( int i = 0; i < n; i++ )
846      {
847         if ( buf[i+oldOffset+2*INTSIZE] != 0 )
848           allZero = false;
849      }
850      // if there is no new data in this synchronizeable reset the data offset to the last state -> dont synchronizes
851      // data that hast not changed
852      if ( allZero )
853      {
854        offset = oldOffset;
855      }
856    } // all synchronizeables written
857
858
859
860    for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
861    {
862      Synchronizeable & sync = **it;
863
864      // again exclude all unwanted syncs
865      if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED)
866        continue;
867
868      sync.handleSentState( peer->second.userId, currentState, peer->second.lastAckedState );
869    }
870
871
872    assert( Converter::intToByteArray( offset, buf, INTSIZE ) == INTSIZE );
873
874    // now compress the data with the zip library
875    int compLength = 0;
876    if ( SharedNetworkData::getInstance()->isMasterServer() ||
877         SharedNetworkData::getInstance()->isProxyServerActive())
878      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictServer );
879    else
880      compLength = Zip::getInstance()->zip( buf, offset, compBuf, UDP_PACKET_SIZE, dictClient );
881
882    if ( compLength <= 0 )
883    {
884      PRINTF(1)("compression failed!\n");
885      continue;
886    }
887
888    assert( peer->second.socket->writePacket( compBuf, compLength ) );
889
890    if ( this->remainingBytesToWriteToDict > 0 )
891      writeToNewDict( buf, offset, true );
892
893    peer->second.connectionMonitor->processUnzippedOutgoingPacket( tick, buf, offset, currentState );
894    peer->second.connectionMonitor->processZippedOutgoingPacket( tick, compBuf, compLength, currentState );
895
896  }
897}
898
899/**
900 * handle downstream network traffic
901 */
902void NetworkStream::handleDownstream( int tick )
903{
904  int offset = 0;
905
906  int length = 0;
907  int packetLength = 0;
908  int compLength = 0;
909  int uniqueId = 0;
910  int state = 0;
911  int ackedState = 0;
912  int fromState = 0;
913  int syncDataLength = 0;
914
915  for ( PeerList::iterator peer = peers.begin(); peer != peers.end(); peer++ )
916  {
917
918    if ( !peer->second.socket )
919      continue;
920
921    while ( 0 < (compLength = peer->second.socket->readPacket( compBuf, UDP_PACKET_SIZE )) )
922    {
923      peer->second.connectionMonitor->processZippedIncomingPacket( tick, compBuf, compLength );
924
925      packetLength = Zip::getInstance()->unZip( compBuf, compLength, buf, UDP_PACKET_SIZE );
926
927      if ( packetLength < 4*INTSIZE )
928      {
929        if ( packetLength != 0 )
930          PRINTF(1)("got too small packet: %d\n", packetLength);
931        continue;
932      }
933
934      if ( this->remainingBytesToWriteToDict > 0 )
935        writeToNewDict( buf, packetLength, false );
936
937      assert( Converter::byteArrayToInt( buf, &length ) == INTSIZE );
938      assert( Converter::byteArrayToInt( buf + INTSIZE, &state ) == INTSIZE );
939      assert( Converter::byteArrayToInt( buf + 2*INTSIZE, &fromState ) == INTSIZE );
940      assert( Converter::byteArrayToInt( buf + 3*INTSIZE, &ackedState ) == INTSIZE );
941      offset = 4*INTSIZE;
942
943      peer->second.connectionMonitor->processUnzippedIncomingPacket( tick, buf, packetLength, state, ackedState );
944
945
946      //if this is an old state drop it
947      if ( state <= peer->second.lastRecvedState )
948        continue;
949
950      if ( packetLength != length )
951      {
952        PRINTF(1)("real packet length (%d) and transmitted packet length (%d) do not match!\n", packetLength, length);
953        peer->second.socket->disconnectServer();
954        continue;
955      }
956
957      while ( offset + 2 * INTSIZE < length )
958      {
959        // read the unique id of the sync
960        assert( offset > 0 );
961        assert( Converter::byteArrayToInt( buf + offset, &uniqueId ) == INTSIZE );
962        offset += INTSIZE;
963
964        // read the data length
965        assert( Converter::byteArrayToInt( buf + offset, &syncDataLength ) == INTSIZE );
966        offset += INTSIZE;
967
968        assert( syncDataLength > 0 );
969        assert( syncDataLength < 10000 );
970
971        Synchronizeable * sync = NULL;
972
973        // look for the synchronizeable in question
974        for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
975        {
976          // client thinks his handshake has a special id: hostId * 1000 (host id of this server)
977          if ( (*it)->getUniqueID() == uniqueId ||
978                 ( uniqueId == 0  && (*it)->getUniqueID() == peer->second.userId ) )
979          {
980            sync = *it;
981            break;
982          }
983        }
984
985        // this synchronizeable does not yet exist! create it
986        if ( sync == NULL )
987        {
988          PRINTF(0)("could not find sync with id %d. try to create it\n", uniqueId);
989
990          // if it is an old synchronizeable already removed, ignore it
991          if ( oldSynchronizeables.find( uniqueId ) != oldSynchronizeables.end() )
992          {
993            offset += syncDataLength;
994            continue;
995          }
996
997          // if the node we got this unknown sync we ignore it if:
998          //  - the remote host is a client
999          //  - the remote host is a proxy server and we are master server
1000          // (since it has no rights to create a new sync)
1001          if ( peers[peer->second.userId].isClient() ||
1002               (peers[peer->second.userId].isProxyServerActive() && SharedNetworkData::getInstance()->isMasterServer()))
1003          {
1004            offset += syncDataLength;
1005            continue;
1006          }
1007
1008          int leafClassId;
1009          if ( INTSIZE > length - offset )
1010          {
1011            offset += syncDataLength;
1012            continue;
1013          }
1014
1015          Converter::byteArrayToInt( buf + offset, &leafClassId );
1016
1017          assert( leafClassId != 0 );
1018
1019
1020          BaseObject * b = NULL;
1021          /* These are some small exeptions in creation: Not all objects can/should be created via Factory */
1022          /* Exception 1: NullParent */
1023          if( leafClassId == CL_NULL_PARENT || leafClassId == CL_SYNCHRONIZEABLE || leafClassId == CL_NETWORK_GAME_MANAGER )
1024          {
1025            PRINTF(1)("Don't create Object with ID %x, ignored!\n", (int)leafClassId);
1026            offset += syncDataLength;
1027            continue;
1028          }
1029          else
1030            b = Factory::fabricate( (ClassID)leafClassId );
1031
1032          if ( !b )
1033          {
1034            PRINTF(1)("Could not fabricate Object with classID %x\n", leafClassId);
1035            offset += syncDataLength;
1036            continue;
1037          }
1038
1039          if ( b->isA(CL_SYNCHRONIZEABLE) )
1040          {
1041            sync = dynamic_cast<Synchronizeable*>(b);
1042            sync->setUniqueID( uniqueId );
1043            sync->setSynchronized(true);
1044
1045            PRINTF(0)("Fabricated %s with id %d\n", sync->getClassCName(), sync->getUniqueID());
1046          }
1047          else
1048          {
1049            PRINTF(1)("Class with ID %x is not a synchronizeable!\n", (int)leafClassId);
1050            delete b;
1051            offset += syncDataLength;
1052            continue;
1053          }
1054        }
1055
1056
1057        int n = sync->setStateDiff( peer->second.userId, buf+offset, syncDataLength, state, fromState );
1058        offset += n;
1059
1060      }
1061
1062      if ( offset != length )
1063      {
1064        PRINTF(0)("offset (%d) != length (%d)\n", offset, length);
1065        peer->second.socket->disconnectServer();
1066      }
1067
1068
1069      for ( SynchronizeableList::iterator it = synchronizeables.begin(); it != synchronizeables.end(); it++ )
1070      {
1071        Synchronizeable & sync = **it;
1072
1073        if ( !sync.beSynchronized() || sync.getUniqueID() <= NET_UID_UNASSIGNED )
1074          continue;
1075
1076        sync.handleRecvState( peer->second.userId, state, fromState );
1077      }
1078
1079      assert( peer->second.lastAckedState <= ackedState );
1080      peer->second.lastAckedState = ackedState;
1081
1082      assert( peer->second.lastRecvedState < state );
1083      peer->second.lastRecvedState = state;
1084
1085    }
1086
1087  }
1088
1089}
1090
1091/**
1092 * is executed when a handshake has finished
1093 */
1094void NetworkStream::handleNewClient( int userId )
1095{
1096  // init and assign the message manager
1097  MessageManager::getInstance()->initUser( userId );
1098  // do all game relevant stuff here
1099  networkGameManager->signalNewPlayer( userId );
1100}
1101
1102
1103/**
1104 * removes old items from oldSynchronizeables
1105 */
1106void NetworkStream::cleanUpOldSyncList( )
1107{
1108  int now = SDL_GetTicks();
1109
1110  for ( std::map<int,int>::iterator it = oldSynchronizeables.begin(); it != oldSynchronizeables.end();  )
1111  {
1112    if ( it->second < now - 10*1000 )
1113    {
1114      std::map<int,int>::iterator delIt = it;
1115      it++;
1116      oldSynchronizeables.erase( delIt );
1117      continue;
1118    }
1119    it++;
1120  }
1121}
1122
1123/**
1124 * writes data to DATA/dicts/newdict
1125 * @param data pointer to data
1126 * @param length length
1127 */
1128void NetworkStream::writeToNewDict( byte * data, int length, bool upstream )
1129{
1130  if ( remainingBytesToWriteToDict <= 0 )
1131    return;
1132
1133  if ( length > remainingBytesToWriteToDict )
1134    length = remainingBytesToWriteToDict;
1135
1136  std::string fileName = ResourceManager::getInstance()->getDataDir();
1137  fileName += "/dicts/newdict";
1138
1139  if ( upstream )
1140    fileName += "_upstream";
1141  else
1142    fileName += "_downstream";
1143
1144  FILE * f = fopen( fileName.c_str(), "a" );
1145
1146  if ( !f )
1147  {
1148    PRINTF(2)("could not open %s\n", fileName.c_str());
1149    remainingBytesToWriteToDict = 0;
1150    return;
1151  }
1152
1153  if ( fwrite( data, 1, length, f ) != length )
1154  {
1155    PRINTF(2)("could not write to file\n");
1156    fclose( f );
1157    return;
1158  }
1159
1160  fclose( f );
1161
1162  remainingBytesToWriteToDict -= length;
1163}
1164
1165
1166
1167
1168
1169
Note: See TracBrowser for help on using the repository browser.