Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

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

Last change on this file since 9547 was 9545, checked in by patrick, 19 years ago

initializing the default values in a better way to prevent missunderstandings

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