source: source/ariba/communication/BaseCommunication.cpp@ 5637

// [License]
// The Ariba-Underlay Copyright
//
// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)
//
// Institute of Telematics
// UniversitÀt Karlsruhe (TH)
// Zirkel 2, 76128 Karlsruhe
// Germany
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// 2. Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
//
// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR
// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// The views and conclusions contained in the software and documentation
// are those of the authors and should not be interpreted as representing
// official policies, either expressed or implied, of the Institute of
// Telematics.
// [License]

#include "BaseCommunication.h"

#ifdef UNDERLAY_OMNET
  #include "ariba/communication/modules/transport/omnet/AribaOmnetModule.h"
  #include "ariba/communication/modules/network/omnet/OmnetNetworkProtocol.h"
  #include "ariba/utility/system/StartupWrapper.h"

  using ariba::communication::AribaOmnetModule;
  using ariba::communication::OmnetNetworkProtocol;
  using ariba::utility::StartupWrapper;
#endif

namespace ariba {
namespace communication {

#include "networkinfo/AddressDiscovery.hpp"

use_logging_cpp(BaseCommunication);

/// adds an endpoint to the list
void BaseCommunication::add_endpoint( const address_v* endpoint ) {
        if (endpoint==NULL) return;
        BOOST_FOREACH( endpoint_reference& ref, remote_endpoints ) {
                if (ref.endpoint->type_id() == endpoint->type_id() && *ref.endpoint == *endpoint) {
                        ref.count++;
                        return;
                }
        }
        endpoint_reference ref;
        ref.endpoint = endpoint->clone();
        ref.count = 1;
        remote_endpoints.push_back(ref);
}

/// removes an endpoint from the list
void BaseCommunication::remove_endpoint( const address_v* endpoint ) {
        if (endpoint==NULL) return;
        for (vector<endpoint_reference>::iterator i = remote_endpoints.begin();
                i != remote_endpoints.end(); i++) {
                if (*i->endpoint == *endpoint) {
                        i->count--;
                        if (i->count==0) {
                                logging_info("No more links to " << i->endpoint->to_string() << ": terminating transports!");
                                transport->terminate(i->endpoint);
                                delete i->endpoint;
                                remote_endpoints.erase(i);
                        }
                        return;
                }
        }
}

BaseCommunication::BaseCommunication() {
        this->transport = NULL;
        this->started = false;
}

BaseCommunication::~BaseCommunication(){
}

void BaseCommunication::start() {
        logging_info( "Starting up ..." );
        currentSeqnum = 0;

        // creating transports
        logging_info( "Creating transports ..." );

#ifdef UNDERLAY_OMNET
        AribaOmnetModule* module = StartupWrapper::getCurrentModule();
        module->setServerPort( listenport );

        transport = module;
        network = new OmnetNetworkProtocol( module );
#else
        transport = new transport_peer( localDescriptor.getEndpoints() );
#endif

        logging_info( "Searching for local locators ..." );
        discover_endpoints(localDescriptor.getEndpoints());
        logging_info( "Done. Local endpoints = " << localDescriptor.toString() );

        transport->register_listener( this );
        transport->start();

#ifndef UNDERLAY_OMNET
        // bind to the network change detection
        networkMonitor.registerNotification( this );
#endif

        // base comm startup done
        started = true;
        logging_info( "Started up." );
}

void BaseCommunication::stop() {
        logging_info( "Stopping transports ..." );

        transport->stop();
        delete transport;
        started = false;

        logging_info( "Stopped." );
}

bool BaseCommunication::isStarted(){
        return started;
}

/// Sets the endpoints
void BaseCommunication::setEndpoints( string& _endpoints ) {
        localDescriptor.getEndpoints().assign(_endpoints);
        logging_info("Setting local end-points: "
                << localDescriptor.getEndpoints().to_string());
}

const LinkID BaseCommunication::establishLink(
        const EndpointDescriptor& descriptor,
        const LinkID& link_id,
        const QoSParameterSet& qos,
        const SecurityParameterSet& sec) {

        // copy link id
        LinkID linkid = link_id;

        // debug
        logging_debug( "Request to establish link" );

        // create link identifier
        if (linkid.isUnspecified())     linkid = LinkID::create();

        // create link descriptor
        logging_debug( "Creating new descriptor entry with local link id=" << linkid.toString() );
        LinkDescriptor* ld = new LinkDescriptor();
        ld->localLink = linkid;
        addLink( ld );

        // send a message to request new link to remote
        logging_debug( "Send messages with request to open link to " << descriptor.toString() );
        AribaBaseMsg baseMsg( AribaBaseMsg::typeLinkRequest, linkid );
        baseMsg.getLocalDescriptor() = localDescriptor;

        // serialize and send message
        send( &baseMsg, descriptor );

        return linkid;
}

void BaseCommunication::dropLink(const LinkID link) {

        logging_debug( "Starting to drop link " + link.toString() );

        // see if we have the link
        LinkDescriptor& ld = queryLocalLink( link );
        if( ld.isUnspecified() ) {
                logging_error( "Don't know the link you want to drop "+ link.toString() );
                return;
        }

        // tell the registered listeners
        BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
                i->onLinkDown( link, ld.localLocator, ld.remoteLocator );
        }

        // create message to drop the link
        logging_debug( "Sending out link close request. for us, the link is closed now" );
        AribaBaseMsg msg( AribaBaseMsg::typeLinkClose, ld.localLink, ld.remoteLink );

        // send message to drop the link
        send( &msg, ld );

        // remove from map
        removeLink(link);
}

seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {

        logging_debug( "Sending out message to link " << lid.toString() );

        // query local link info
        LinkDescriptor& ld = queryLocalLink(lid);
        if( ld.isUnspecified() ){
                logging_error( "Don't know the link with id " << lid.toString() );
                return -1;
        }

        // link not up-> error
        if( !ld.up ) {
                logging_error("Can not send on link " << lid.toString() << ": link not up");
                return -1;
        }

        // create message
        AribaBaseMsg msg( AribaBaseMsg::typeData, ld.localLink, ld.remoteLink );

        // encapsulate the payload message
        msg.encapsulate( const_cast<Message*>(message) );

        // send message
        send( &msg, ld );

        // return sequence number
        return ++currentSeqnum;
}

const EndpointDescriptor& BaseCommunication::getEndpointDescriptor(const LinkID link) const {
        if( link == LinkID::UNSPECIFIED){
                return localDescriptor;
        } else {
                LinkDescriptor& linkDesc = queryLocalLink(link);
                if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
                return linkDesc.remoteEndpoint;
        }
}

void BaseCommunication::registerEventListener(CommunicationEvents* _events){
        if( eventListener.find( _events ) == eventListener.end() )
                eventListener.insert( _events );
}

void BaseCommunication::unregisterEventListener(CommunicationEvents* _events){
        EventListenerSet::iterator i = eventListener.find( _events );
        if( i != eventListener.end() )
                eventListener.erase( i );
}

SystemEventType TransportEvent("Transport");
SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );

class DispatchMsg {
public:
        address_v* local;
        address_v* remote;
        Message* message;
};

/// called when a system event is emitted by system queue
void BaseCommunication::handleSystemEvent(const SystemEvent& event) {

        // dispatch received messages
        if ( event.getType() == MessageDispatchEvent ){
                logging_debug( "Forwarding message receiver" );
                DispatchMsg* dmsg = event.getData<DispatchMsg>();
                Message* msg = dmsg->message;
                receiveMessage(msg, dmsg->local, dmsg->remote);
                msg->dropPayload();
                delete dmsg;
                delete msg;
        }
}

/// called when a message is received from transport_peer
void BaseCommunication::receive_message(transport_protocol* transport,
        const address_vf local, const address_vf remote, const uint8_t* data,
        size_t size) {

//      logging_debug( "Dispatching message" );

        // convert data
        Data data_( const_cast<uint8_t*>(data), size * 8 );
        DispatchMsg* dmsg = new DispatchMsg();

        Message* msg = new Message(data_);
        dmsg->local = local->clone();
        dmsg->remote = remote->clone();
        dmsg->message = msg;

        SystemQueue::instance().scheduleEvent(
                SystemEvent( this, MessageDispatchEvent, dmsg )
        );
}

/// handles a message from the underlay transport
void BaseCommunication::receiveMessage(const Message* message,
        const address_v* local, const address_v* remote ){

        /// decapsulate message
        AribaBaseMsg* msg = ((Message*)message)->decapsulate<AribaBaseMsg>();
        logging_debug( "Receiving message of type " << msg->getTypeString() );

        // handle message
        switch (msg->getType()) {

                // ---------------------------------------------------------------------
                // data message
                // ---------------------------------------------------------------------
                case AribaBaseMsg::typeData: {
                        logging_debug( "Received data message, forwarding to overlay" );
                        if( messageReceiver != NULL ) {
                                messageReceiver->receiveMessage(
                                        msg, msg->getRemoteLink(), NodeID::UNSPECIFIED
                                );
                        }
                        break;
                }

                // ---------------------------------------------------------------------
                // handle link request from remote
                // ---------------------------------------------------------------------
                case AribaBaseMsg::typeLinkRequest: {
                        logging_debug( "Received link open request" );

                        /// only answer the first request
                        if (!queryRemoteLink(msg->getLocalLink()).isUnspecified()) {
                                logging_debug("Link request already received. Ignore!");
                                break;
                        }

                        /// create link ids
                        LinkID localLink  = LinkID::create();
                        LinkID remoteLink = msg->getLocalLink();
                        logging_debug( "local=" << local->to_string()
                                << " remote=" << remote->to_string()
                        );

                        // check if link creation is allowed by ALL listeners
                        bool allowlink = true;
                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
                                allowlink &= i->onLinkRequest( localLink, local, remote );
                        }

                        // not allowed-> warn
                        if( !allowlink ){
                                logging_warn( "Overlay denied creation of link" );
                                return;
                        }

                        // create descriptor
                        LinkDescriptor* ld = new LinkDescriptor();
                        ld->localLink = localLink;
                        ld->remoteLink = remoteLink;
                        ld->localLocator = local->clone();
                        ld->remoteLocator = remote->clone();
                        ld->remoteEndpoint = msg->getLocalDescriptor();
                        add_endpoint(ld->remoteLocator);

                        // add layer 1-3 addresses
                        ld->remoteEndpoint.getEndpoints().add(
                                ld->remoteLocator, endpoint_set::Layer1_3);
                        localDescriptor.getEndpoints().add(
                                local, endpoint_set::Layer1_3
                        );

                        // link is now up-> add it
                        ld->up = true;
                        addLink(ld);

                        // link is up!
                        logging_debug( "Link (initiated from remote) is up with "
                                << "local(id=" << ld->localLink.toString() << ","
                                << "locator=" << ld->localLocator->to_string() << ") "
                                << "remote(id=" << ld->remoteLink.toString() << ", "
                                << "locator=" << ld->remoteLocator->to_string() << ")"
                        );

                        // sending link request reply
                        logging_debug( "Sending link request reply with ids "
                                << "local=" << localLink.toString() << ", "
                                << "remote=" << remoteLink.toString() );
                        AribaBaseMsg reply( AribaBaseMsg::typeLinkReply, localLink, remoteLink );
                        reply.getLocalDescriptor() = localDescriptor;
                        reply.getRemoteDescriptor() = ld->remoteEndpoint;

                        send( &reply, *ld );

                        // inform listeners about new open link
                        BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
                                i->onLinkUp( localLink, ld->localLocator, ld->remoteLocator);
                        }

                        // done
                        break;
                }

                // ---------------------------------------------------------------------
                // handle link request reply
                // ---------------------------------------------------------------------
                case AribaBaseMsg::typeLinkReply: {
                        logging_debug( "Received link open reply for a link we initiated" );

                        // this is a reply to a link open request, so we have already
                        // a link mapping and can now set the remote link to valid
                        LinkDescriptor& ld = queryLocalLink( msg->getRemoteLink() );

                        // no link found-> warn!
                        if (ld.isUnspecified()) {
                                logging_warn("Failed to find local link " << msg->getRemoteLink().toString());
                                return;
                        }

                        // set remote locator and link id
                        ld.remoteLink = msg->getLocalLink();
                        ld.remoteLocator = remote->clone();
                        localDescriptor.getEndpoints().add(
                                msg->getRemoteDescriptor().getEndpoints(),
                                endpoint_set::Layer1_3
                        );
                        ld.up = true;
                        add_endpoint(ld.remoteLocator);

                        logging_debug( "Link is now up with local id "
                                << ld.localLink.toString() << " and remote id "
                                << ld.remoteLink.toString() );


                        // inform lisneters about link up event
                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
                                i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
                        }

                        // done
                        break;
                }

                // ---------------------------------------------------------------------
                // handle link close requests
                // ---------------------------------------------------------------------
                case AribaBaseMsg::typeLinkClose: {
                        // get remote link
                        const LinkID& localLink = msg->getRemoteLink();
                        logging_debug( "Received link close request for link " << localLink.toString() );

                        // searching for link, not found-> warn
                        LinkDescriptor& linkDesc = queryLocalLink( localLink );
                        if (linkDesc.isUnspecified()) {
                                logging_warn("Failed to find local link " << localLink.toString());
                                return;
                        }

                        // inform listeners
                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
                                i->onLinkDown( linkDesc.localLink,
                                                linkDesc.localLocator, linkDesc.remoteLocator );
                        }

                        // remove the link descriptor
                        removeLink( localLink );

                        // done
                        break;
                }

                // ---------------------------------------------------------------------
                // handle link locator changes
                // ---------------------------------------------------------------------
                case AribaBaseMsg::typeLinkUpdate: {
                        const LinkID& localLink = msg->getRemoteLink();
                        logging_debug( "Received link update for link "
                                << localLink.toString() );

                        // find the link description
                        LinkDescriptor& linkDesc = queryLocalLink( localLink );
                        if (linkDesc.isUnspecified()) {
                                logging_warn("Failed to update local link "
                                        << localLink.toString());
                                return;
                        }

                        // update the remote locator
                        const address_v* oldremote = linkDesc.remoteLocator;
                        linkDesc.remoteLocator = remote->clone();

                        // inform the listeners (local link has _not_ changed!)
                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
                                i->onLinkChanged(
                                        linkDesc.localLink,     // linkid
                                        linkDesc.localLocator,  // old local
                                        linkDesc.localLocator,  // new local
                                        oldremote,              // old remote
                                        linkDesc.remoteLocator  // new remote
                                );
                        }

                        // done
                        break;
                }
        }
}

/// add a newly allocated link to the set of links
void BaseCommunication::addLink( LinkDescriptor* link ) {
        linkSet.push_back( link );
}

/// remove a link from set
void BaseCommunication::removeLink( const LinkID& localLink ) {
        for(LinkSet::iterator i=linkSet.begin(); i != linkSet.end(); i++){
                if( (*i)->localLink != localLink) continue;
                remove_endpoint((*i)->remoteLocator);
                delete *i;
                linkSet.erase( i );
                break;
        }
}

/// query a descriptor by local link id
BaseCommunication::LinkDescriptor& BaseCommunication::queryLocalLink( const LinkID& link ) const {
        for (int i=0; i<linkSet.size();i++)
                if (linkSet[i]->localLink == link) return (LinkDescriptor&)*linkSet[i];

        return LinkDescriptor::UNSPECIFIED();
}

/// query a descriptor by remote link id
BaseCommunication::LinkDescriptor& BaseCommunication::queryRemoteLink( const LinkID& link ) const {
        for (int i=0; i<linkSet.size();i++)
                if (linkSet[i]->remoteLink == link) return (LinkDescriptor&)*linkSet[i];

        return LinkDescriptor::UNSPECIFIED();
}

LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
        LinkIDs ids;
        for (int i=0; i<linkSet.size(); i++){
                if( addr == NULL ){
                        ids.push_back( linkSet[i]->localLink );
                } else {
                        if ( *linkSet[i]->remoteLocator == *addr )
                                ids.push_back( linkSet[i]->localLink );
                }
        }
        return ids;
}

void BaseCommunication::onNetworkChange(const NetworkChangeInterface::NetworkChangeInfo& info){

#ifdef UNDERLAY_OMNET

        // we have no mobility support for simulations
        return

#endif // UNDERLAY_OMNET

/*- disabled!

        // we only care about address changes, not about interface changes
        // as address changes are triggered by interface changes, we are safe here
        if( info.type != NetworkChangeInterface::EventTypeAddressNew &&
                info.type != NetworkChangeInterface::EventTypeAddressDelete ) return;

        logging_info( "base communication is handling network address changes" );

        // get all now available addresses
        NetworkInformation networkInformation;
        AddressInformation addressInformation;

        NetworkInterfaceList interfaces = networkInformation.getInterfaces();
        AddressList addresses;

        for( NetworkInterfaceList::iterator i = interfaces.begin(); i != interfaces.end(); i++ ){
                AddressList newaddr = addressInformation.getAddresses(*i);
                addresses.insert( addresses.end(), newaddr.begin(), newaddr.end() );
        }

        //
        // get current locators for the local endpoint
        // TODO: this code is dublicate of the ctor code!!! cleanup!
        //

        NetworkProtocol::NetworkLocatorSet locators = network->getAddresses();
        NetworkProtocol::NetworkLocatorSet::iterator i = locators.begin();
        NetworkProtocol::NetworkLocatorSet::iterator iend = locators.end();

        //
        // remember the old local endpoint, in case it changes
        //

        EndpointDescriptor oldLocalDescriptor( localDescriptor );

        //
        // look for local locators that we can use in communication
        //
        // choose the first locator that is not localhost
        //

        bool foundLocator = false;
        bool changedLocator = false;

        for( ; i != iend; i++){
                logging_debug( "local locator found " << (*i)->toString() );
                IPv4Locator* ipv4locator = dynamic_cast<IPv4Locator*>(*i);

                if( *ipv4locator != IPv4Locator::LOCALHOST &&
                    *ipv4locator != IPv4Locator::ANY       &&
                    *ipv4locator != IPv4Locator::BROADCAST  ){

                        ipv4locator->setPort( listenport );
                        changedLocator = *localDescriptor.locator != *ipv4locator;
                        localDescriptor.locator = ipv4locator;
                        logging_info( "binding to addr = " << ipv4locator->toString() );
                        foundLocator = true;
                        break;
                }
        } // for( ; i != iend; i++)

        //
        // if we found no locator, bind to localhost
        //

        if( !foundLocator ){
                changedLocator = *localDescriptor.locator != IPv4Locator::LOCALHOST;
                localDescriptor.locator = new IPv4Locator( IPv4Locator::LOCALHOST );
                ((IPv4Locator*)(localDescriptor.locator))->setPort( listenport );
                logging_info( "found no good local lcoator, binding to addr = " <<
                                                localDescriptor.locator->toString() );
        }

        //
        // if we have connections that have no more longer endpoints
        // close these. they will be automatically built up again.
        // also update the local locator in the linkset mapping
        //

        if( changedLocator ){

                logging_debug( "local endp locator has changed to " << localDescriptor.toString() <<
                                ", resettings connections that end at old locator " <<
                                        oldLocalDescriptor.toString());

                LinkSet::iterator i = linkSet.begin();
                LinkSet::iterator iend = linkSet.end();

                for( ; i != iend; i++ ){

                        logging_debug( "checking connection for locator change: " <<
                                        " local " << (*i).localLocator->toString() <<
                                        " old " << oldLocalDescriptor.locator->toString() );

                        if( *((*i).localLocator) == *(oldLocalDescriptor.locator) ){

                                logging_debug("terminating connection to " << (*i).remoteLocator->toString() );
                                transport->terminate( oldLocalDescriptor.locator, (*i).remoteLocator );

                                (*i).localLocator = localDescriptor.locator;
                        }
                } // for( ; i != iend; i++ )

                // wait 500ms to give the sockets time to shut down
                usleep( 500000 );

        } else {

                logging_debug( "locator has not changed, not resetting connections" );

        }

        //
        // handle the connections that have no longer any
        // valid locator. send update messages with the new
        // locator,  so the remote node updates its locator/link mapping
        //

        LinkSet::iterator iAffected = linkSet.begin();
        LinkSet::iterator endAffected = linkSet.end();

        for( ; iAffected != endAffected; iAffected++ ){
                LinkDescriptor descr = *iAffected;
                logging_debug( "sending out link locator update to " << descr.remoteLocator->toString() );

                AribaBaseMsg updateMsg(         descr.remoteLocator,
                                                AribaBaseMsg::LINK_STATE_UPDATE,
                                                descr.localLink, descr.remoteLink );

                transport->sendMessage( &updateMsg );
        }
*/
}

/// sends a message to all end-points in the end-point descriptor
void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) {
        Data data = data_serialize( message, DEFAULT_V );
        transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8);
}

/// sends a message to the remote locator inside the link descriptor
void BaseCommunication::send(Message* message, const LinkDescriptor& desc) {
        Data data = data_serialize( message, DEFAULT_V );
        transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8);
}

}} // namespace ariba, communication