Planet
navi homePPSaboutscreenshotsdownloaddevelopmentforum

source: orxonox.OLD/trunk/src/lib/network/message_manager.cc @ 9850

Last change on this file since 9850 was 9656, checked in by bensch, 18 years ago

orxonox/trunk: merged the proxy bache back with no conflicts

File size: 16.3 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
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 (boenzlip@ee.ethz.ch)
17     July 2006: message forwarding algorithms
18*/
19
20//#define DEBUG_SPECIAL_MODULE DEBUG_MODULE_
21
22#include "message_manager.h"
23
24#include "network_stream.h"
25#include "shared_network_data.h"
26#include "converter.h"
27#include <cassert>
28
29
30
31MessageManager* MessageManager::singletonRef = NULL;
32
33
34/**
35 * standard constructor
36*/
37MessageManager::MessageManager ()
38{
39  this->setClassID( CL_MESSAGE_MANAGER, "MessageManager" );
40  newNumber = 1;
41  setSynchronized( true );
42}
43
44
45/**
46 * standard deconstructor
47*/
48MessageManager::~MessageManager ()
49{
50  for ( MessageQueue::iterator it = outgoingMessageQueue.begin(); it != outgoingMessageQueue.end(); it++ )
51  {
52    for ( std::list<NetworkMessage>::iterator it2 = it->second.messages.begin(); it2 != it->second.messages.end(); it2++ )
53    {
54      if ( it2->data )
55      {
56        delete [] it2->data;
57        it2->data = NULL;
58      }
59    }
60
61    it->second.messages.clear();
62    it->second.toAck.clear();
63  }
64
65  outgoingMessageQueue.clear();
66
67  this->messageHandlerMap.clear();
68
69  MessageManager::singletonRef = NULL;
70}
71
72/**
73 * get the diff to last acked state of userId
74 *
75 * this class does not use the normal SynchronizeableVars for zynchronisation. instead
76 * it defines its own protocol
77 *
78 * @param userId user to create diff for
79 * @param data buffer to copy diff in
80 * @param maxLength max bytes to copy into data
81 * @param stateId id of current state
82 * @param fromStateId the reference state for the delta state
83 * @param priorityTH tells getStateDiff to not send element with priority \< priorityTH
84 * @return n bytes copied into data
85 */
86int MessageManager::getStateDiff( int userId, byte * data, int maxLength, int stateId, int fromStateId, int priorityTH )
87{
88  int i = 0;
89  int n;
90
91  n = Converter::intToByteArray( outgoingMessageQueue[userId].toAck.size(), data + i, maxLength );
92  i += n;
93  assert( n == INTSIZE );
94
95  for ( std::list<int>::iterator it = outgoingMessageQueue[userId].toAck.begin(); it != outgoingMessageQueue[userId].toAck.end(); it++)
96  {
97    n = Converter::intToByteArray( *it, data + i, maxLength );
98    i += n;
99    assert( n == INTSIZE );
100  }
101
102  outgoingMessageQueue[userId].toAck.clear();
103
104  n = Converter::intToByteArray( outgoingMessageQueue[userId].messages.size(), data + i, maxLength );
105  i += n;
106  assert( n == INTSIZE );
107
108  // write the message down, a message has this structure:
109  //   | data_length | serial_number | message_type | source_id | dest_id | ...data... |
110  //      4byte        4byte            4byte         4byte      4byte     data_length
111  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end(); it++ )
112  {
113    // send data length
114    n = Converter::intToByteArray( it->length, data + i, maxLength );
115    i += n;
116    assert( n == INTSIZE );
117
118    // send serial number
119    n = Converter::intToByteArray( it->number, data + i, maxLength );
120    i += n;
121    assert( n == INTSIZE );
122
123    // send message type
124    n = Converter::intToByteArray( it->messageType, data + i, maxLength );
125    i += n;
126    assert( n == INTSIZE );
127
128    // send sender id
129    n = Converter::intToByteArray( it->senderId, data + i, maxLength );
130    i += n;
131    assert( n == INTSIZE );
132
133    // send destination id
134    n = Converter::intToByteArray( it->destinationId, data + i, maxLength );
135    i += n;
136    assert( n == INTSIZE );
137
138    // send receiver type
139    n = Converter::intToByteArray( it->recieverType, data + i, maxLength );
140    i += n;
141    assert( n == INTSIZE );
142
143    // and copy the data
144    assert( i + it->length <= maxLength );
145    memcpy( data + i, it->data, it->length );
146    i += it->length;
147  }
148
149  return i;
150}
151
152/**
153 * sets a new state out of a diff created on another host
154 * @param userId hostId of user who send me that diff
155 * @param data pointer to diff
156 * @param length length of diff
157 * @param stateId id of current state
158 * @param fromStateId id of the base state id
159 * @return number bytes read
160 * @todo check for permissions
161 */
162int MessageManager::setStateDiff( int userId, byte * data, int length, int stateId, int fromStateId )
163{
164  int i = 0;
165  int n;
166
167  int nAcks;
168
169
170  assert( i + INTSIZE <= length );
171  n = Converter::byteArrayToInt( data + i, &nAcks );
172  assert( n == INTSIZE );
173  i += n;
174
175  std::list<int> acks;
176
177  int number;
178
179  for ( int j = 0; j < nAcks; j++ )
180  {
181    assert( i + INTSIZE <= length );
182    n = Converter::byteArrayToInt( data + i, &number );
183    assert( n == INTSIZE );
184    i += n;
185
186    acks.push_back( number );
187  }
188
189  int nMessages;
190
191  assert( i + INTSIZE <= length );
192  n = Converter::byteArrayToInt( data + i, &nMessages );
193  assert( n == INTSIZE );
194  i += n;
195
196  int messageLength, messageType;
197  int senderId, destinationId, recieverType;
198
199  // now go through all newly received messages and assemble them
200  for ( int j = 0; j < nMessages; j++ )
201  {
202    // read the length
203    assert( i + INTSIZE <= length );
204    n = Converter::byteArrayToInt( data + i, &messageLength );
205    assert( n == INTSIZE );
206    i += n;
207
208    // read the serial number
209    assert( i + INTSIZE <= length );
210    n = Converter::byteArrayToInt( data + i, &number );
211    assert( n == INTSIZE );
212    i += n;
213
214    // read the message type
215    assert( i + INTSIZE <= length );
216    n = Converter::byteArrayToInt( data + i, &messageType );
217    assert( n == INTSIZE );
218    i += n;
219
220    // read the sender id
221    assert( i + INTSIZE <= length );
222    n = Converter::byteArrayToInt( data + i, &senderId );
223    assert( n == INTSIZE );
224    i += n;
225
226    //read the destination id
227    assert( i + INTSIZE <= length );
228    n = Converter::byteArrayToInt( data + i, &destinationId);
229    assert( n == INTSIZE );
230    i += n;
231
232    // read the receiver type
233    assert( i + INTSIZE <= length );
234    n = Converter::byteArrayToInt( data + i, &recieverType);
235    assert( n == INTSIZE );
236    i += n;
237
238    if ( number > 0 )
239      outgoingMessageQueue[userId].toAck.push_back( number );
240
241//     PRINTF(0)("got message with type: %i\n", messageType);
242    assert( i + messageLength <= length );
243    // make sure there is a message handler for this message type
244    assert( messageHandlerMap.find( (MessageType)messageType ) != messageHandlerMap.end());
245
246
247    if ( std::find( outgoingMessageQueue[userId].recievedMessages.begin(), outgoingMessageQueue[userId].recievedMessages.end(), number ) ==
248         outgoingMessageQueue[userId].recievedMessages.end() )
249    {
250
251      // find out if this message is addressed for this client too
252      if( recieverType == RT_ALL_BUT_ME  && SharedNetworkData::getInstance()->getHostID() != senderId ||
253          recieverType == RT_ALL_ME ||
254          recieverType == RT_NOT_USER && SharedNetworkData::getInstance()->getHostID() != destinationId ||
255          recieverType == RT_USER  && SharedNetworkData::getInstance()->getHostID() == destinationId ||
256          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isMasterServer() ||
257          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isProxyServerActive())
258      {
259
260        PRINTF(0)("<<< MessageManager: got msg with type: %i, from sender %i, to rec: %i\n", messageType, senderId, destinationId);
261      // call the handler function and handle errors
262        if ( !(*(messageHandlerMap[(MessageType)messageType].cb))( (MessageType)messageType, data + i, messageLength,
263                 messageHandlerMap[(MessageType)messageType].someData, senderId, destinationId ) )
264        {
265        // if the message is not handled correctly, bush it back to the incoming packets therefore trying it later
266          NetworkMessage msg;
267
268          msg.data = new byte[messageLength];
269          memcpy( msg.data, data + i, messageLength );
270          msg.length = messageLength;
271          msg.messageType = (MessageType)messageType;
272          msg.number = userId;
273          msg.senderId = senderId;
274          msg.recieverType = (RecieverType)recieverType;
275          msg.destinationId = destinationId;
276
277          incomingMessageQueue.push_back( msg );
278        }
279      }
280
281
282      // check if the message needs to be forwarded
283      if( recieverType == RT_ALL_BUT_ME ||
284          recieverType == RT_ALL_ME ||
285          recieverType == RT_NOT_USER ||
286          recieverType == RT_USER  && SharedNetworkData::getInstance()->getHostID() != destinationId ||
287          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isProxyServerActive() )
288      {
289        // forwarding the messages but only if its a proxy
290        if( SharedNetworkData::getInstance()->isProxyServerActive())
291        {
292          PRINTF(0)("===========>> Forwarding Message msg with type: %i, from sender %i, to rec: %i\n", messageType, senderId, destinationId);
293          NetworkMessage msg;
294
295          msg.data = new byte[messageLength];
296          memcpy( msg.data, data + i, messageLength );
297          msg.length = messageLength;
298          msg.messageType = (MessageType)messageType;
299          msg.number = userId;
300          msg.senderId = senderId;
301          msg.destinationId = destinationId;
302          msg.recieverType = (RecieverType)recieverType;
303
304          this->sendMessage(msg.messageType, msg.data, msg.length, msg.recieverType, msg.senderId = senderId, msg.destinationId, MP_HIGHBANDWIDTH);
305        }
306      }
307
308      // save the serial number for ack signaling
309      outgoingMessageQueue[userId].recievedMessages.push_back( number );
310    }
311
312    i += messageLength;
313  }
314
315
316  //walk throu message queue and remove acked messages
317  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end();  )
318  {
319    if ( std::find( acks.begin(), acks.end(), it->number) != acks.end() )
320    {
321      std::list<NetworkMessage>::iterator delIt = it;
322      it++;
323      outgoingMessageQueue[userId].messages.erase( delIt );
324      continue;
325    }
326    it++;
327  }
328
329  //TODO find bether way. maybe with timestamp
330  if ( outgoingMessageQueue[userId].recievedMessages.size() > 1000 )
331  {
332    for ( int j = 0; j < (int)outgoingMessageQueue[userId].recievedMessages.size() - 1000; j++ )
333      outgoingMessageQueue[userId].recievedMessages.erase( outgoingMessageQueue[userId].recievedMessages.begin() );
334  }
335
336  return i;
337}
338
339
340
341
342/**
343 * processes the message manager data, specialy check for localy generated messages
344 */
345void MessageManager::processData()
346{
347  // now call the message handlers with the new message
348  for ( std::list<NetworkMessage>::iterator it = incomingMessageQueue.begin(); it != incomingMessageQueue.end();  )
349  {
350    PRINTF(0)("<<< MessageManager: got local msg with type: %i, from sender %i, to rec: %i\n", (*it).messageType, (*it).senderId, (*it).destinationId);
351
352    if ( (*(messageHandlerMap[it->messageType].cb))( it->messageType, it->data, it->length, messageHandlerMap[it->messageType].someData,
353                                                     /*it->number, */it->senderId, it->destinationId ) )
354    {
355      std::list<NetworkMessage>::iterator delIt = it;
356      if ( it->data )
357        delete it->data;
358      it++;
359      incomingMessageQueue.erase( delIt );
360      continue;
361    }
362    it++;
363  }
364
365}
366
367
368
369
370/**
371 * clean up memory reserved for user
372 * @param userId userid
373 */
374void MessageManager::cleanUpUser( int userId )
375{
376  if ( outgoingMessageQueue.find( userId ) == outgoingMessageQueue.end() )
377    return;
378
379  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end(); it++ )
380  {
381    if ( it->data )
382      delete it->data;
383    it->data = NULL;
384  }
385
386  outgoingMessageQueue[userId].toAck.clear();
387
388  outgoingMessageQueue.erase( userId );
389}
390
391/**
392 * registers function to handle messages with id messageType. someData is passed to callbackfuntion
393 * @param messageType message id to handle
394 * @param cb function pointer to callback function
395 * @param someData this pointer is passed to callback function without modification
396 * @return true on success
397 */
398bool MessageManager::registerMessageHandler( MessageType messageType, MessageCallback cb, void * someData )
399{
400  MessageHandler messageHandler;
401
402  messageHandler.cb = cb;
403  messageHandler.messageType = messageType;
404  messageHandler.someData = someData;
405
406  messageHandlerMap[messageType] = messageHandler;
407
408  return true;
409}
410
411/**
412 * initializes buffers for user
413 * @param userId userId
414 */
415void MessageManager::initUser( int userId )
416{
417  // just do something so map creates a new entry
418  outgoingMessageQueue[userId].toAck.clear();
419  //assert( outgoingMessageQueue[userId].messages.size() == 0 );
420}
421
422
423
424/**
425 * send a message to one or more clients
426 * recieverType:
427 *               RT_ALL send to all users. reciever is ignored
428 *               RT_USER send only to reciever
429 *               RT_NOT_USER send to all but reciever
430 *
431 * @param messageType message id
432 * @param data pointer to data
433 * @param dataLength length of data
434 * @param recieverType type of the receiver
435 * @param reciever the userId of the receiver if needed (depends on the ReceiverType)
436 */
437void MessageManager::sendMessage( MessageType messageType, byte * data, int dataLength, RecieverType recieverType, int reciever, MessagePriority messagePriority )
438{
439  this->sendMessage(messageType, data, dataLength, recieverType, SharedNetworkData::getInstance()->getHostID(), reciever, messagePriority);
440}
441
442
443/**
444 * send a message to one or more clients as a special client
445 * recieverType:
446 *               RT_ALL send to all users. reciever is ignored
447 *               RT_USER send only to reciever
448 *               RT_NOT_USER send to all but reciever
449 *
450 * @param messageType message id
451 * @param data pointer to data
452 * @param dataLength length of data
453 * @param recieverType type of the receiver
454 * @param sender the userId of the sender if there is need for shadowing it (eg. for msg forwarding)
455 * @param reciever the userId of the receiver if needed (depends on the ReceiverType)
456 */
457    void MessageManager::sendMessage( MessageType messageType, byte * data, int dataLength, RecieverType recieverType, int sender, int reciever, MessagePriority messagePriority )
458{
459  PRINTF(0)(" >>> MessageManager: sending msg with type: %i, recieverType: %i, reciever %i\n", messageType, recieverType, reciever);
460
461  // go through all outgoing message queues and add the message if its appropriate
462  for ( MessageQueue::iterator it = this->outgoingMessageQueue.begin(); it != this->outgoingMessageQueue.end(); it++ )
463  {
464
465    if (
466         recieverType == RT_ALL_ME      ||
467         recieverType == RT_ALL_BUT_ME  ||
468         recieverType == RT_USER        && it->first == reciever ||
469         recieverType == RT_USER        && reciever == NET_ID_MASTER_SERVER && !getNetworkStream()->isUserMasterServer( it->first ) ||  //(*)
470         recieverType == RT_NOT_USER    && it->first != reciever ||
471         recieverType == RT_SERVER      && getNetworkStream()->isUserMasterServer( it->first ) ||
472         recieverType == RT_SERVER      && getNetworkStream()->isUserProxyServerActive( it->first )
473       )// (*) special case: forward
474    {
475      NetworkMessage msg;
476
477      msg.data = new byte[dataLength];
478      memcpy( msg.data, data, dataLength );
479      msg.length = dataLength;
480      msg.messageType = messageType;
481      msg.number = this->newNumber++;
482      msg.senderId = sender;
483      msg.destinationId = reciever;
484      msg.recieverType = recieverType;
485      msg.priority = messagePriority;
486
487      it->second.messages.push_back( msg );
488    }
489
490
491  }
492
493
494  // if the message is also for myself, handle it here
495  if ( recieverType == RT_ALL_ME ||
496       recieverType == RT_USER   && reciever == SharedNetworkData::getInstance()->getHostID()
497     )
498  {
499    NetworkMessage msg;
500
501    msg.data = new byte[dataLength];
502    memcpy( msg.data, data, dataLength );
503    msg.length = dataLength;
504    msg.messageType = messageType;
505    msg.number = SharedNetworkData::getInstance()->getHostID();
506    msg.senderId = sender;
507    msg.destinationId = reciever;
508    msg.recieverType = recieverType;
509    msg.priority = messagePriority;
510
511    this->incomingMessageQueue.push_back( msg );
512  }
513}
514
515
Note: See TracBrowser for help on using the repository browser.