Changeset 7671 in orxonox.OLD

Timestamp:

May 18, 2006, 10:48:17 AM (18 years ago)

Author:

rennerc

Message:

started implementing MessageManager

Location:

branches/network/src/lib/network

Files:

• message_manager.cc (modified) (6 diffs)
• message_manager.h (modified) (1 diff)
• network_stream.cc (modified) (4 diffs)
• network_stream.h (modified) (1 diff)

branches/network/src/lib/network/message_manager.cc

@@ -20 +20 @@
 using namespace std;
 
+MessageManager* MessageManager::singletonRef = NULL;
+
 
 /**
@@ -35 +37 @@
 MessageManager::~MessageManager ()
 {
+  for ( MessageQueue::iterator it = messageQueue.begin(); it != messageQueue.end(); it++ )
+  {
+    for ( std::list<NetworkMessage>::iterator it2 = it->second.messages.begin(); it2 != it->second.messages.end(); it2++ )
+    {
+      if ( it2->data )
+      {
+        delete it2->data;
+        it2->data = NULL;
+      }
+    }
+    
+    it->second.messages.clear();
+    it->second.toAck.clear();
+  }
+  
+  messageQueue.clear();
 }
 
@@ -40 +58 @@
  * get the diff to last acked state of userId
  *
- * each synchrinizeable defines its own stack of states received and sent over the network. The stack contains
- * a per user entry for the last sent/received state This function returns a delta compressed state of the
- * synchronizeable. This state will be transmitted over the network to the other participants
+ * this class does not use the normal SynchronizeableVars for zynchronisation. instead
+ * it defines its own protocol
  *
  * @param userId user to create diff for
@@ -54 +71 @@
 int MessageManager::getStateDiff( int userId, byte * data, int maxLength, int stateId, int fromStateId, int priorityTH )
 {
+  int i = 0;
+  int n;
+  
+  n = Converter::intToByteArray( messageQueue[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++)
+  {
+    n = Converter::intToByteArray( *it, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+  }
+  
+  messageQueue[userId].toAck.clear();
+  
+  n = Converter::intToByteArray( messageQueue[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++ )
+  {
+    n = Converter::intToByteArray( it->length, data + i, maxLength );
+    i += n;
+    assert( n == INTSIZE );
+    
+    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 );
+    
+    assert( i + it->length <= maxLength );
+    memcpy( data + i, it->data, it->length );
+    i += it->length;
+  }
+  
+  return i;
 }
 
@@ -68 +125 @@
 int MessageManager::setStateDiff( int userId, byte * data, int length, int stateId, int fromStateId )
 {
+  int i = 0;
+  int n;
+  
+  int nAcks;
+  
+  n = Converter::byteArrayToInt( data + i, &nAcks );
+  assert( n == INTSIZE );
+  i += n;
+  
+  std::list<int> acks;
+  
+  int number;
+  
+  for ( int j = 0; j < nAcks; j++ )
+  {
+    n = Converter::byteArrayToInt( data + i, &number );
+    assert( n == INTSIZE );
+    i += n;
+    
+    acks.push_back( number );
+  }
+  
+  int nMessages;
+  
+  n = Converter::byteArrayToInt( data + i, &nMessages );
+  assert( n == INTSIZE );
+  i += n;
+  
+  for ( int j = 0; j < nMessages; j++ )
+  {
+  }
 }
 
@@ -77 +165 @@
 {
 }
+
+/**
+ * registers function to handle messages with id messageId. someData is passed to callbackfuntion
+ * @param messageId 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 )
+{
+}
+
+/**
+ * initializes buffers for user
+ * @param userId userId
+ */
+void MessageManager::initUser( int userId )
+{
+}

branches/network/src/lib/network/message_manager.h

@@ -9 +9 @@
 #include "synchronizeable.h"
 
-// FORWARD DECLARATION
+#include <map>
+#include <list>
+
+/*
+  Protocol:
+    int nacks
+    int acks[1..nacks]
+    int nmsg
+    (
+      int length
+      int number
+      byte MessageId
+      byte * data
+    )[1..nmsg]
+*/
+
+typedef void (*MessageCallback)( byte * data, int dataLength, void * someData, int userId );
+
+
+enum MessageId 
+{
+};
+
+enum RecieverType
+{
+  RT_ALL = 1,   //!< message is sent to all users
+  RT_USER,      //!< message is only sent to reciever
+  RT_NOT_USER   //!< message is sent to all but reciever
+};
+
+struct NetworkMessage
+{
+  MessageId messageId;
+  byte *    data;
+  int       length;
+  int       number;
+};
+
+struct MessageUserQueue
+{
+  std::list<NetworkMessage> messages;
+  std::list<int>            toAck;
+};
+
+typedef std::map<int,MessageUserQueue> MessageQueue;
+
+struct MessageHandler
+{
+  MessageCallback cb;
+  MessageId       messageId;
+};
+
+typedef std::map<MessageId,MessageHandler> MessageHandlerMap;
 
 //! A class for sending messages over network
 class MessageManager : public Synchronizeable {
-
+ protected:
+   MessageManager();
  public:
-   MessageManager();
+   inline static MessageManager * getInstance(){ if (!singletonRef) singletonRef = new MessageManager();  return singletonRef; }
+   
    virtual ~MessageManager();
+   
+   bool registerMessageHandler( MessageId messageId, MessageCallback cb, void * someData );
+   
+   void sendMessage( MessageId messageId, byte * data, int dataLength, RecieverType recieverType, int reciever );
 
    virtual int getStateDiff( int userId, byte* data, int maxLength, int stateId, int fromStateId, int priorityTH );
    virtual int setStateDiff( int userId, byte* data, int length, int stateId, int fromStateId );
    virtual void cleanUpUser( int userId );
+   
+   void initUser( int userId );
 
 
  private:
+   static MessageManager * singletonRef;
+   MessageQueue            messageQueue;        //!< stores messages to send
+   MessageHandlerMap       messageHandlerMap;   //!< contains handlers for messages
 
 };

branches/network/src/lib/network/network_stream.cc

@@ -29 +29 @@
 #include "network_game_manager.h"
 #include "shared_network_data.h"
+#include "message_manager.h"
 
 #include "lib/util/loading/factory.h"
@@ -199 +200 @@
 
   handleHandshakes();
+  handleDownstream();
   handleUpstream();
-  handleDownstream();
-
-
-
-  /* DOWNSTREAM */
-#if 0
-
-
-  int dataLength;
-  int reciever;
-  Header header;
-  int counter;
-
-  for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
-  {
-    counter = 0;
-
-    if ( (*it)!=NULL && (*it)->beSynchronized() /*&& (*it)->getOwner() == myHostId*/ )
-    {
-      do {
-        counter++;
-
-        reciever = 0;
-#warning fix this
-dataLength = 0;
-//TODO fix
-        //dataLength = (*it)->readBytes(downBuffer, DATA_STREAM_BUFFER_SIZE, &reciever);
-
-        if ( dataLength<=0 ){
-          reciever = 0;
-          continue;
-        }
-
-        dataLength = networkProtocol->createHeader((byte*)downBuffer, dataLength, DATA_STREAM_BUFFER_SIZE, static_cast<const Synchronizeable&>(*(*it)));
-
-        Header* header = (Header*)downBuffer;
-        if ( header->synchronizeableID < this->maxConnections+2 )
-        {
-          //if ( !isServer() ) PRINTF(0)("RESET UNIQUEID FROM %d TO 0 maxCon=%d\n", header->synchronizeableID, this->maxConnections);
-          header->synchronizeableID = 0;
-        }
-        else
-        {
-          //if ( !isServer() ) PRINTF(0)("UNIQUEID=%d\n", header->synchronizeableID);
-        }
-
-        if ( dataLength<=0 )
-          continue;
-
-        if ( reciever!=0 )
-        {
-          if ( reciever < 0)
-          {
-            for ( int i = 0; i<networkSockets.size(); i++)
-            {
-              if ( i!=abs(reciever) && networkSockets[i] != NULL )
-              {
-                PRINTF(0)("write %d bytes to socket %d uniqueid %d reciever %d\n", dataLength, i, (*it)->getUniqueID(), reciever);
-                networkSockets[i]->writePacket(downBuffer, dataLength);
-              }
-            }
-          }
-          else
-          {
-            if ( networkSockets[reciever] != NULL )
-            {
-              PRINTF(5)("write %d bytes to socket %d\n", dataLength, reciever);
-              networkSockets[reciever]->writePacket(downBuffer, dataLength);
-            }
-            else
-            {
-              PRINTF(1)("networkSockets[reciever] == NULL\n");
-            }
-          }
-        }
-        else
-        {
-          for ( int i = 0; i<networkSockets.size(); i++)
-          {
-            if ( networkSockets[i] != NULL )
-            {
-              PRINTF(5)("write %d bytes to socket %d\n", dataLength, i);
-              networkSockets[i]->writePacket(downBuffer, dataLength);
-            }
-          }
-        }
-
-      } while( reciever!=0 );
-    }
-  }
-
-  /* UPSTREAM */
-
-  for ( int i = 0; i<networkSockets.size(); i++)
-  {
-    if ( networkSockets[i] )
-    {
-      do {
-        dataLength = networkSockets[i]->readPacket(upBuffer, DATA_STREAM_BUFFER_SIZE);
-
-        if ( dataLength<=0 )
-          continue;
-
-        header = networkProtocol->extractHeader(upBuffer, dataLength);
-        dataLength -= sizeof(header);
-
-        PRINTF(5)("read %d bytes from socket uniqueID = %d\n", dataLength, header.synchronizeableID);
-
-        if ( dataLength != header.length )
-        {
-          PRINTF(1)("packetsize in header and real packetsize do not match! %d:%d\n", dataLength, header.length);
-          continue;
-        }
-
-        if ( header.synchronizeableID == 0 )
-        {
-          header.synchronizeableID = i;
-        }
-
-        for (SynchronizeableList::iterator it = synchronizeables.begin(); it!=synchronizeables.end(); it++)
-        {
-#warning fix this
-
-          if ( *it && (*it)->getUniqueID()==header.synchronizeableID )
-          {
-            if ( (*it)->writeBytes(upBuffer+sizeof(header), dataLength, i) != header.length )
-            {
-              PRINTF(1)("%s did not read all the data id = %d!\n", (*it)->getClassName(), (*it)->getUniqueID());
-              break;
-            }
-            continue;
-          }
-
-        }
-
-      } while ( dataLength>0 );
-    }
-
-  }
-#endif
+
 }
 
@@ -476 +339 @@
           delete it->second.handshake;
           it->second.handshake = NULL;
+          
+          handleNewClient( it->second.userId );
         }
         else
@@ -699 +564 @@
 }
 
-
-
-
-
-
+/**
+ * is executed when a handshake has finished
+ * @todo create playable for new user
+ */
+void NetworkStream::handleNewClient( int userId )
+{
+  MessageManager::getInstance()->initUser( userId );
+}
+
+
+
+
+
+

branches/network/src/lib/network/network_stream.h

@@ -81 +81 @@
     void handleUpstream();
     void handleDownstream(); 
+    void handleNewClient( int userId );