rtig.cc

certi-SHM-3.0.tar 不错的开源的分布式方针软件 大家多多支持 他是linux

// -*- mode:C++ ; tab-width:4 ; c-basic-offset:4 ; indent-tabs-mode:nil -*-
// ----------------------------------------------------------------------------
// CERTI - HLA RunTime Infrastructure
// Copyright (C) 2002, 2003  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.cc,v 3.7 2003/03/21 15:06:46 breholee Exp $
// ----------------------------------------------------------------------------

#include "RTIG.hh"

namespace certi {
namespace rtig {

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

// ----------------------------------------------------------------------------
// Constructor

RTIG::RTIG()
{
    terminate = false ;

    nextFederationHandle = 1 ;

    // Start RTIG services
    const char *tcp_port_s = getenv("CERTI_TCP_PORT");
    const char *udp_port_s = getenv("CERTI_UDP_PORT");
    if (tcp_port_s==NULL) tcp_port_s = PORT_TCP_RTIG ;
    if (udp_port_s==NULL) udp_port_s = PORT_UDP_RTIG ;
    tcpPort = atoi(tcp_port_s);
    udpPort = atoi(udp_port_s);

    socketServer = new SocketServer(&tcpSocketServer, &udpSocketServer, udpPort);

    if (socketServer == NULL)
        throw RTIinternalError("No memory to create socketServer.");

    auditServer = new AuditFile(RTIG_AUDIT_FILENAME);

    if (auditServer == NULL)
        throw RTIinternalError("No memory to create auditServer.");

    federations = new FederationsList(socketServer, auditServer);
    if (federations == NULL)
        throw RTIinternalError("No memory to create federations.");
}

// ----------------------------------------------------------------------------
// Destructor

RTIG::~RTIG()
{
    tcpSocketServer.close();
    udpSocketServer.close();

    delete socketServer ;
    delete federations ;
    delete auditServer ;

    cout << endl << "Stopping RTIG" << endl ;
}

// ----------------------------------------------------------------------------
//! Choose the right processing module to call.
/*! This module chooses the right processing module to call. This process is
  done by examinating the message type.
*/
Socket*
RTIG::chooseProcessingMethod(Socket *link, NetworkMessage *msg)
{

    // This may throw a security error
    D.Out(pdTrace,"link->returnSocket() : %d",link->returnSocket());
    socketServer->checkMessage(link->returnSocket(), msg);   

    switch(msg->type) {
    case m_MESSAGE_NULL:
        D.Out(pdDebug, "Message Null.");
        auditServer->setLevel(0);
        processMessageNull(msg);
        break ;

    case m_UPDATE_ATTRIBUTE_VALUES:
        D.Out(pdDebug, "UpdateAttributeValue.");
        auditServer->setLevel(1);
        processUpdateAttributeValues(link, msg);
        break ;

    case m_SEND_INTERACTION:
        D.Out(pdTrace, "send interaction.");
        auditServer->setLevel(2);
        processSendInteraction(link, msg);
        break ;

    case m_CLOSE_CONNEXION:
        D.Out(pdTrace, "Close connection %ld.", link->returnSocket());
        auditServer->setLevel(9);
        auditServer->addToLinef("Socket %ld", link->returnSocket());
        closeConnection(link, false);
        link = NULL ;
        break ;

    case m_CREATE_FEDERATION_EXECUTION:         
        D.Out(pdTrace, "Create federation \"%s\".", msg->federationName);
        auditServer->setLevel(9);
        processCreateFederation(link, msg);
        break ;

    case m_DESTROY_FEDERATION_EXECUTION:
        D.Out(pdTrace, "Destroy federation \"%s\".", msg->federationName);
        auditServer->setLevel(9);
        processDestroyFederation(link, msg);
        break ;

    case m_JOIN_FEDERATION_EXECUTION:
        D.Out(pdTrace, "federate \"%s\" joins federation \"%s\".",
              msg->federateName, msg->federationName);
        auditServer->setLevel(9);
        processJoinFederation(link, msg);
        break ;

    case m_RESIGN_FEDERATION_EXECUTION:
        D.Out(pdTrace, "Federate no %u leaves federation no %u .",
              msg->federate, msg->federation);
        auditServer->setLevel(9);
        processResignFederation(msg->federation, msg->federate);
        break ;

    case m_REGISTER_FEDERATION_SYNCHRONIZATION_POINT:
        D.Out(pdTrace,
              "Federation %u: registerFedSyncPoint from federate %u.",
              msg->federation, msg->federate);
	      D.Out(pdTrace,"Call set Level");
        auditServer->setLevel(8);
        D.Out(pdTrace,"Call processRegisterSynchronization");
	processRegisterSynchronization(link, msg);

        break ;

    case m_SYNCHRONIZATION_POINT_ACHIEVED:
        D.Out(pdTrace,
              "Federation %u: synchronizationPointAchieved from federate %u.",
              msg->federation, msg->federate);
        auditServer->setLevel(8);
        processSynchronizationAchieved(link, msg);
        break ;

    case m_SET_TIME_REGULATING:
        D.Out(pdTrace, "SetTimeRegulating du federe %u(date=%f).",
              msg->federate, msg->date);
        auditServer->setLevel(8);
        processSetTimeRegulating(msg);
        break ;

    case m_SET_TIME_CONSTRAINED:
        D.Out(pdTrace, "SetTimeConstrained du federe %u.", msg->federate);
        auditServer->setLevel(8);
        processSetTimeConstrained(msg);
        break ;

    case m_PUBLISH_OBJECT_CLASS:
    case m_UNPUBLISH_OBJECT_CLASS:
        D.Out(pdTrace, "un/publishObjectClass.");
        auditServer->setLevel(7);
        processPublishObjectClass(link, msg);
        break ;

    case m_PUBLISH_INTERACTION_CLASS:
    case m_UNPUBLISH_INTERACTION_CLASS:
        D.Out(pdTrace, "un/publishInteractionClass.");
        auditServer->setLevel(7);
        processPublishInteractionClass(link, msg);
        break ;

    case m_SUBSCRIBE_OBJECT_CLASS:
    case m_UNSUBSCRIBE_OBJECT_CLASS:
        D.Out(pdTrace, "un/subscribeObjectClass.");
        auditServer->setLevel(7);
        processSubscribeObjectClass(link, msg);
        break ;

    case m_SUBSCRIBE_INTERACTION_CLASS:
    case m_UNSUBSCRIBE_INTERACTION_CLASS:
        D.Out(pdTrace, "un/subscribeInteractionClass.");
        auditServer->setLevel(7);
        processSubscribeInteractionClass(link, msg);
        break ;

    case m_REQUEST_ID:
        D.Out(pdTrace, "requestID.");
        auditServer->setLevel(6);
        processRequestId(link, msg);
        break ;

    case m_REGISTER_OBJECT:
        D.Out(pdTrace, "registerObject.");
        auditServer->setLevel(6);
        processRegisterObject(link, msg);
        break ;

    case m_DELETE_OBJECT:
        D.Out(pdTrace, "DeleteObject..");
        auditServer->setLevel(6);
        processDeleteObject(link, msg);
        break ;

    case m_IS_ATTRIBUTE_OWNED_BY_FEDERATE:
        D.Out(pdTrace, "isAttributeOwnedByFederate..");
        auditServer->setLevel(2);
        processAttributeOwnedByFederate(link, msg);
        break ;

    case m_QUERY_ATTRIBUTE_OWNERSHIP:
        D.Out(pdTrace, "queryAttributeOwnership..");
        auditServer->setLevel(2);
        processQueryAttributeOwnership(link, msg);
        break ;

    case m_NEGOTIATED_ATTRIBUTE_OWNERSHIP_DIVESTITURE:
        D.Out(pdTrace, "negotiatedAttributeOwnershipDivestiture..");
        auditServer->setLevel(6);
        processNegotiatedOwnershipDivestiture(link, msg);
        break ;

    case m_ATTRIBUTE_OWNERSHIP_ACQUISITION_IF_AVAILABLE:
        D.Out(pdTrace, "attributeOwnershipAcquisitionIfAvailable..");
        auditServer->setLevel(6);
        processAcquisitionIfAvailable(link, msg);
        break ;

    case m_UNCONDITIONAL_ATTRIBUTE_OWNERSHIP_DIVESTITURE:
        D.Out(pdTrace, "unconditionalAttributeOwnershipDivestiture..");
        auditServer->setLevel(6);
        processUnconditionalDivestiture(link, msg);
        break ;

    case m_ATTRIBUTE_OWNERSHIP_ACQUISITION:
        D.Out(pdTrace, "attributeOwnershipAcquisition..");
        auditServer->setLevel(6);
        processOwnershipAcquisition(link, msg);
        break ;

    case m_CANCEL_NEGOTIATED_ATTRIBUTE_OWNERSHIP_DIVESTITURE:
        D.Out(pdTrace, "cancelNegociatedAttributeOwnershipDivestiture..");
        auditServer->setLevel(6);
        processCancelNegotiatedDivestiture(link, msg);
        break ;

    case m_ATTRIBUTE_OWNERSHIP_RELEASE_RESPONSE:
        D.Out(pdTrace, "attributeOwnershipReleaseResponse..");
        auditServer->setLevel(6);
        processReleaseResponse(link, msg);
        break ;

    case m_CANCEL_ATTRIBUTE_OWNERSHIP_ACQUISITION:
        D.Out(pdTrace, "cancelAttributeOwnershipAcquisition..");
        auditServer->setLevel(6);
        processCancelAcquisition(link, msg);
        break ;

    default:
        // FIXME: Sould treat other cases CHANGE_*_ORDER/TRANSPORT_TYPE
        D.Out(pdError, "processMessageRecu: unknown type %u.", msg->type);
        throw RTIinternalError("Unknown Message Type");
    }

    return link ; // It may have been set to NULL by closeConnection.
}

// ----------------------------------------------------------------------------
//! closeConnection
/*! If a connection is closed in emergency, KillFederate will be called on
  federations attribute to remove all references to this federate.
*/
void
RTIG::closeConnection(Socket *link, bool emergency)
{
    FederationHandle federation ;
    FederateHandle federate ;

    try {
        socketServer->close(link->returnSocket(), federation, federate);
    }
    catch (RTIinternalError &e) {
        D.Out(pdError, "Connection not found while trying to close it.");
    }

    if (emergency) {
        D.Out(pdExcept, "Killing Federate(%u, %u)...", federation, federate);
        federations->killFederate(federation, federate);
        D.Out(pdExcept, "Federate(%u, %u)Killed... ", federation, federate);
    }
}

// ----------------------------------------------------------------------------
// execute
void
RTIG::execute(void)
{
    int result ;
    fd_set fd ;
    SocketTuple *link ;

    // create TCP and UDP connections for the RTIG server
    udpSocketServer.createUDPServer(udpPort);
    tcpSocketServer.createTCPServer(tcpPort);
    // udpSocketServer.createUDPServer(PORT_UDP_RTIG);

    cout << "RTIG up and running." << endl ;

    terminate = false ;
    
    FD_ZERO(&fd);
    FD_SET(tcpSocketServer.returnSocket(), &fd);
    ::select(ulimit(4, 0), &fd, NULL, NULL,NULL);

    while (!terminate) {
        

        D.Out(pdTrace, "Waiting for request");
        link = socketServer->select(&fd, &terminate);
        if (terminate)
           break;
        if (link != NULL) {
            D.Out(pdCom, "Incoming message on socket %ld.", link->QuickLink->returnSocket());
            try {
                do {
                    link->QuickLink = (SocketSHM*)processIncomingMessage(link->QuickLink);
                    if (link ->QuickLink== NULL)break ;
		    D.Out(pdTrace, "Request processed");
                } while (link->QuickLink->isDataReady()== RTI_TRUE);
            }
            catch (NetworkError &e) {
                if (e._reason != NULL)
                    D.Out(pdExcept, "Catching Network Error, reason : %s", e._reason);
                else
                    D.Out(pdExcept, "Catching Network Error, no reason string.");
                cout << "RTIG dropping client connection " << link->QuickLink->returnSocket()
                     << '.' << endl ;
                closeConnection((SocketSHM *)link->QuickLink, true);
                link = NULL ;
            }
        }

        // Or on the server socket ?
        else{
            D.Out(pdCom, "Demande de connexion.");
            openConnection();
        }
    }
}

// ----------------------------------------------------------------------------
// openConnection

void
RTIG::openConnection()
{
    try {
        socketServer->open();
    }
    catch (RTIinternalError &e) {
        D.Out(pdExcept, "Error while accepting new connection : %s.", e._reason);
    }

    D.Out(pdInit, "Accepting new connection.");
}

// ----------------------------------------------------------------------------
//! process incoming messages.
/*! This module works as follows:

Each processXXX module processes its own answer and any broadcast needed.
processXXX module calling is decided by the ChooseProcessingMethod module.
But if an exception occurs while processing a message, the exception is
caught by this module. Then a message, similar to the received one is sent
on the link. This message only holds the exception.
*/
Socket*
RTIG::processIncomingMessage(Socket *link)
{
    NetworkMessage msg ;
    NetworkMessage rep ; // Server Answer(only if an exception is raised)
    char buffer[256] ; // To store the exception reason

    if (link == NULL) {
        D.Out(pdError, "NULL socket in processMessageRecu.");
        return NULL ;
    }

    msg.read(link);

    rep.type = msg.type ;
    rep.exception = e_NO_EXCEPTION ;
    rep.federate = msg.federate ;

    auditServer->startLine(msg.federation, msg.federate, msg.type);

    // This macro is used to copy any non null exception reason
    // string into our buffer(used for Audit purpose).

#define CPY_NOT_NULL(A) { if (A._reason != NULL)strcpy(buffer, A._reason); }

    buffer[0] = 0 ;

    try {
        link = chooseProcessingMethod(link, &msg);
    }
    catch (ArrayIndexOutOfBounds &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_ArrayIndexOutOfBounds ;
    }
    catch (AttributeAlreadyOwned &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeAlreadyOwned ;
    }
    catch (AttributeAlreadyBeingAcquired &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);