MessagePump.cpp

Go to the documentation of this file.

#include "MessagePump.h"

#include "ClientLoginMessage.h"
#include "IncommingConnectionMessage.h"
#include "GetClientCharacterListMessage.h"
#include "CreateCharacterMessage.h"
#include "JoinChannelMessage.h"

namespace SraNetwork
{
    // The idea is ok, but we need to properly delete all classes
    // when destroying the msg pump so we need to store the instances.
    // Processor function definition:
    // typedef void (SraBaseMessage::*process)(SraPacket*, RakNet::SystemAddress*);
    
    // A map of SraNetwork::MessageIdentifier->BaseMessageImpl
    tbb::concurrent_hash_map<int, SraBaseMessage*>  MessagePump::messageProcessorTable;

    // The current list of tasks.
    MessagePump* MessagePump::Instance = new MessagePump();

    bool MessagePump::isRunning = true;

    tbb::concurrent_queue<MessageData> MessagePump::messageList;

    // Creates a new message pump and initalizes the processor tasks.
    MessagePump::MessagePump(void)
    {
        // Add all message processors 
        // TODO: We should be able to to this in a nicer way:
        tbb::concurrent_hash_map<int, SraBaseMessage*>::accessor acc;
        messageProcessorTable.insert(acc, (int)ID_LOGIN_REQ);
        acc->second = new ClientLoginMessage();

        messageProcessorTable.insert(acc, (int)ID_NEW_INCOMING_CONNECTION);
        acc->second = new IncommingConnectionMessage();

        messageProcessorTable.insert(acc, (int)ID_GET_CLIENT_CHARS);
        acc->second = new GetClientCharacterListMessage();
        
        messageProcessorTable.insert(acc, (int)ID_CREATE_CHAR);
        acc->second = new CreateCharacterMessage();

        messageProcessorTable.insert(acc, (int)ID_CHAT_CHANNEL_REGISTER);
        acc->second = new JoinChannelMessage();
    }

    // Destroys all data.
    MessagePump::~MessagePump(void)
    {
        
    }

    // Called from the tbb tasks and adds a processor to the queue.
    void MessagePump::AddMessage(RakNet::Packet* packet)
    {
        // Create message data from packet and address 
        MessageData nMsg;
        nMsg.address = packet->systemAddress;;
        
        // Make a copy of the data because we will dispose
        // this package as soon as we will return from this 
        // call !
        nMsg.stream = RakNet::BitStream( packet->data, packet->length, true);

        messageList.push(nMsg );
        
        isRunning = true;
    }

    // Stops all processing
    void MessagePump::Shutdown()
    {
        isRunning = false;
    }

    // Run and process messages.
    void MessagePump::Run()
    {
        // This is of course a potential bottle-neck here, because we might have
        // a large number of messages in the queue, so what we want to do is:
        // Create a number of concurrent threads that read the messages from the
        // pump and process the data. 
        while (isRunning)
        {
            MessageData message;
            if (messageList.try_pop(message))
            {
                // First de-serialize the base package
                // this will give us the op-code
                // We could simply do a stream.read<unsigned char>
                // but this way we could store more base data
                // later on.
                SraPacket p;
                p.Deserialize(&message.stream);

                int opCode = p.opCode;

                std::cout << "Processing message " << opCode << std::endl;
                tbb::concurrent_hash_map<int, SraBaseMessage*>::accessor acc;
                if (messageProcessorTable.find(acc, opCode))
                {
                    acc->second->process( &message );
                }
                // We might want to add an else case to send "unknown opcode" messages 
                // back to the client, but do we really want to ?
            } else
            {
                Sleep(10);
            }
        }

        //  } else if (packet.opcode == ID_CONNECTION_LOST || packet.opcode == ID_DISCONNECTION_NOTIFICATION) 
        //  {
        //      printf("Connection lost, but maybe the client is just changing the zone, so do not remove client\n");
        //      //TODO: We do need some sort of keepAlive mechanism to detect dead clients....
        //      /*
        //      ClientRegister* reg = ClientRegister::getInstance();
        //      reg->removeClient( address );
        //      */
        //  }  else if (packet.opcode == ID_CONNECT_TO_SRV)
        //  {
        //      //User tries to connect to 'his' server, we won't, however, let 
        //      //the client decide which zone, because we can't trust him.
        //      //This implies that a zone-port event needs to be sent from the udk
        //      //server instance if a player is moving from zoneA to zoneB. 
        //      ClientRegister* reg = ClientRegister::getInstance();
        //      const SraClientData *cdat = reg->getClientFromAddress( address );
        //      //cdat will be NULL if the client has not logged in yet
        //      if (cdat != NULL) {
        //          RakNet::RakPeerInterface *zoneI = WorldServer::getZoneInterface();
        //          if ( zoneI != NULL && zoneI->NumberOfConnections() > 0 ) 
        //          {
        //              printf("Client logged in and tries to connect to zone %d\n",cdat->zoneID);
        //              SraZonePacket p;
        //              ZeroMemory(&p, sizeof(p));
        //              p.clientID = cdat->clientID;
        //              p.fromZoneID = -1;
        //              p.toZoneID = cdat->zoneID;
        //              zoneI->Send( (char*)&p, sizeof(p), MEDIUM_PRIORITY, RELIABLE_ORDERED, 0,  RakNet::UNASSIGNED_SYSTEM_ADDRESS, true);
        //              //We are done now, because the main zone thread will receive the package
        //          } else 
        //          {
        //              printf("No zone server connected, sending error msg to client\n");

        //          }
        //      }
        //  }  else if (packet.opcode == ID_CHAT_MSG_SENT) 
        //  {
        //      printf("Chat message\n");
        //  } else if (packet.opcode == ID_CHAT_MSG_REQUEST_PENDING) 
        //  {

        //  }
        //  else {
        //      printf("Unknown opcode %d \n", packet.data[0]);
        //  }

    }
}