rtiambassador.cc

分布式仿真 开放源码

// ----------------------------------------------------------------------------
// CERTI - HLA RunTime Infrastructure
// Copyright (C) 2002, 2003  ONERA
//
// This file is part of CERTI-libRTI
//
// CERTI-libRTI is free software ; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public License
// as published by the Free Software Foundation ; either version 2 of
// the License, or (at your option) any later version.
//
// CERTI-libRTI 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
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser 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: RTIambassador.cc,v 3.36 2004/03/04 20:19:05 breholee Exp $
// ----------------------------------------------------------------------------

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

#include "FederateAmbassador.hh"
#include "Message.hh"
#include "PrettyDebug.hh"

#include <signal.h>
#include <iostream>
#include <unistd.h>

using std::cout ;
using std::cerr ;
using std::endl ;

namespace certi {

static pdCDebug D("LIBRTI", __FILE__);

// ----------------------------------------------------------------------------
void
RTIambassador::leave(const char *msg)
{
    cout << "libRTI: " << msg << endl ;
    exit(EXIT_FAILURE);
}

// ----------------------------------------------------------------------------
void
RTIambassador::executeService(Message *req, Message *rep)
{
    // raise exception if reentrant call.
    if (is_reentrant)
        throw ConcurrentAccessAttempted();

    D.Out(pdDebug, "sending request to RTIA.");

    is_reentrant = true ;
    try {
        req->write((SocketUN *) this);
    }
    catch (NetworkError) {
        cerr << "libRTI: exception: NetworkError (write)" << endl ;
        throw RTIinternalError("libRTI: Network Write Error");
    }

    D.Out(pdDebug, "waiting RTIA reply.");

    // waiting RTI reply.
    try {
        rep->read((SocketUN *) this);
    }
    catch (NetworkError) {
        cerr << "libRTI: exception: NetworkError (read)" << endl ;
        throw RTIinternalError("libRTI: Network Read Error");
    }

    D.Out(pdDebug, "RTIA reply received.");

    if (rep->type != req->type) {
        cout << "LibRTI: Assertion failed: rep->type != req->type" << endl ;
        throw RTIinternalError("RTIambassador::executeService: "
                               "rep->type != req->type");
    }

    is_reentrant = false ;
    D.Out(pdDebug, "processing returned exception (from reply).");
    processException(rep);
    D.Out(pdDebug, "exception processed.");
}

// ----------------------------------------------------------------------------
//! Start RTIambassador processes for communication with RTIG.
/*! When a new RTIambassador is created in the application, a new process rtia
  is launched. This process is used for data exchange with rtig server.
  This process connects to rtia after one second delay (UNIX socket).
*/
RTIambassador::RTIambassador()
    throw (MemoryExhausted, RTIinternalError) : SocketUN(stIgnoreSignal)
{
    is_reentrant = false ;
    const char *rtiaexec = "rtia" ;
    const char *rtiaenv = getenv("CERTI_RTIA");
    const char *rtiacall ;
    if (rtiaenv) rtiacall = rtiaenv ;
    else rtiacall = rtiaexec ;

    // creating RTIA process.
    switch((pid_RTIA = fork())) {
      case -1: // fork failed.
        perror("fork");
        throw RTIinternalError("fork failed in RTIambassador constructor");
        break ;

      case 0: // child process (RTIA).
        execlp(rtiacall, NULL);
        perror("execlp");

        cerr << "Could not launch RTIA process." << endl
             << "Maybe RTIA is not in search PATH environment." << endl ;

        exit(-1);

      default: // father process (Federe).
        sleep(1);
        connectUN(pid_RTIA);
        break ;
    }
}

// ----------------------------------------------------------------------------
//! Closes processes.
/*! When destructor is called, kill rtia process.
 */
RTIambassador::~RTIambassador()
{
    kill(pid_RTIA, SIGINT);
}

// ----------------------------------------------------------------------------
// Tick
Boolean
RTIambassador::tick()
    throw (SpecifiedSaveLabelDoesNotExist,
           ConcurrentAccessAttempted,
           RTIinternalError)
{
    Message vers_RTI, vers_Fed ;

    // Throw exception if reentrant call.
    if (is_reentrant)
        throw ConcurrentAccessAttempted();

    is_reentrant = true ;

    // Prevenir le RTI
    vers_RTI.type = Message::TICK_REQUEST ;

    try {
        vers_RTI.write((SocketUN *) this);
    }
    catch (NetworkError) {
        cout << "tick 1." << endl ;
        cout << "LibRTI: Catched NetworkError, throw RTIinternalError."
             << endl ;
        throw RTIinternalError();
    }

    for (;;) {

        // Lire la reponse du RTIA local
        try {
            vers_Fed.read((SocketUN *) this);
        }
        catch (NetworkError) {
            cout << "tick 2." << endl ;
            cout << "LibRTI: Catched NetworkError, throw RTIinternalError."
                 << endl ;
            throw RTIinternalError();
        }

        // Si c'est de type TICK_REQUEST, il n'y a qu'a traiter l'exception.
        if (vers_Fed.type == Message::TICK_REQUEST) {
            is_reentrant = false ;
            processException(&vers_Fed);
            return vers_Fed.getBoolean();
        }

        // Sinon, le RTI nous demande un service, donc on appele une methode
        // du FederateAmbassador.

        vers_RTI.setException(e_NO_EXCEPTION);
        try {
            switch (vers_Fed.type) {

              case Message::SYNCHRONIZATION_POINT_REGISTRATION_SUCCEEDED:
                fed_amb->synchronizationPointRegistrationSucceeded(
		    vers_Fed.getLabel());
                break ;

              case Message::ANNOUNCE_SYNCHRONIZATION_POINT:
                fed_amb->announceSynchronizationPoint(vers_Fed.getLabel(),
                                                      vers_Fed.getTag());
                break ;

              case Message::FEDERATION_SYNCHRONIZED:
                fed_amb->federationSynchronized(vers_Fed.getLabel());
                break ;

              case Message::INITIATE_FEDERATE_SAVE:
                fed_amb->initiateFederateSave(vers_Fed.getLabel());
                break ;

              case Message::FEDERATION_SAVED:
                fed_amb->federationSaved();
                break ;

              case Message::REQUEST_FEDERATION_RESTORE_SUCCEEDED:
                fed_amb->requestFederationRestoreSucceeded(
                    vers_Fed.getLabel());
                break ;

              case Message::REQUEST_FEDERATION_RESTORE_FAILED:
                fed_amb->requestFederationRestoreFailed(vers_Fed.getLabel(),
                                                        vers_Fed.getTag());
                break ;

              case Message::FEDERATION_RESTORE_BEGUN:
                fed_amb->federationRestoreBegun();
                break ;

              case Message::INITIATE_FEDERATE_RESTORE:
                fed_amb->initiateFederateRestore(vers_Fed.getLabel(),
                                                 vers_Fed.getFederate());
                break ;

              case Message::FEDERATION_RESTORED:
                fed_amb->federationRestored();
                break ;

              case Message::FEDERATION_NOT_RESTORED:
                fed_amb->federationNotRestored();
                break ;

              case Message::START_REGISTRATION_FOR_OBJECT_CLASS: {
                  fed_amb->startRegistrationForObjectClass(
		      vers_Fed.getObjectClass());
              } break ;

              case Message::STOP_REGISTRATION_FOR_OBJECT_CLASS: {
                  fed_amb->
                      stopRegistrationForObjectClass(vers_Fed.getObjectClass());
              } break ;

              case Message::TURN_INTERACTIONS_ON: {
                  fed_amb->turnInteractionsOn(vers_Fed.getInteractionClass());
              } break ;

              case Message::TURN_INTERACTIONS_OFF: {
                  fed_amb->turnInteractionsOff(vers_Fed.getInteractionClass());
              } break ;

              case Message::DISCOVER_OBJECT_INSTANCE: {
                  fed_amb->
                      discoverObjectInstance(vers_Fed.getObject(),
                                             vers_Fed.getObjectClass(),
                                             vers_Fed.getName());
              } break ;

              case Message::REFLECT_ATTRIBUTE_VALUES: {
                  AttributeHandleValuePairSet *attributes = vers_Fed.getAHVPS();
                  fed_amb->
                      reflectAttributeValues(vers_Fed.getObject(),
                                             *attributes,
                                             vers_Fed.getFedTime(),
                                             vers_Fed.getTag(),
                                             vers_Fed.getEventRetraction());

                  delete attributes ;
              } break ;

              case Message::RECEIVE_INTERACTION: {
                  ParameterHandleValuePairSet *parameters = vers_Fed.getPHVPS();

                  fed_amb->receiveInteraction(vers_Fed.getInteractionClass(),
                                              *parameters,
                                              vers_Fed.getFedTime(),
                                              vers_Fed.getTag(),
                                              vers_Fed.getEventRetraction());

                  delete parameters ;
              } break ;

              case Message::REMOVE_OBJECT_INSTANCE: {
                  fed_amb->removeObjectInstance(vers_Fed.getObject(),
                                                vers_Fed.getFedTime(),
                                                vers_Fed.getTag(),
                                                vers_Fed.getEventRetraction());
              } break ;

              case Message::PROVIDE_ATTRIBUTE_VALUE_UPDATE: {



                  // fed_amb->provideAttributeValueUpdate();
              } break ;

              case Message::REQUEST_RETRACTION: {

              } break ;

              case Message::REQUEST_ATTRIBUTE_OWNERSHIP_ASSUMPTION: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->
                      requestAttributeOwnershipAssumption(vers_Fed.getObject(),
                                                          *attributeSet,
                                                          vers_Fed.getTag());
                  delete attributeSet ;
              } break ;

              case Message::REQUEST_ATTRIBUTE_OWNERSHIP_RELEASE: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->requestAttributeOwnershipRelease(
		      vers_Fed.getObject(),
		      *attributeSet,
		      vers_Fed.getTag());

                  delete attributeSet ;
              } break ;

              case Message::ATTRIBUTE_OWNERSHIP_UNAVAILABLE: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->attributeOwnershipUnavailable(vers_Fed.getObject(),
                                                         *attributeSet);

                  delete attributeSet ;
              } break ;

              case Message::ATTRIBUTE_OWNERSHIP_ACQUISITION_NOTIFICATION: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->attributeOwnershipAcquisitionNotification(
		      vers_Fed.getObject(),
		      *attributeSet);

                  delete attributeSet ;
              } break ;

              case Message::ATTRIBUTE_OWNERSHIP_DIVESTITURE_NOTIFICATION: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->attributeOwnershipDivestitureNotification(
		      vers_Fed.getObject(),
		      *attributeSet);

                  delete attributeSet ;
              } break ;

              case Message::CONFIRM_ATTRIBUTE_OWNERSHIP_ACQUISITION_CANCELLATION: {
                  AttributeHandleSet *attributeSet = vers_Fed.getAHS();

                  fed_amb->
                      confirmAttributeOwnershipAcquisitionCancellation
                      (vers_Fed.getObject(), *attributeSet);

                  delete attributeSet ;
              } break ;

              case Message::INFORM_ATTRIBUTE_OWNERSHIP: {
                  fed_amb->
                      informAttributeOwnership(vers_Fed.getObject(),
                                               vers_Fed.getAttribute(),
                                               vers_Fed.getFederate());
              } break ;

              case Message::ATTRIBUTE_IS_NOT_OWNED: {
                  fed_amb->attributeIsNotOwned(vers_Fed.getObject(),
                                               vers_Fed.getAttribute());
              } break ;

              case Message::TIME_ADVANCE_GRANT: {
                  fed_amb->timeAdvanceGrant(vers_Fed.getFedTime());
              } break ;

              default: {
                  leave("RTI service requested by RTI is unknown.");
              }
            }
        }
        catch (InvalidFederationTime &e) {
            vers_RTI.setException(e_InvalidFederationTime, e._reason);
        }
        catch (TimeAdvanceWasNotInProgress &e) {
            vers_RTI.setException(e_TimeAdvanceWasNotInProgress, e._reason);
        }
        catch (FederationTimeAlreadyPassed &e) {
            vers_RTI.setException(e_FederationTimeAlreadyPassed, e._reason);
        }
        catch (FederateInternalError &e) {
            vers_RTI.setException(e_FederateInternalError, e._reason);
        }
        catch (Exception &e) {
            vers_RTI.setException(e_RTIinternalError, e._reason);
        }

        // retourner au RTI la reponse du service demande
        vers_RTI.type = vers_Fed.type ;

        try {
            vers_RTI.write((SocketUN *) this);
        }
        catch (NetworkError) {
            cout << "tick 3." << endl ;
            cout << "LibRTI: Catched NetworkError, throw RTIinternalError."
                 << endl ;
            throw RTIinternalError();
        }
    }
}

// ----------------------------------------------------------------------------
Boolean
RTIambassador::tick(TickTime, TickTime)
    throw (SpecifiedSaveLabelDoesNotExist, ConcurrentAccessAttempted,
           RTIinternalError)
{
    return tick();
}

// ----------------------------------------------------------------------------
//! Process exception from received message.
/*! When a message is received from RTIA, it can contains an exception.
  This exception is processed by this module and a new exception is thrown.
*/
void
RTIambassador::processException(Message *msg)
{
    D.Out(pdExcept, "n