rtig_processing.cc

分布式仿真 开放源码

// ----------------------------------------------------------------------------
// CERTI - HLA RunTime Infrastructure
// Copyright (C) 2002, 2003, 2004  ONERA
//
// This file is part of CERTI
//
// CERTI is free software ; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation ; either version 2 of the License, or
// (at your option) any later version.
//
// CERTI is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY ; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program ; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
//
// $Id: RTIG_processing.cc,v 3.21 2004/03/04 20:19:05 breholee Exp $
// ----------------------------------------------------------------------------

#include <config.h>
#include "RTIG.hh"

#include <assert.h>

using std::endl ;

namespace certi {
namespace rtig {

static pdCDebug D("RTIG", "(RTIG)- ");

// ----------------------------------------------------------------------------
//! Creates a new federation.
void
RTIG::processCreateFederation(Socket *link, NetworkMessage *req)
{
    char *federation = req->federationName ;

    if (federation == NULL) throw RTIinternalError("Invalid Federation Name.");
    auditServer->addToLinef("Federation Name : %s", federation);
    Handle h = federationHandles.provide();

#ifdef FEDERATION_USES_MULTICAST
    // multicast base address
    unsigned long base_adr_mc = inet_addr(ADRESSE_MULTICAST);
    SocketMC *com_mc = NULL ;

    // creer la communication multicast
    com_mc = new SocketMC();
    if (com_mc == NULL) {
        D.Out(pdExcept, "Unable to allocate Multicast socket.");
        throw RTIinternalError("Unable to allocate Multicast socket.");
    }

    com_mc->CreerSocketMC(base_adr_mc + h, MC_PORT);

    // inserer la nouvelle federation dans la liste des federations
    federations->createFederation(federation, h, com_mc);

    // inserer descripteur fichier pour le prochain appel a un select
    ClientSockets.push_front(com_mc);

#else
    federations->createFederation(federation, h);
#endif

    // Prepare answer for RTIA
    NetworkMessage rep ;
    rep.type = NetworkMessage::CREATE_FEDERATION_EXECUTION ;
    rep.exception = e_NO_EXCEPTION ;
    rep.federation = h ;

    rep.write(link); // Send answer to RTIA

    D.Out(pdInit, "Federation \"%s\" created with Handle %d.",
          federation, rep.federation);
}

// ----------------------------------------------------------------------------
//! Add a new federate to federation.
void
RTIG::processJoinFederation(Socket *link, NetworkMessage *req)
{
    char *federation = req->federationName ;
    char *federate = req->federateName ;

    unsigned int peer = req->bestEffortPeer ;
    unsigned long address = req->bestEffortAddress ;

    Handle num_federation ;
    FederateHandle num_federe ;

    int nb_regulateurs ;
    int nb_federes ;
    bool pause ;

    if ((federation == NULL) || (federate == NULL))
        throw RTIinternalError("Invalid Federation/Federate Name.");

    auditServer->addToLinef("Federate \"%s\" joins Federation \"%s\"",
                            federate, federation);

    federations->exists(federation, num_federation);

    num_federe = federations->addFederate(num_federation,
                                          federate,
                                          (SecureTCPSocket *) link);

#ifdef FEDERATION_USES_MULTICAST
    SocketMC *com_mc = NULL ;

    federations->info(num_federation, nb_federes, nb_regulateurs,
                      pause, com_mc);
    assert(com_mc != NULL);
#else
    federations->info(num_federation, nb_federes, nb_regulateurs, pause);
#endif

    // Store Federate <->Socket reference.
    socketServer->setReferences(link->returnSocket(),
                                num_federation,
                                num_federe,
                                address,
                                peer);

    auditServer->addToLinef("(%d)with handle %d. Socket %d",
                            num_federation, num_federe,
                            link->returnSocket());

    // Prepare answer
    NetworkMessage rep ;
    rep.type = NetworkMessage::JOIN_FEDERATION_EXECUTION ;
    rep.exception = e_NO_EXCEPTION ;
    rep.federate = num_federe ;
    rep.federation = num_federation ;
    rep.numberOfRegulators = nb_regulateurs ;
    rep.bestEffortPeer = peer ;
    rep.bestEffortAddress = address ;

#ifdef FEDERATION_USES_MULTICAST
    rep.AdresseMulticast = com_mc->returnAdress();
#endif

    D.Out(pdInit, "Federate \"%s\" has joined Federation %u under handle %u.",
          federate, num_federation, num_federe);

    // Send answer
    rep.write(link);
}

// ----------------------------------------------------------------------------
//! Removes a federate from federation.
void
RTIG::processResignFederation(Handle federation,
                              FederateHandle federe)
{
    federations->remove(federation, federe);
    D.Out(pdInit, "Federate %u is resigning from federation %u.",
          federe, federation);
}

// ----------------------------------------------------------------------------
//! Removes a federation.
void
RTIG::processDestroyFederation(Socket *link, NetworkMessage *req)
{
    Handle num_federation ;
    char *federation = req->federationName ;

    if (federation == NULL) throw RTIinternalError("Invalid Federation Name.");

    auditServer->addToLinef("Name \"%s\"", federation);
    federations->exists(federation, num_federation);
    federations->destroyFederation(num_federation);
    federationHandles.free(num_federation);
    D.Out(pdInit, "Federation \"%s\" has been destroyed.", federation);

    NetworkMessage rep ;
    rep.type = NetworkMessage::DESTROY_FEDERATION_EXECUTION ;
    rep.exception = e_NO_EXCEPTION ;
    rep.federate = req->federate ;

    rep.write(link);
}

// ----------------------------------------------------------------------------
//! Put federate as time regulating.
void
RTIG::processSetTimeRegulating(NetworkMessage *msg)
{
    if (msg->regulator) {
        auditServer->addToLinef("ON at time %f", msg->date);

        federations->createRegulator(msg->federation,
                                     msg->federate,
                                     msg->date);
        D.Out(pdTerm, "Federate %u of Federation %u sets TimeRegulation ON.",
              msg->federate, msg->federation);
    }
    else {
        auditServer->addToLine("OFF");

        federations->removeRegulator(msg->federation,
                                     msg->federate);
        D.Out(pdTerm, "Federate %u of Federation %u sets TimeRegulation OFF.",
              msg->federate, msg->federation);
    }
}

// ----------------------------------------------------------------------------
//! Put federate as time constrained
void
RTIG::processSetTimeConstrained(NetworkMessage *msg)
{
    if (msg->constrained) {
        auditServer->addToLinef("ON at time %f", msg->date);

        federations->addConstrained(msg->federation,
                                    msg->federate);
        D.Out(pdTerm, "Federate %u of Federation %u is now constrained.",
              msg->federate, msg->federation);
    }
    else {
        auditServer->addToLine("OFF");

        federations->removeConstrained(msg->federation,
                                       msg->federate);
        D.Out(pdTerm, "Federate %u of Federation %u is no more constrained.",
              msg->federate, msg->federation);
    }
}

// ----------------------------------------------------------------------------
//! processMessageNull.
void
RTIG::processMessageNull(NetworkMessage *msg)
{
    auditServer->addToLinef("Date %f", msg->date);

    // Catch all exceptions because RTIA does not expect an answer anyway.
    try {
        federations->updateRegulator(msg->federation,
                                     msg->federate,
                                     msg->date);
    } catch (Exception &e) {}
}

// ----------------------------------------------------------------------------
//! processRegisterSynchronization.
void
RTIG::processRegisterSynchronization(Socket *link, NetworkMessage *req)
{
    auditServer->addToLinef("Label \"%s\" registered. Tag is \"%s\"",
                            req->label, req->tag);
    federations->manageSynchronization(req->federation,
                                       req->federate,
                                       true,
                                       req->label,
                                       req->tag);
    D.Out(pdTerm, "Federation %u is now synchronizing.", req->federation);

    // send synchronizationPointRegistrationSucceeded() to federate.
    NetworkMessage rep ;
    rep.type = NetworkMessage::SYNCHRONIZATION_POINT_REGISTRATION_SUCCEEDED ;
    rep.federate = req->federate ;
    rep.federation = req->federation ;
    rep.setLabel(req->label);
    rep.write(link);

    federations->broadcastSynchronization(req->federation,
                                          req->federate,
                                          req->label,
                                          req->tag);
}

// ----------------------------------------------------------------------------
//! processSynchronizationAchieved.
void
RTIG::processSynchronizationAchieved(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Label \"%s\" ended.", req->label);

    federations->manageSynchronization(req->federation,
                                       req->federate,
                                       false,
                                       req->label,
                                       "");
    D.Out(pdTerm, "Federate %u has synchronized.", req->federate);
}

// ----------------------------------------------------------------------------
void
RTIG::processRequestFederationSave(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Federation save request.");

    federations->requestFederationSave(req->federation, req->federate,
                                       req->label, req->date);
}
// ----------------------------------------------------------------------------
void
RTIG::processFederateSaveBegun(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Federate %u save begun.", req->federate);

    federations->federateSaveBegun(req->federation, req->federate);
}

// ----------------------------------------------------------------------------
void
RTIG::processFederateSaveStatus(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Federate %u save ended.", req->federate);

    bool status = req->type == NetworkMessage::FEDERATE_SAVE_COMPLETE ;
    federations->federateSaveStatus(req->federation, req->federate, status);
}

// ----------------------------------------------------------------------------
void
RTIG::processRequestFederationRestore(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Federate %u request restore.", req->federate);

    federations->requestFederationRestore(req->federation, req->federate,
                                          req->label);
}

// ----------------------------------------------------------------------------
void
RTIG::processFederateRestoreStatus(Socket *, NetworkMessage *req)
{
    auditServer->addToLinef("Federate %u restore ended.", req->federate);

    bool status = (req->type == NetworkMessage::FEDERATE_RESTORE_COMPLETE) ? true : false ;
    federations->federateRestoreStatus(req->federation, req->federate, status);
}

// ----------------------------------------------------------------------------
// processPublishObjectClass
void
RTIG::processPublishObjectClass(Socket *link, NetworkMessage *req)
{
    bool pub = (req->type == NetworkMessage::PUBLISH_OBJECT_CLASS);