serverframework.cpp

/***************************************************************************
 * The contents of this file are subject to the Mozilla Public             *
 * License Version 1.1 (the "License"); you may not use this file          *
 * except in compliance with the License. You may obtain a copy of         *
 * the License at http://www.mozilla.org/MPL/                              *
 *                                                                         *
 * Software distributed under the License is distributed on an "AS         *
 * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or              *
 * implied. See the License for the specific language governing            *
 * rights and limitations under the License.                               *
 *                                                                         *
 * The Original Code is Game Network Framework (GaNeF).                    *
 *                                                                         *
 * The Initial Developers of the Original Code are                         *
 * Lars Langer and Emanuel Greisen                                         *
 * Copyright (C) 2005. Lars Langer & Emanuel Greisen                       *
 * All Rights Reserved.                                                    *
 *                                                                         *
 * Contributor(s):                                                         *
 *   none yet....                                                          *
 *                                                                         *
 ***************************************************************************/
/// \example testserver/testserverframework.cpp

#include "serverframework.h"

#include "../common/pendingacklist.h"
#include "../common/queue.h"
#include "../common/packetreceiver.h"
#include "../common/frameworkpacket.h"
#include "../common/packetsender.h"
#include "serverdata.h"
#include "../common/plugin.h"
#include "../common/presendingchain.h"

ServerFramework::ServerFramework( int port ) :
   Framework<ServerPacket,ipmsgserver>(port),
   connection_timeout_length(2000)
{
   connected_clients.clear();
   clients.clear();
   //pending_acks = new PendingAckList(this);
   //outqueue = new OutBoundQueue( pending_acks );
   //inqueue = new InBoundQueue( pending_acks );
   //receiver = new PacketReceiver( socket_resv, inqueue, this );
   //sender = new PacketSender( socket_resv, outqueue, pending_acks, this);
   default_pre_send_chain = new PreSendingChain<ServerPacket>( std::vector<Plugin<ServerPacket> *>(), outqueue );
   setDefaultPreSendingChain( );
}


ServerFramework::~ServerFramework()
{
   // TODO: clean other pre-send-chains
   delete default_pre_send_chain;
}

void ServerFramework::registerData( ServerData * data )
{
   server_data.insert( std::make_pair( data->getFrameworkId(), data ) );
}

void ServerFramework::unregisterData( FrameworkData * data )
{
   if ( server_data.find( data->getFrameworkId() ) == server_data.end() )
   {
      std::cout << "WARNING: Remove data not in serverframework, id:" << data->getFrameworkId() << std::endl;
   }
   else
   {
      server_data.erase( data->getFrameworkId() );
   }
}

ServerData * ServerFramework::getDataObject( unsigned int objid )
{
   if ( server_data.find( objid ) == server_data.end() )
      return NULL;
   return server_data[ objid ];
}

void ServerFramework::start( )
{
   pending_acks->start();
   receiver->startListening();
   sender->startSending();
}

void ServerFramework::enqueuePacket( ServerPacket * packet, const std::map< unsigned int, Client * > & clientlist )
{
   if(clientlist.size() == 1)
   {
      std::map< unsigned int, Client * >::const_iterator it = clientlist.begin();
      enqueuePacket(packet, it->second);
   }
   else
   {
      for ( std::map< unsigned int, Client * >::const_iterator it = clientlist.begin(); it != clientlist.end(); ++it )
      {
         ServerPacket * p = new ServerPacket( * packet );
         p->copyData(packet);
         enqueuePacket(p, it->second);
      }
      delete packet;
   }
}

void ServerFramework::enqueuePacket( ServerPacket * packet, Client * client )
{
   packet->client = client;
   packet->setPeer( packet->client->getAddress(), packet->client->getPort() );
   if(packet->isReliable())
   {
      packet->setSequenceNumber(packet->client->getNextOutgoingSeqNo(packet));
   }
   current_pre_send_chain->pack( packet );
}


void ServerFramework::registerPreSendingChain( const std::string & chainname, const std::vector<Plugin<ServerPacket> *> & stations )
{
   if ( pre_sending_chains.find( chainname ) != pre_sending_chains.end() )
      throw FrameworkError( "Overwriting Presend Chain not allowed" );

   pre_sending_chains.insert( std::make_pair( chainname, new PreSendingChain<ServerPacket>( stations, outqueue ) ) );
}

void ServerFramework::setPreSendingChain( const std::string & chainname )
{
   if ( pre_sending_chains.find( chainname ) == pre_sending_chains.end() )
      throw FrameworkError( std::string( "Unknown Presend Chain: " ) + chainname );

   current_pre_send_chain = pre_sending_chains[ chainname ];
}

void ServerFramework::purgeCurrentPreSendingChain( )
{
   current_pre_send_chain->purge();
}

void ServerFramework::setDefaultPreSendingChain( )
{
   current_pre_send_chain = default_pre_send_chain;
}

void ServerFramework::registerClient( Client * client )
{
   //std::cout << "registerClient( Client * " << client->getId() << " )\n";
   clients.insert( std::make_pair( client->getId(), client ) );
   if(client->getState() == Client::CONNECTED && client->hasCalledOnConnect())
   {
      connected_clients.insert( std::make_pair( client->getId(), client ) );
   }
}

void ServerFramework::unregisterClient( Client * client )
{
   clients.erase(client->getId());
   //std::cout << "unregisterClient( Client * " << client->getId() << " ("<<client<<"), "<<clients.size()<<" left )\n";
   // remove all messages from queue
   pending_acks->removeWithComparer(QueueDeletor(client));
   outqueue->removeWithComparer(QueueDeletor(client));
}

void ServerFramework::handleSystemPacket( ServerPacket * p)
{
   //std::cout << "handleSystemPacket(" << p->getReceiver() << ")\n";
   // And handle the different types.
   switch ( p->getReceiver() )
   {
   // The following are for ACK/NACKS
   case ServerPacket::PACKET_ACK:
      {
         p->client = getClientFromPacket(p);
         if ( p->client )
         {
            p->client->registerRTT(handleAckPacket(p, p->client->getId()));
         }
         else
         {
            std::cout << "We got an ackpacket from an unknown client: " << *p << std::endl;
         }
      }
      break;
   case ServerPacket::PACKET_HEARTBEAT:
      {
         if(p->client)
         {
            //std::cout << "Received a heartbeat from: " << client->getId() << std::endl;
         }
      }
      break;

   case ServerPacket::HANDSHAKE_INIT:
      {
         if(p->client->getState() == Client::NEWCLIENT)
         {
            ServerPacket * dummy_reply = ServerPacket::createDummyPacket();
            bool success = onInitialHandshake(p, dummy_reply);

            ServerPacket * reply;
            if(success)
            {
               // Init the reply
               reply = ServerPacket::createSystemPacket(ServerPacket::SETUP_DESCRIPTION);

               // write how long long we want between heartbeats
               reply->writeUInt32(connection_timeout_length);

               // write about the plugins
               std::set<unsigned int> pluginids;
               for(std::map<std::string, PreSendingChain<ServerPacket> *>::iterator it = pre_sending_chains.begin(); it != pre_sending_chains.end(); ++it)
               {
                  it->second->appendUniqueIds(pluginids);
               }
               //std::cout << "We use " << pluginids.size() << " different plugins" << std::endl;
               reply->writeUInt32(pluginids.size());
               for(std::set<unsigned int>::iterator it = pluginids.begin(); it != pluginids.end(); ++it)
               {
                  //std::cout << "Plugin: " << *it << std::endl;
                  reply->writeUInt32(*it);
               }
            }
            else
            {
               reply = ServerPacket::createSystemPacket(ServerPacket::DISCONTINUE_CLIENT);
            }
            // Do we have a tag-along ?
            reply->appendData(dummy_reply);
            delete dummy_reply;

            reply->setPeerFromPacket(p);
            enqueuePacket(reply, p->client);
         }
         else
         {
            // CRAZY we got an init from a known client
            std::cerr << "Received INIT_HANDSHAKE from a known client[state:" << p->client->getState() << "]" << std::endl;
         }
      }
      break;
   case ServerPacket::DISCONTINUE_SERVER:
      {
         // The client does not want to talk to us: if this was expected, remove the client
         if(p->client)
         {
            if(p->client->getState() == Client::NEWCLIENT)
            {
               unregisterClient(p->client);
               delete p->client;
            }
            else
            {
               std::cerr << "We received a DISCONTINUE_SERVER call from a client in state: " << p->client->getState() << std::endl;
            }
         }
      }
      break;
   case ServerPacket::READY_CLIENT:
      {
         if(p->client && p->client->getState() == Client::NEWCLIENT)
         {
            // The client is set up and ready for playing: invoke the newClientJoined(client)
            p->client->setState(Client::CONNECTED);
         }
         else
         {
            std::cerr << "We received a READY_CLIENT call from a client in state: " << p->client->getState() << std::endl;
         }
      }
      break;

   default:
      std::cerr << "Error unknown system-packet type: " << p->getReceiver() << std::endl;
      break;
   }
   // And the packet has been handled.
   delete p;
}

Client * ServerFramework::getClientFromPacket( ServerPacket * packet )
{
   // OPTIMIZE: we could do this better, linaer scan for every packet received is a good idea: I THINK NOT !!
   for( ConstClientIterator it = clients.begin(); it != clients.end(); ++it )
   {
      Client * c = it->second;
      if ( packet->peer.host == c->getAddress() && packet->peer.port == c->getPort() )
         return c;
   }
   return NULL;
}

void ServerFramework::keepAlive( unsigned int sleepmillis )
{
   // See if any of the client have silently disconnected
   unsigned int currentmillis = Framework<ServerPacket,ipmsgserver>::currentTimeMillis();
   for(ConstClientIterator it = clients.begin(); it != clients.end(); ++it)
   {
      if(it->second->getLastPacketReceivedTime() + connection_timeout_length < currentmillis)
      {
         std::cout << "Client: " << it->second->getId() << " timed out"<<std::endl;
         if(it->second->getState() == Client::CONNECTED)
         {
            it->second->setState(Client::DICONNECTED);
         }
         else
         {
            it->second->setState(Client::CONNECTION_ABORTED);
         }
      }
   }

   // Tell the Game about these (dis)connect events.
   // If new clients have connected or old have disconnected, call onConnect(client) or onDisconnect(client)
   std::vector<Client *> disconnectedclients;
   disconnectedclients.clear();
   for(std::map<unsigned int, Client *>::iterator it = clients.begin(); it != clients.end(); ++it)
   {
      Client * client = it->second;
      if(client) // This is required, because we sometimes erase elements in "clients"
      {
         if(client->getState() == Client::CONNECTED && !client->hasCalledOnConnect())
         {
            client->setCalledOnConnect(true);

            // Call to pure virtual method
            onClientConnected(client);

            // Reregister as connected client
            registerClient(client);
         }
         else if((client->getState() == Client::DICONNECTED || client->getState() == Client::CONNECTION_ABORTED) && !client->hasCalledOnDisconnect())
         {
            disconnectedclients.push_back(client);
            // We must first unregister to make sure it is no longer in the clients list (that way if we make any sendXXXXToAll(...) calls, this client receives nothing.
            connected_clients.erase(client->getId());
         }
      }
   }
   // Unregister them (while not iterating the clients-map
   std::vector<Client *>::const_iterator it;
   for(it = disconnectedclients.begin(); it != disconnectedclients.end(); ++it)
   {
      Client * client = * it;
      unregisterClient(client);
      client->setCalledOnDisconnect(true);
      if(client->getState() == Client::DICONNECTED)
      {
         // Call to pure virtual method
         onClientDisconnected(client);
      }
      delete client;
   }

   // Handle incomming packets
   while(inqueue->size() > 0)
   {
      ServerPacket * p = inqueue->next();
      if(p->client)
      {
         ServerData * receiver = getDataObject(p->getReceiver());
         if(receiver)
         {
            receiver->clientPacket(p->client, p->getType(), p);
         }
         else
         {
            std::cerr << "Packet for non-existing data:" << p->getReceiver() << std::endl;
         }
      }
      else
      {
         std::cerr << "Throw away packet " << *p << " from non-existing client" << std::endl;
      }
      delete p;
   }

   // Sleep if we have to
   if(sleepmillis != 0)
   {
      //thread::sleep(sleepmillis);
      std::cout << "TODO:: create a static sleep method"<<std::endl;
   }
}

ServerPacket * ServerFramework::deliverPacket( ServerPacket * packet )
{
   // Next packet, maybe ?
   ServerPacket * next_packet = NULL;

   // Make sure the client stays connect for `connection_timeout_length` milliseconds
   if(packet->client)
      packet->client->registerLastPacketReceivedTime(Framework<ServerPacket,ipmsgserver>::currentTimeMillis());

   // Then we increment to make us self ready for the next packet
   if(packet->isReliable())
   {
      next_packet = packet->client->incrementNextIngoingSeqNo(packet);
   }

   if(packet->getType() == 0)
   {
      handleSystemPacket(packet);
   }
   else
   {
      inqueue->insert(packet);
   }

   return next_packet;
}

bool ServerFramework::isPacketReadyForDelivery( ServerPacket * packet )
{
   // First get the client sending the packet
   packet->client = getClientFromPacket( packet );
   if(packet->isHandshakeInitPacket() && packet->client == 0)
   {
      // Create a new one
      //client = new Client(this, packet->peer.host, packet->peer.port);
      packet->client = createClient(packet->peer.host, packet->peer.port);
      registerClient(packet->client);
   }

   return (packet->client == 0 ? false : packet->client->isPacketReadyForDelivery(packet));
}

void ServerFramework::addPendingAck( ServerPacket * packet, bool haslock)
{
   //std::cout << "[Client"<<client->getId()<<"]RTT:" << client->getRTT() << std::endl;
   if(packet->client)
   {
      pending_acks->addPendingAck(packet->getSequenceNumber(), packet->getPriority(), packet->client->getId(), packet->client->getRTT(), packet, haslock);
   }
   else
   {
      std::cerr << "Crap tried to add pending ack to non-existing client\n";
      throw new FrameworkError("Tried to add a pending ACK on a packet with no Client as receiver");
   }
}

unsigned int ServerFramework::bufferedPackets( ) const
{
   unsigned int buff_p_count = 0;
   for(ConstClientIterator it = clients.begin(); it != clients.end(); ++it)
   {
      buff_p_count += it->second->bufferedPackets( );
   }
   return buff_p_count;
}

void ServerFramework::sentReliablePacket( ServerPacket * packet )
{
   /* IGNORE
   Client * client = getClientFromPacket(packet);
   if(client)
      client->setLastReliablePacketSentTime(Framework::currentTimeMillis());
   */
}

void ServerFramework::addPluginSequenceNumber( ServerPacket * packet )
{
   if(packet->getSequenceNumber() == 0)
   {
      packet->setSequenceNumber(packet->client->getNextOutgoingSeqNo(packet));
   }
}

---