rtig.cc

分布式仿真 开放源码

        break ;

      case NetworkMessage::DDM_ASSOCIATE_REGION:
	D[pdTrace] << "associateRegionForUpdates" << endl ;
        auditServer->setLevel(6);
        processAssociateRegion(link, msg);
        break ;	

      case NetworkMessage::DDM_UNASSOCIATE_REGION:
	D[pdTrace] << "unassociateRegionForUpdates" << endl ;
        auditServer->setLevel(6);
        processUnassociateRegion(link, msg);
        break ;	

      case NetworkMessage::DDM_SUBSCRIBE_ATTRIBUTES:
	D[pdTrace] << "subscribeObjectClassAttributes (DDM)" << endl ;
        auditServer->setLevel(6);
        processSubscribeAttributesWR(link, msg);
        break ;
	
      case NetworkMessage::DDM_UNSUBSCRIBE_ATTRIBUTES:
	D[pdTrace] << "unsubscribeObjectClassAttributes (DDM)" << endl ;
        auditServer->setLevel(6);
        processUnsubscribeAttributesWR(link, msg);
        break ;
	
      case NetworkMessage::DDM_SUBSCRIBE_INTERACTION:
	D[pdTrace] << "subscribeInteraction (DDM)" << endl ;
        auditServer->setLevel(6);
        processSubscribeInteractionWR(link, msg);
        break ;
	
      case NetworkMessage::DDM_UNSUBSCRIBE_INTERACTION:
	D[pdTrace] << "unsubscribeInteraction (DDM)" << endl ;
        auditServer->setLevel(6);
        processUnsubscribeInteractionWR(link, msg);
        break ;	
	
      default:
        // FIXME: Should 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)
{
    Handle 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()
{
    int result ;
    fd_set fd ;
    Socket *link ;

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

    if (verbose) {
	cout << "CERTI RTIG up and running ..." << endl ;
    }
    terminate = false ;

    while (!terminate) {
        // Initialize fd_set structure with all opened sockets.
        FD_ZERO(&fd);
        FD_SET(tcpSocketServer.returnSocket(), &fd);
        socketServer->addToFDSet(&fd);

        // Wait for an incoming message.
        result = 0 ;
        result = select(sysconf(_SC_OPEN_MAX), &fd, NULL, NULL, NULL);
        if ((result == -1)&& (errno == EINTR)) break ;

        // Is it a message from an already opened connection?
        link = socketServer->getActiveSocket(&fd);
        if (link != NULL) {
            D.Out(pdCom, "Incoming message on socket %ld.",
		  link->returnSocket());
            try {
                do {
                    link = processIncomingMessage((SecureTCPSocket *)link);
                    if (link == NULL)break ;
                } while (link->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->returnSocket()
                     << '.' << endl ;
                closeConnection((SecureTCPSocket *)link, true);
                link = NULL ;
            }
        }

        // Or on the server socket ?
        if (FD_ISSET(tcpSocketServer.returnSocket(), &fd)) {
            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);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeAlreadyBeingAcquired ;
    }
    catch (AttributeAlreadyBeingDivested &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeAlreadyBeingDivested ;
    }
    catch (AttributeDivestitureWasNotRequested &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeDivestitureWasNotRequested ;
    }
    catch (AttributeAcquisitionWasNotRequested &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeAcquisitionWasNotRequested ;
    }
    catch (AttributeNotDefined &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeNotDefined ;
    }
    catch (AttributeNotKnown &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeNotKnown ;
    }
    catch (AttributeNotOwned &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeNotOwned ;
    }
    catch (AttributeNotPublished &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeNotPublished ;
    }
    catch (AttributeNotSubscribed &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_AttributeNotSubscribed ;
    }
    catch (ConcurrentAccessAttempted &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_ConcurrentAccessAttempted ;
    }
    catch (CouldNotDiscover &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_CouldNotDiscover ;
    }
    catch (CouldNotOpenRID &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_CouldNotOpenRID ;
    }
    catch (CouldNotRestore &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_CouldNotRestore ;
    }
    catch (DeletePrivilegeNotHeld &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_DeletePrivilegeNotHeld ;
    }
    catch (ErrorReadingRID &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_ErrorReadingRID ;
    }
    catch (EventNotKnown &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_EventNotKnown ;
    }
    catch (FederateAlreadyPaused &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateAlreadyPaused ;
    }
    catch (FederateAlreadyExecutionMember &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateAlreadyExecutionMember ;
    }
    catch (FederateDoesNotExist &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateDoesNotExist ;
    }
    catch (FederateInternalError &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateInternalError ;
    }
    catch (FederateNameAlreadyInUse &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateNameAlreadyInUse ;
    }
    catch (FederateNotExecutionMember &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateNotExecutionMember ;
    }
    catch (FederateNotPaused &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateNotPaused ;
    }
    catch (FederateNotPublishing &e) {
        D.Out(pdExcept, "Catching \"%s\" exception.", e._name);
        CPY_NOT_NULL(e);
        rep.exception = e_FederateNotPublishing ;