#include <MessagePump.h>

Public Member Functions
void	AddMessage (RakNet::Packet *packet)
void	Shutdown ()
Static Public Member Functions
static void	Run ()
Static Public Attributes
static MessagePump *	Instance = new MessagePump()
Private Member Functions
	MessagePump (void)
	~MessagePump (void)
Static Private Attributes
static bool	isRunning = true
static tbb::concurrent_queue < MessageData >	messageList
static tbb::concurrent_hash_map< int, SraBaseMessage * >	messageProcessorTable

Detailed Description

Definition at line 42 of file MessagePump.h.

Constructor & Destructor Documentation

SraNetwork::MessagePump::MessagePump ( void ) [private]

Definition at line 27 of file MessagePump.cpp.

References SraNetwork::ID_CHAT_CHANNEL_REGISTER, SraNetwork::ID_CREATE_CHAR, SraNetwork::ID_GET_CLIENT_CHARS, SraNetwork::ID_LOGIN_REQ, ID_NEW_INCOMING_CONNECTION, and messageProcessorTable.

    {
        // 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();
    }

SraNetwork::MessagePump::~MessagePump ( void ) [private]

Definition at line 49 of file MessagePump.cpp.

    {
        
    }

Member Function Documentation

void SraNetwork::MessagePump::AddMessage ( RakNet::Packet * packet )

Definition at line 55 of file MessagePump.cpp.

References SraNetwork::MessageData::address, RakNet::Packet::data, isRunning, RakNet::Packet::length, messageList, SraNetwork::MessageData::stream, and RakNet::Packet::systemAddress.

Referenced by SraNetwork::WorldServer::ClientConnectionThread().

    {
        // 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;
    }

void SraNetwork::MessagePump::Run ( ) [static]

Definition at line 78 of file MessagePump.cpp.

References SraNetwork::SraPacket::Deserialize(), isRunning, messageList, messageProcessorTable, SraNetwork::SraPacket::opCode, opCode, and SraNetwork::MessageData::stream.

Referenced by main().

    {
        // 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]);
        //  }

    }