Changeset 9656 in orxonox.OLD for trunk/src/lib/network/message_manager.cc

Timestamp:

Aug 4, 2006, 11:01:28 PM (18 years ago)

Author:

bensch

Message:

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

File:

• trunk/src/lib/network/message_manager.cc (modified) (17 diffs)

trunk/src/lib/network/message_manager.cc

@@ -11 +11 @@
    ### File Specific:
    main-programmer: Christoph Renner
-   co-programmer: ...
+   co-programmer: Patrick Boenzli (patrick@orxonox.ethz.ch)
+
+     June 2006: finishing work on the network stream for pps presentation (rennerc@ee.ethz.ch)
+     July 2006: some code rearangement and integration of the proxy server mechanism (boenzlip@ee.ethz.ch)
+     July 2006: message forwarding algorithms
 */
 
@@ -44 +48 @@
 MessageManager::~MessageManager ()
 {
-  for ( MessageQueue::iterator it = messageQueue.begin(); it != messageQueue.end(); it++ )
+  for ( MessageQueue::iterator it = outgoingMessageQueue.begin(); it != outgoingMessageQueue.end(); it++ )
   {
     for ( std::list<NetworkMessage>::iterator it2 = it->second.messages.begin(); it2 != it->second.messages.end(); it2++ )
@@ -59 +63 @@
   }
 
-  messageQueue.clear();
+  outgoingMessageQueue.clear();
 
   this->messageHandlerMap.clear();
@@ -85 +89 @@
   int n;
 
-  n = Converter::intToByteArray( messageQueue[userId].toAck.size(), data + i, maxLength );
+  n = Converter::intToByteArray( outgoingMessageQueue[userId].toAck.size(), data + i, maxLength );
   i += n;
   assert( n == INTSIZE );
 
-  for ( std::list<int>::iterator it = messageQueue[userId].toAck.begin(); it != messageQueue[userId].toAck.end(); it++)
+  for ( std::list<int>::iterator it = outgoingMessageQueue[userId].toAck.begin(); it != outgoingMessageQueue[userId].toAck.end(); it++)
   {
     n = Converter::intToByteArray( *it, data + i, maxLength );
@@ -96 +100 @@
   }
 
-  messageQueue[userId].toAck.clear();
-
-  n = Converter::intToByteArray( messageQueue[userId].messages.size(), data + i, maxLength );
+  outgoingMessageQueue[userId].toAck.clear();
+
+  n = Converter::intToByteArray( outgoingMessageQueue[userId].messages.size(), data + i, maxLength );
   i += n;
   assert( n == INTSIZE );
 
-  for ( std::list<NetworkMessage>::iterator it = messageQueue[userId].messages.begin(); it != messageQueue[userId].messages.end(); it++ )
-  {
+  // write the message down, a message has this structure:
+  //   | data_length | serial_number | message_type | source_id | dest_id | ...data... |
+  //      4byte        4byte            4byte         4byte      4byte     data_length
+  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end(); it++ )
+  {
+    // send data length
     n = Converter::intToByteArray( it->length, data + i, maxLength );
     i += n;
     assert( n == INTSIZE );
 
+    // send serial number
     n = Converter::intToByteArray( it->number, data + i, maxLength );
     i += n;
     assert( n == INTSIZE );
 
-    n = Converter::intToByteArray( it->messageId, data + i, maxLength );
-    i += n;
-    assert( n == INTSIZE );
-
+    // send message type
+    n = Converter::intToByteArray( it->messageType, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+
+    // send sender id
+    n = Converter::intToByteArray( it->senderId, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+
+    // send destination id
+    n = Converter::intToByteArray( it->destinationId, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+
+    // send receiver type
+    n = Converter::intToByteArray( it->recieverType, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+
+    // and copy the data
     assert( i + it->length <= maxLength );
     memcpy( data + i, it->data, it->length );
@@ -141 +167 @@
   int nAcks;
 
+
   assert( i + INTSIZE <= length );
   n = Converter::byteArrayToInt( data + i, &nAcks );
@@ -167 +194 @@
   i += n;
 
-  int messageLength, messageId;
-
+  int messageLength, messageType;
+  int senderId, destinationId, recieverType;
+
+  // now go through all newly received messages and assemble them
   for ( int j = 0; j < nMessages; j++ )
   {
+    // read the length
     assert( i + INTSIZE <= length );
     n = Converter::byteArrayToInt( data + i, &messageLength );
@@ -176 +206 @@
     i += n;
 
+    // read the serial number
     assert( i + INTSIZE <= length );
     n = Converter::byteArrayToInt( data + i, &number );
@@ -181 +212 @@
     i += n;
 
-    assert( i + INTSIZE <= length );
-    n = Converter::byteArrayToInt( data + i, &messageId );
+    // read the message type
+    assert( i + INTSIZE <= length );
+    n = Converter::byteArrayToInt( data + i, &messageType );
+    assert( n == INTSIZE );
+    i += n;
+
+    // read the sender id
+    assert( i + INTSIZE <= length );
+    n = Converter::byteArrayToInt( data + i, &senderId );
+    assert( n == INTSIZE );
+    i += n;
+
+    //read the destination id
+    assert( i + INTSIZE <= length );
+    n = Converter::byteArrayToInt( data + i, &destinationId);
+    assert( n == INTSIZE );
+    i += n;
+
+    // read the receiver type
+    assert( i + INTSIZE <= length );
+    n = Converter::byteArrayToInt( data + i, &recieverType);
     assert( n == INTSIZE );
     i += n;
 
     if ( number > 0 )
-      messageQueue[userId].toAck.push_back( number );
-
+      outgoingMessageQueue[userId].toAck.push_back( number );
+
+//     PRINTF(0)("got message with type: %i\n", messageType);
     assert( i + messageLength <= length );
-    assert( messageHandlerMap.find( (MessageId)messageId ) != messageHandlerMap.end() );
-    if ( std::find( messageQueue[userId].recievedMessages.begin(), messageQueue[userId].recievedMessages.end(), number )== messageQueue[userId].recievedMessages.end() )
+    // make sure there is a message handler for this message type
+    assert( messageHandlerMap.find( (MessageType)messageType ) != messageHandlerMap.end());
+
+
+    if ( std::find( outgoingMessageQueue[userId].recievedMessages.begin(), outgoingMessageQueue[userId].recievedMessages.end(), number ) ==
+         outgoingMessageQueue[userId].recievedMessages.end() )
     {
-      if ( !(*(messageHandlerMap[(MessageId)messageId].cb))( (MessageId)messageId, data + i, messageLength, messageHandlerMap[(MessageId)messageId].someData, userId ) )
+
+      // find out if this message is addressed for this client too
+      if( recieverType == RT_ALL_BUT_ME  && SharedNetworkData::getInstance()->getHostID() != senderId ||
+          recieverType == RT_ALL_ME ||
+          recieverType == RT_NOT_USER && SharedNetworkData::getInstance()->getHostID() != destinationId ||
+          recieverType == RT_USER  && SharedNetworkData::getInstance()->getHostID() == destinationId ||
+          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isMasterServer() ||
+          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isProxyServerActive())
       {
-        NetworkMessage msg;
-
-        msg.data = new byte[messageLength];
-        memcpy( msg.data, data + i, messageLength );
-        msg.length = messageLength;
-        msg.messageId = (MessageId)messageId;
-        msg.number = userId;
-
-        incomingMessageBuffer.push_back( msg );
+
+        PRINTF(0)("<<< MessageManager: got msg with type: %i, from sender %i, to rec: %i\n", messageType, senderId, destinationId);
+      // call the handler function and handle errors
+        if ( !(*(messageHandlerMap[(MessageType)messageType].cb))( (MessageType)messageType, data + i, messageLength,
+                 messageHandlerMap[(MessageType)messageType].someData, senderId, destinationId ) )
+        {
+        // if the message is not handled correctly, bush it back to the incoming packets therefore trying it later
+          NetworkMessage msg;
+
+          msg.data = new byte[messageLength];
+          memcpy( msg.data, data + i, messageLength );
+          msg.length = messageLength;
+          msg.messageType = (MessageType)messageType;
+          msg.number = userId;
+          msg.senderId = senderId;
+          msg.recieverType = (RecieverType)recieverType;
+          msg.destinationId = destinationId;
+
+          incomingMessageQueue.push_back( msg );
+        }
       }
-      messageQueue[userId].recievedMessages.push_back( number );
+
+
+      // check if the message needs to be forwarded
+      if( recieverType == RT_ALL_BUT_ME ||
+          recieverType == RT_ALL_ME ||
+          recieverType == RT_NOT_USER ||
+          recieverType == RT_USER  && SharedNetworkData::getInstance()->getHostID() != destinationId ||
+          recieverType == RT_SERVER && SharedNetworkData::getInstance()->isProxyServerActive() )
+      {
+        // forwarding the messages but only if its a proxy
+        if( SharedNetworkData::getInstance()->isProxyServerActive())
+        {
+          PRINTF(0)("===========>> Forwarding Message msg with type: %i, from sender %i, to rec: %i\n", messageType, senderId, destinationId);
+          NetworkMessage msg;
+
+          msg.data = new byte[messageLength];
+          memcpy( msg.data, data + i, messageLength );
+          msg.length = messageLength;
+          msg.messageType = (MessageType)messageType;
+          msg.number = userId;
+          msg.senderId = senderId;
+          msg.destinationId = destinationId;
+          msg.recieverType = (RecieverType)recieverType;
+
+          this->sendMessage(msg.messageType, msg.data, msg.length, msg.recieverType, msg.senderId = senderId, msg.destinationId, MP_HIGHBANDWIDTH);
+        }
+      }
+
+      // save the serial number for ack signaling
+      outgoingMessageQueue[userId].recievedMessages.push_back( number );
     }
+
     i += messageLength;
   }
 
 
-  //TODO maybe handle incomingMessage in tick function. else local messages will not be handled if no clients are connected
-  for ( std::list<NetworkMessage>::iterator it = incomingMessageBuffer.begin(); it != incomingMessageBuffer.end();  )
-  {
-    if ( (*(messageHandlerMap[it->messageId].cb))( it->messageId, it->data, it->length, messageHandlerMap[it->messageId].someData, it->number ) )
+  //walk throu message queue and remove acked messages
+  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end();  )
+  {
+    if ( std::find( acks.begin(), acks.end(), it->number) != acks.end() )
+    {
+      std::list<NetworkMessage>::iterator delIt = it;
+      it++;
+      outgoingMessageQueue[userId].messages.erase( delIt );
+      continue;
+    }
+    it++;
+  }
+
+  //TODO find bether way. maybe with timestamp
+  if ( outgoingMessageQueue[userId].recievedMessages.size() > 1000 )
+  {
+    for ( int j = 0; j < (int)outgoingMessageQueue[userId].recievedMessages.size() - 1000; j++ )
+      outgoingMessageQueue[userId].recievedMessages.erase( outgoingMessageQueue[userId].recievedMessages.begin() );
+  }
+
+  return i;
+}
+
+
+
+
+/**
+ * processes the message manager data, specialy check for localy generated messages
+ */
+void MessageManager::processData()
+{
+  // now call the message handlers with the new message
+  for ( std::list<NetworkMessage>::iterator it = incomingMessageQueue.begin(); it != incomingMessageQueue.end();  )
+  {
+    PRINTF(0)("<<< MessageManager: got local msg with type: %i, from sender %i, to rec: %i\n", (*it).messageType, (*it).senderId, (*it).destinationId);
+
+    if ( (*(messageHandlerMap[it->messageType].cb))( it->messageType, it->data, it->length, messageHandlerMap[it->messageType].someData,
+                                                     /*it->number, */it->senderId, it->destinationId ) )
     {
       std::list<NetworkMessage>::iterator delIt = it;
@@ -220 +357 @@
         delete it->data;
       it++;
-      incomingMessageBuffer.erase( delIt );
+      incomingMessageQueue.erase( delIt );
       continue;
     }
@@ -226 +363 @@
   }
 
-  //walk throu message queue and remove acked messages
-  for ( std::list<NetworkMessage>::iterator it = messageQueue[userId].messages.begin(); it != messageQueue[userId].messages.end();  )
-  {
-    if ( std::find( acks.begin(), acks.end(), it->number) != acks.end() )
-    {
-      std::list<NetworkMessage>::iterator delIt = it;
-      it++;
-      messageQueue[userId].messages.erase( delIt );
-      continue;
-    }
-    it++;
-  }
-
-  //TODO find bether way. maybe with timestamp
-  if ( messageQueue[userId].recievedMessages.size() > 1000 )
-  {
-    for ( int j = 0; j < messageQueue[userId].recievedMessages.size() - 1000; j++ )
-      messageQueue[userId].recievedMessages.erase( messageQueue[userId].recievedMessages.begin() );
-  }
-
-  return i;
-}
+}
+
+
+
 
 /**
@@ -255 +374 @@
 void MessageManager::cleanUpUser( int userId )
 {
-  if ( messageQueue.find( userId ) == messageQueue.end() )
+  if ( outgoingMessageQueue.find( userId ) == outgoingMessageQueue.end() )
     return;
 
-  for ( std::list<NetworkMessage>::iterator it = messageQueue[userId].messages.begin(); it != messageQueue[userId].messages.end(); it++ )
+  for ( std::list<NetworkMessage>::iterator it = outgoingMessageQueue[userId].messages.begin(); it != outgoingMessageQueue[userId].messages.end(); it++ )
   {
     if ( it->data )
@@ -265 +384 @@
   }
 
-  messageQueue[userId].toAck.clear();
-
-  messageQueue.erase( userId );
-}
-
-/**
- * registers function to handle messages with id messageId. someData is passed to callbackfuntion
- * @param messageId message id to handle
+  outgoingMessageQueue[userId].toAck.clear();
+
+  outgoingMessageQueue.erase( userId );
+}
+
+/**
+ * registers function to handle messages with id messageType. someData is passed to callbackfuntion
+ * @param messageType message id to handle
  * @param cb function pointer to callback function
  * @param someData this pointer is passed to callback function without modification
  * @return true on success
  */
-bool MessageManager::registerMessageHandler( MessageId messageId, MessageCallback cb, void * someData )
+bool MessageManager::registerMessageHandler( MessageType messageType, MessageCallback cb, void * someData )
 {
   MessageHandler messageHandler;
 
   messageHandler.cb = cb;
-  messageHandler.messageId = messageId;
+  messageHandler.messageType = messageType;
   messageHandler.someData = someData;
 
-  messageHandlerMap[messageId] = messageHandler;
+  messageHandlerMap[messageType] = messageHandler;
 
   return true;
@@ -297 +416 @@
 {
   // just do something so map creates a new entry
-  messageQueue[userId].toAck.clear();
-  //assert( messageQueue[userId].messages.size() == 0 );
-}
+  outgoingMessageQueue[userId].toAck.clear();
+  //assert( outgoingMessageQueue[userId].messages.size() == 0 );
+}
+
+
 
 /**
@@ -308 +429 @@
  *               RT_NOT_USER send to all but reciever
  *
- * @param messageId message id
+ * @param messageType message id
  * @param data pointer to data
  * @param dataLength length of data
- * @param recieverType
- * @param reciever
- */
-void MessageManager::sendMessage( MessageId messageId, byte * data, int dataLength, RecieverType recieverType, int reciever, MessagePriority messagePriority )
-{
-  for ( MessageQueue::iterator it = messageQueue.begin(); it != messageQueue.end(); it++ )
-  {
+ * @param recieverType type of the receiver
+ * @param reciever the userId of the receiver if needed (depends on the ReceiverType)
+ */
+void MessageManager::sendMessage( MessageType messageType, byte * data, int dataLength, RecieverType recieverType, int reciever, MessagePriority messagePriority )
+{
+  this->sendMessage(messageType, data, dataLength, recieverType, SharedNetworkData::getInstance()->getHostID(), reciever, messagePriority);
+}
+
+
+/**
+ * send a message to one or more clients as a special client
+ * recieverType:
+ *               RT_ALL send to all users. reciever is ignored
+ *               RT_USER send only to reciever
+ *               RT_NOT_USER send to all but reciever
+ *
+ * @param messageType message id
+ * @param data pointer to data
+ * @param dataLength length of data
+ * @param recieverType type of the receiver
+ * @param sender the userId of the sender if there is need for shadowing it (eg. for msg forwarding)
+ * @param reciever the userId of the receiver if needed (depends on the ReceiverType)
+ */
+    void MessageManager::sendMessage( MessageType messageType, byte * data, int dataLength, RecieverType recieverType, int sender, int reciever, MessagePriority messagePriority )
+{
+  PRINTF(0)(" >>> MessageManager: sending msg with type: %i, recieverType: %i, reciever %i\n", messageType, recieverType, reciever);
+
+  // go through all outgoing message queues and add the message if its appropriate
+  for ( MessageQueue::iterator it = this->outgoingMessageQueue.begin(); it != this->outgoingMessageQueue.end(); it++ )
+  {
+
     if (
-         recieverType == RT_ALL_ME ||
-         recieverType == RT_ALL_BUT_ME ||
-         recieverType == RT_USER && it->first == reciever ||
-         recieverType == RT_NOT_USER && it->first != reciever ||
-         recieverType == RT_SERVER && getNetworkStream()->isUserMasterServer( it->first ) ||
-         recieverType == RT_SERVER && getNetworkStream()->isUserProxyServerActive( it->first )
-       )
+         recieverType == RT_ALL_ME      ||
+         recieverType == RT_ALL_BUT_ME  ||
+         recieverType == RT_USER        && it->first == reciever ||
+         recieverType == RT_USER        && reciever == NET_ID_MASTER_SERVER && !getNetworkStream()->isUserMasterServer( it->first ) ||  //(*)
+         recieverType == RT_NOT_USER    && it->first != reciever ||
+         recieverType == RT_SERVER      && getNetworkStream()->isUserMasterServer( it->first ) ||
+         recieverType == RT_SERVER      && getNetworkStream()->isUserProxyServerActive( it->first )
+       )// (*) special case: forward
     {
       NetworkMessage msg;
@@ -332 +478 @@
       memcpy( msg.data, data, dataLength );
       msg.length = dataLength;
-      msg.messageId = messageId;
-      msg.number = newNumber++;
+      msg.messageType = messageType;
+      msg.number = this->newNumber++;
+      msg.senderId = sender;
+      msg.destinationId = reciever;
+      msg.recieverType = recieverType;
       msg.priority = messagePriority;
 
       it->second.messages.push_back( msg );
     }
-  }
-
-  if ( recieverType == RT_ALL_ME )
+
+
+  }
+
+
+  // if the message is also for myself, handle it here
+  if ( recieverType == RT_ALL_ME ||
+       recieverType == RT_USER   && reciever == SharedNetworkData::getInstance()->getHostID()
+     )
   {
     NetworkMessage msg;
@@ -347 +502 @@
     memcpy( msg.data, data, dataLength );
     msg.length = dataLength;
-    msg.messageId = messageId;
+    msg.messageType = messageType;
     msg.number = SharedNetworkData::getInstance()->getHostID();
+    msg.senderId = sender;
+    msg.destinationId = reciever;
+    msg.recieverType = recieverType;
     msg.priority = messagePriority;
 
-    incomingMessageBuffer.push_back( msg );
-  }
-}
-
-
+    this->incomingMessageQueue.push_back( msg );
+  }
+}
+
+