Changeset 12060 for source/ariba/communication/BaseCommunication.cpp

Timestamp:

Jun 19, 2013, 11:05:49 AM (12 years ago)

Author:

hock@…

Message:

Reintegrate branch: 20130111-hock-message_classes

improvements:

• new message classes (reboost, zero-copy)
• "fast path" for direct links (skip overlay layer)
• link-properties accessible from the application
• SystemQueue can call boost::bind functions
• protlib compatibility removed (32bit overhead saved in every message)
• addressing2
• AddressDiscovery discoveres only addresses on which we're actually listening
• ariba serialization usage reduced (sill used in OverlayMsg)
• Node::connect, easier and cleaner interface to start-up ariba from the application
• ariba configs via JSON, XML, etc (boost::property_tree)
• keep-alive overhead greatly reduced
• (relayed) overlay links can actually be closed now
• lost messages are detected in most cases
• notification to the application when link is transformed into direct-link
• overlay routing: send message to second best hop if it would be dropped otherwise
• SequenceNumbers (only mechanisms, so for: upward compatibility)
• various small fixes

regressions:

• bluetooth is not yet working again
• bootstrap modules deactivated
• liblog4xx is not working (use cout-logging)

This patch brings great performance and stability improvements at cost of backward compatibility.
Also bluetooth and the bootstrap modules have not been ported to the new interfaces, yet.

File:

• source/ariba/communication/BaseCommunication.cpp (modified) (23 diffs)

source/ariba/communication/BaseCommunication.cpp

@@ -57 +57 @@
 namespace communication {
 
+using namespace ariba::addressing2;
+
 using ariba::utility::PeerID;
 using ariba::utility::SystemQueue;
 
 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).type_id() == endpoint->type_id() && (*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;
-                }
-        }
-}
 
 
@@ -100 +69 @@
           transport( NULL ),
           messageReceiver( NULL ),
-          started( false )
+          started( false ),
+          listenOn_endpoints(new addressing2::endpoint_set())
 {
 }
@@ -109 +79 @@
 
 
-void BaseCommunication::start() {
+void BaseCommunication::start(EndpointSetPtr listen_on) {
+    assert ( ! started );
+    
+    listenOn_endpoints = listen_on;
+    logging_info("Setting local end-points: " << listenOn_endpoints->to_string());
+    
         logging_info( "Starting up ..." );
         currentSeqnum = 0;
 
-        // set local peer id
-        localDescriptor.getPeerId() = PeerID::random();
-        logging_info( "Using PeerID: " << localDescriptor.getPeerId() );
-
         // creating transports
+        //  ---> transport_peer holds the set of the active endpoints we're listening on
         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
+        transport = new transport_peer();
+        active_listenOn_endpoints = transport->add_listenOn_endpoints(listenOn_endpoints);
+        logging_info( "XXX. Active endpoints = " << active_listenOn_endpoints->to_string() ); // XXX
 
         logging_info( "Searching for local locators ..." );
-        /**
-         * DONT DO THAT: if(localDescriptor.getEndpoints().to_string().length() == 0)
-         * since addresses are used to initialize transport addresses
-         */
-        AddressDiscovery::discover_endpoints( localDescriptor.getEndpoints() );
-        logging_info( "Done. Local endpoints = " << localDescriptor.toString() );
-
+        local_endpoints = AddressDiscovery::discover_endpoints(active_listenOn_endpoints);
+        if ( local_endpoints->count() > 0 )
+        {
+            logging_info( "Done. Discovered local endpoints: " << local_endpoints->to_string() );
+        }
+        else
+        {
+            logging_warn("WARING!! No local endpoints found, NO COMMUNICATION POSSIBLE!!");
+            
+            // TODO notify application, so that it may react properly. throw exception..?
+            assert( false );
+        }
+
+        
+    // create local EndpointDescriptor
+        //  ---> localDescriptor hold the set endpoints that can be used to reach us
+    localDescriptor.getPeerId() = PeerID::random();
+    localDescriptor.replace_endpoint_set(local_endpoints);
+    logging_info( "Using PeerID: " << localDescriptor.getPeerId() );
+
+    // start transport_peer
         transport->register_listener( this );
         transport->start();
 
-#ifndef UNDERLAY_OMNET
         // bind to the network change detection
         networkMonitor.registerNotification( this );
-#endif
 
         // base comm startup done
@@ -163 +140 @@
 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());
 }
 
@@ -193 +163 @@
         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;
-        baseMsg.getRemoteDescriptor().getPeerId() = descriptor.getPeerId();
-
-        // serialize and send message
-        send( &baseMsg, descriptor );
+
+    /* send a message to request new link to remote */
+    logging_debug( "Send messages with request to open link to " << descriptor.toString() );
+
+    /*
+         * Create Link-Request Message:
+         * NOTE: - Their PeerID (in parent message)
+         * - Our LinkID
+         * - Our PeerID
+         * - Our EndpointDescriptor
+         */
+        reboost::message_t linkmsg;
+        linkmsg.push_back(linkid.serialize());
+        linkmsg.push_back(localDescriptor.getPeerId().serialize());
+        linkmsg.push_back(localDescriptor.endpoints->serialize());
+        
+//      // XXX AKTUELL BUG FINDING...
+//      reboost::shared_buffer_t xxx = localDescriptor.endpoints->serialize();
+//      EndpointSetPtr xxx_set = endpoint_set::create_EndpointSet();
+//      xxx_set->deserialize(xxx);
+//      cout << "/// MARIO VORHER: " << localDescriptor.endpoints->to_string() << endl;
+//      cout << "/// MARIO NACHHER: " << xxx_set->to_string() << endl;
+        
+        // send message
+        // TODO move enum to BaseComm
+        send_to_peer(AribaBaseMsg::typeLinkRequest, descriptor.getPeerId(), linkmsg,
+                descriptor, system_priority::OVERLAY);
 
         return linkid;
@@ -217 +206 @@
 
         // tell the registered listeners
-        BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
+        foreach( CommunicationEvents* i, eventListener ) {
                 i->onLinkDown( link, ld.localLocator, ld.remoteLocator );
         }
 
-        // create message to drop the link
+
+        // * send 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 );
+        reboost::message_t empty_message;
+        send_over_link( AribaBaseMsg::typeLinkClose, empty_message, ld, system_priority::OVERLAY );
 
         // remove from map
@@ -232 +220 @@
 }
 
-seqnum_t BaseCommunication::sendMessage( const LinkID lid, const Message* message) {
-
-        logging_debug( "Sending out message to link " << lid.toString() );
-
+
+seqnum_t BaseCommunication::sendMessage( const LinkID& lid,
+        reboost::message_t message,
+        uint8_t priority,
+        bool bypass_overlay) throw(communication_message_not_sent)
+{
+    // message type: direct data or (normal) data 
+    AribaBaseMsg::type_ type;
+    if ( bypass_overlay )
+    {
+        type = AribaBaseMsg::typeDirectData;
+        logging_debug( "Sending out direct-message to link " << lid.toString() );
+    }
+    else
+    {
+        type = AribaBaseMsg::typeData;
+        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;
+        if( ld.isUnspecified() )
+        {
+            throw communication_message_not_sent("Don't know the link with id "
+                    + lid.toString());
         }
 
         // 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
+        if( !ld.up )
+        {
+            throw communication_message_not_sent("Can not send on link "
+                    + lid.toString() + ": link not up");
+        }
+
+
+        // * send message *
+        bool okay = send_over_link( type, message, ld, priority );
+
+        if ( ! okay )
+        {
+            throw communication_message_not_sent("send_over_link failed!");
+        }
+        
         return ++currentSeqnum;
 }
@@ -268 +273 @@
                 LinkDescriptor& linkDesc = queryLocalLink(link);
                 if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
-                return linkDesc.remoteEndpoint;
+                return linkDesc.remoteDescriptor;
         }
 }
@@ -283 +288 @@
 }
 
-SystemEventType TransportEvent("Transport");
-SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent );
-
-/// 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" );
-                boost::function0<void>* handler = event.getData< boost::function0<void> >();
-                (*handler)();
-                delete handler;
-        }
-}
-
-/**
- * called within the ASIO thread
- * when a message is received from underlay transport
- */ 
+
+
+/*------------------------------
+ | ASIO thread --> SystemQueue |
+ ------------------------------*/
+
+/// ASIO thread
 void BaseCommunication::receive_message(transport_connection::sptr connection,
-        reboost::message_t msg) {
+        reboost::shared_buffer_t msg) {
 
         logging_debug( "Dispatching message" );
         
-    boost::function0<void>* handler = new boost::function0<void>(
+    SystemQueue::instance().scheduleCall(
             boost::bind(
                     &BaseCommunication::receiveMessage,
@@ -313 +306 @@
                     connection,
                     msg)
-    );
-    
-    SystemQueue::instance().scheduleEvent(
-        SystemEvent(this, MessageDispatchEvent, handler)
-    );
-}
-
-/**
- * called within the ARIBA thread (System Queue)
- * when a message is received from underlay transport
- */ 
+        );
+}
+
+/// ASIO thread
+void BaseCommunication::connection_terminated(transport_connection::sptr connection)
+{
+    SystemQueue::instance().scheduleCall(
+            boost::bind(
+                    &BaseCommunication::connectionTerminated,
+                    this,
+                    connection)
+        );
+}
+
+/*--------------------------------
+ | [ASIO thread --> SystemQueue] |
+ -------------------------------*/
+
+/// ARIBA thread (System Queue)
+void BaseCommunication::connectionTerminated(transport_connection::sptr connection)
+{
+    vector<LinkID*> links = connection->get_communication_links();
+    
+    logging_debug("[BaseCommunication] Connection terminated: "
+            << connection->getLocalEndpoint()->to_string()
+            << " <--> " << connection->getRemoteEndpoint()->to_string()
+            << " (" << links.size() << " links)");
+    
+    // remove all links that used the terminated connection
+    for ( vector<LinkID*>::iterator it = links.begin(); it != links.end(); ++it )
+    {
+        LinkID& link_id = **it;
+        
+        logging_debug("  ---> Removing link: " << link_id.toString());
+        
+        // searching for link, not found-> warn
+        LinkDescriptor& linkDesc = queryLocalLink( link_id );
+        if (linkDesc.isUnspecified()) {
+            logging_warn("Failed to find local link " << link_id.toString());
+            continue;
+        }
+
+        // inform listeners
+        foreach( CommunicationEvents* i, eventListener ){
+            i->onLinkFail( linkDesc.localLink,
+                    linkDesc.localLocator, linkDesc.remoteLocator );
+        }
+
+        // remove the link descriptor
+        removeLink( link_id );
+    }
+}
+
+/// ARIBA thread (System Queue)
 void BaseCommunication::receiveMessage(transport_connection::sptr connection,
-        reboost::message_t message)
+        reboost::shared_buffer_t message)
 {
-    
-    //// Adapt to old message system ////
-    // Copy data
-    size_t bytes_len = message.size();
-    uint8_t* bytes = new uint8_t[bytes_len];
-    message.read(bytes, 0, bytes_len);
-    
-    Data data(bytes, bytes_len * 8);
-    
-    Message legacy_message;
-    legacy_message.setPayload(data);
-    
-    
-    
-        /// decapsulate message
-        AribaBaseMsg* msg = legacy_message.decapsulate<AribaBaseMsg>();
-        logging_debug( "Receiving message of type " << msg->getTypeString() );
-
+    // XXX
+    logging_debug("/// [receiveMessage] buffersize: " << message.size());
+        
+    // get type
+    uint8_t type = message.data()[0];
+    reboost::shared_buffer_t sub_buff = message(1);
+    
+    // get link id
+    LinkID link_id;
+    if ( type != AribaBaseMsg::typeLinkRequest)
+    {
+        sub_buff = link_id.deserialize(sub_buff);
+    }
+    
         // handle message
-        switch (msg->getType()) {
-
+        switch ( type )
+        {
                 // ---------------------------------------------------------------------
                 // data message
                 // ---------------------------------------------------------------------
-                case AribaBaseMsg::typeData: {
-                        logging_debug( "Received data message, forwarding to overlay" );
-                        if( messageReceiver != NULL ) {
+                case AribaBaseMsg::typeData:
+                {
+                        logging_debug( "Received data message, forwarding to overlay." );
+                        if( messageReceiver != NULL )
+                        {
                                 messageReceiver->receiveMessage(
-                                        msg, msg->getRemoteLink(), NodeID::UNSPECIFIED
+                                        sub_buff, link_id, NodeID::UNSPECIFIED, false
                                 );
                         }
+                        
                         break;
                 }
 
+        // ---------------------------------------------------------------------
+        // direct data message (bypass overlay-layer)
+        // ---------------------------------------------------------------------
+        case AribaBaseMsg::typeDirectData:
+        {
+            logging_debug( "Received direct data message, forwarding to application." );
+            
+            if( messageReceiver != NULL )
+            {
+                messageReceiver->receiveMessage(
+                    sub_buff, link_id, NodeID::UNSPECIFIED, true
+                );
+            }
+            
+            break;
+        }
+
+
+        
                 // ---------------------------------------------------------------------
                 // handle link request from remote
                 // ---------------------------------------------------------------------
-                case AribaBaseMsg::typeLinkRequest: {
-                        logging_debug( "Received link open request" );
-
-                        /// not the correct peer id-> skip request
-                        if (!msg->getRemoteDescriptor().getPeerId().isUnspecified()
-                                && msg->getRemoteDescriptor().getPeerId() != localDescriptor.getPeerId()) {
-                                logging_info("Received link request for "
-                                        << msg->getRemoteDescriptor().getPeerId().toString()
-                                        << "but i'm "
-                                        << localDescriptor.getPeerId()
-                                        << ": Ignoring!");
-                                break;
-                        }
-
+                case AribaBaseMsg::typeLinkRequest:
+                {
+                        logging_debug( "Received link open request on "
+                                << connection->getLocalEndpoint()->to_string() );
+                        
+                        /*
+                         * Deserialize Peer Message
+                         * - Our PeerID 
+                         */
+                        PeerID our_peer_id;
+                        sub_buff = our_peer_id.deserialize(sub_buff);
+
+            /// not the correct peer id-> skip request
+            if ( our_peer_id != localDescriptor.getPeerId() &&
+                    ! our_peer_id.isUnspecified() /* overlay bootstrap */ )
+            {
+                logging_info("Received link request for "
+                    << our_peer_id.toString()
+                    << "but i'm "
+                    << localDescriptor.getPeerId()
+                    << ": Ignoring!");
+                
+                // TODO terminate connection?
+                
+                break;
+            }
+
+            
+                    /*
+                     * Deserialize Link-Request Message:
+                     * - Their LinkID
+                     * - Their PeerID
+                     * - Their EndpointDescriptor
+                     */
+            LinkID their_link_id;
+            PeerID their_peer_id;
+            EndpointSetPtr their_endpoints = endpoint_set::create_EndpointSet();
+            sub_buff = their_link_id.deserialize(sub_buff);
+            sub_buff = their_peer_id.deserialize(sub_buff);
+            sub_buff = their_endpoints->deserialize(sub_buff);
+            /* [ Deserialize Link-Request Message ] */
+
+            
                         /// only answer the first request
-                        if (!queryRemoteLink(msg->getLocalLink()).isUnspecified()) {
+                        if (!queryRemoteLink(their_link_id).isUnspecified())
+                        {
+                            
+                            // TODO aktuell: When will these connections be closed?
+                            // ---> Close it now (if it services no links) ?
+                            //   (see also ! allowlink below)
+                            
+                            // XXX AKTUELL TESTING !! This will cause race conditions. So this is test-code only!
+                            if ( connection->get_communication_links().size() == 0 )
+                            {
+                                connection->terminate();
+                            }
+                            
                                 logging_debug("Link request already received. Ignore!");
                                 break;
@@ -386 +481 @@
                         /// create link ids
                         LinkID localLink  = LinkID::create();
-                        LinkID remoteLink = msg->getLocalLink();
+                        LinkID remoteLink = their_link_id;  // XXX intermediate variable is unnecessary
                         logging_debug(
                                 "local=" << connection->getLocalEndpoint()->to_string()
@@ -394 +489 @@
                         // check if link creation is allowed by ALL listeners
                         bool allowlink = true;
-                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
+                        foreach( CommunicationEvents* i, eventListener ){
                                 allowlink &= i->onLinkRequest( localLink,
                                         connection->getLocalEndpoint(),
@@ -403 +498 @@
                         if( !allowlink ){
                                 logging_warn( "Overlay denied creation of link" );
-                                delete msg;
                                 return;
                         }
@@ -411 +505 @@
                         ld->localLink = localLink;
                         ld->remoteLink = remoteLink;
-                        ld->localLocator = connection->getLocalEndpoint()->clone();
-                        ld->remoteLocator = connection->getRemoteEndpoint()->clone();
-                        ld->connection = connection;
-                        ld->remoteEndpoint = msg->getLocalDescriptor();
-                        add_endpoint(ld->remoteLocator);
-
-                        // add layer 1-3 addresses
-                        ld->remoteEndpoint.getEndpoints().add(
-                                ld->remoteLocator, endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
-                        localDescriptor.getEndpoints().add(
-                                connection->getLocalEndpoint(),
-                                endpoint_set::Layer1_3 | endpoint_set::NoLoopback);
+                        ld->localLocator = connection->getLocalEndpoint();
+                        ld->remoteLocator = connection->getRemoteEndpoint();
+                        ld->remoteDescriptor = EndpointDescriptor(their_peer_id, their_endpoints);
+                        ld->set_connection(connection);
+
+            
+                        // update endpoints (should only have any effect in case of NAT)
+                        ld->remoteDescriptor.endpoints->add_endpoint(connection->getRemoteEndpoint());
+//                      localDescriptor.endpoints->add_endpoint(connection->getLocalEndpoint());  // XXX 0.0.0.0:0
 
                         // link is now up-> add it
@@ -428 +519 @@
                         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 );
+
+                        
+            /* sending link reply */
+            logging_debug( "Sending link reply with ids "
+                << "local=" << localLink.toString() << ", "
+                << "remote=" << remoteLink.toString() );
+
+                    /*
+                     * Create Link-Reply Message:
+                     * - Our LinkID
+                     * - Our Endpoint_Set (as update)
+                     * - Their EndpointDescriptor (maybe they learn something about NAT)
+                     */
+                    reboost::message_t linkmsg;
+                    linkmsg.push_back(localLink.serialize());
+                    linkmsg.push_back(localDescriptor.endpoints->serialize());
+                    linkmsg.push_back(ld->remoteDescriptor.endpoints->serialize());
+                    
+                    // XXX
+                    cout << "/// MARIO: " << ld->get_connection()->getRemoteEndpoint()->to_string() << endl;
+                    
+                    // send message
+                        bool sent = send_over_link( AribaBaseMsg::typeLinkReply, linkmsg, *ld, system_priority::OVERLAY );
+                        
+                        if ( ! sent )
+                        {
+                            logging_error("ERROR: Could not send LinkReply to: " << ld->remoteLocator->to_string());
+                            
+                            // TODO remove link, close link, ..?
+                            
+                            break;
+                        }
+
+
+            // 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() << ")"
+            );
 
                         // inform listeners about new open link
-                        BOOST_FOREACH( CommunicationEvents* i, eventListener ) {
+                        foreach( CommunicationEvents* i, eventListener ) {
                                 i->onLinkUp( localLink, ld->localLocator, ld->remoteLocator);
                         }
@@ -458 +573 @@
                 // handle link request reply
                 // ---------------------------------------------------------------------
-                case AribaBaseMsg::typeLinkReply: {
+                case AribaBaseMsg::typeLinkReply:
+                {
                         logging_debug( "Received link open reply for a link we initiated" );
 
+            /*
+             * Deserialize Link-Reply Message:
+             * - Their LinkID
+             * - Their Endpoint_Set (as update)
+             * - Our EndpointDescriptor (maybe we can learn something about NAT)
+             */
+                        LinkID their_link_id;
+                        EndpointSetPtr their_endpoints = endpoint_set::create_EndpointSet();
+                        EndpointSetPtr our_endpoints = endpoint_set::create_EndpointSet();
+                        sub_buff = their_link_id.deserialize(sub_buff);
+                        sub_buff = their_endpoints->deserialize(sub_buff);
+                        sub_buff = our_endpoints->deserialize(sub_buff);
+
+            
                         // 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() );
+                        LinkDescriptor& ld = queryLocalLink( link_id );
 
                         // no link found-> warn!
                         if (ld.isUnspecified()) {
-                                logging_warn("Failed to find local link " << msg->getRemoteLink().toString());
-                                delete msg;
+                                logging_warn("Failed to find local link " << link_id.toString());
                                 return;
                         }
+                        
+                        if ( ld.up )
+                        {
+                logging_warn("Got link replay for already open link. Ignore. LinkID: " << link_id.toString());
+                
+                // TODO send LinkClose ?
+                return;                     
+                        }
 
                         // store the connection
-                        ld.connection = connection;
+                        ld.set_connection(connection);
                         
                         // set remote locator and link id
-                        ld.remoteLink = msg->getLocalLink();
-                        ld.remoteLocator = connection->getRemoteEndpoint()->clone();
-                        ld.remoteEndpoint.getEndpoints().add(
-                                                        msg->getLocalDescriptor().getEndpoints(),
-                                                        endpoint_set::Layer1_4
-                                                );
-
-                        localDescriptor.getEndpoints().add(
-                                msg->getRemoteDescriptor().getEndpoints(),
-                                endpoint_set::Layer1_3
-                        );
+                        ld.remoteLink = their_link_id;
+                        ld.remoteLocator = connection->getRemoteEndpoint();
+                        
+                        
+                        /* Update endpoints */
+                        // NOTE: we might loose some information here, but it's our only chance to get rid of outdated information.
+                        ld.remoteDescriptor.replace_endpoint_set(their_endpoints);
+                        
+                        // add actual remote endpoint to this set (should only have any effect in case of NAT)
+                        ld.remoteDescriptor.endpoints->add_endpoint(connection->getRemoteEndpoint());
+                        
+                        // TODO In case of NAT, we could learn something about our external IP.
+                        //   ---> But we must trust the remote peer about this information!!  
+//                      localDescriptor.endpoints->add_endpoints(our_endpoints);
+                        
+                        
+                        
+                        
                         ld.up = true;
-                        add_endpoint(ld.remoteLocator);
 
                         logging_debug( "Link is now up with local id "
@@ -496 +639 @@
 
                         // inform lisneters about link up event
-                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
+                        foreach( CommunicationEvents* i, eventListener ){
                                 i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
                         }
@@ -509 +652 @@
                 case AribaBaseMsg::typeLinkClose: {
                         // get remote link
-                        const LinkID& localLink = msg->getRemoteLink();
-                        logging_debug( "Received link close request for link " << localLink.toString() );
+//                      const LinkID& localLink = msg.getRemoteLink();
+                        logging_debug( "Received link close request for link " << link_id.toString() );
 
                         // searching for link, not found-> warn
-                        LinkDescriptor& linkDesc = queryLocalLink( localLink );
+                        LinkDescriptor& linkDesc = queryLocalLink( link_id );
                         if (linkDesc.isUnspecified()) {
-                                logging_warn("Failed to find local link " << localLink.toString());
-                                delete msg;
+                                logging_warn("Failed to find local link " << link_id.toString());
                                 return;
                         }
 
                         // inform listeners
-                        BOOST_FOREACH( CommunicationEvents* i, eventListener ){
+                        foreach( CommunicationEvents* i, eventListener ){
                                 i->onLinkDown( linkDesc.localLink,
                                                 linkDesc.localLocator, linkDesc.remoteLocator );
@@ -527 +669 @@
 
                         // remove the link descriptor
-                        removeLink( localLink );
+                        removeLink( link_id );
 
                         // done
@@ -534 +676 @@
 
                 // ---------------------------------------------------------------------
-                // handle link locator changes
+                // handle link locator changes  -- TODO is this ever called..?
                 // ---------------------------------------------------------------------
-                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());
-                                delete msg;
-                                return;
-                        }
-
-                        // update the remote locator
-                        const address_v* oldremote = linkDesc.remoteLocator;
-                        linkDesc.remoteLocator = connection->getRemoteEndpoint()->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;
-                }
-        }
-
-        delete msg;
+//              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
+//                      addressing2::EndpointPtr oldremote = linkDesc.remoteLocator;
+//                      linkDesc.remoteLocator = connection->getRemoteEndpoint();
+//                      
+//                      // TODO update linkDesc.connection ?
+//
+//                      // inform the listeners (local link has _not_ changed!)
+//                      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;
+//              }
+                
+                
+        default: {
+            logging_warn( "Received unknown message type!" );
+            break;
+        }
+
+        }
 }
 
@@ -582 +730 @@
         for(LinkSet::iterator i=linkSet.begin(); i != linkSet.end(); i++){
                 if( (*i)->localLink != localLink) continue;
-                remove_endpoint((*i)->remoteLocator);
+//              remove_endpoint((*i)->remoteLocator);  // XXX
                 delete *i;
                 linkSet.erase( i );
@@ -605 +753 @@
 }
 
-LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
-        LinkIDs ids;
-        for (size_t 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;
-}
+//LinkIDs BaseCommunication::getLocalLinks( const address_v* addr ) const {
+//      LinkIDs ids;
+//      for (size_t 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!
@@ -762 +903 @@
 }
 
-/// sends a message to all end-points in the end-point descriptor
-void BaseCommunication::send(Message* legacy_message, const EndpointDescriptor& endpoint) {
-        Data data = data_serialize(legacy_message, DEFAULT_V);
-        
-        //// Adapt to new message system ////
-        // transfer data buffer ownership to the shared_buffer
-    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
-        
-        reboost::message_t message;
-        message.push_back(buf);
-        
-        transport->send(endpoint.getEndpoints(), message);
-}
-
-/// sends a message to the remote locator inside the link descriptor
-void BaseCommunication::send(Message* legacy_message, const LinkDescriptor& desc) {
-        if (desc.remoteLocator==NULL) return;
-        
-        Data data = data_serialize(legacy_message, DEFAULT_V);
-    
-    //// Adapt to new message system ////
-    // transfer data buffer ownership to the shared_buffer
-    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
-    
-    reboost::message_t message;
-    message.push_back(buf);
-    
-        desc.connection->send(message);
-}
+
+addressing2::EndpointPtr BaseCommunication::get_local_endpoint_of_link(
+        const LinkID& linkid)
+{
+    LinkDescriptor& ld = queryLocalLink(linkid);
+    
+    return ld.get_connection()->getLocalEndpoint();
+}
+
+addressing2::EndpointPtr BaseCommunication::get_remote_endpoint_of_link(
+        const LinkID& linkid)
+{
+    LinkDescriptor& ld = queryLocalLink(linkid);
+    
+    return ld.get_connection()->getRemoteEndpoint();
+}
+
+
+
+bool BaseCommunication::send_over_link(
+        const uint8_t type,
+        reboost::message_t message,
+        const LinkDescriptor& desc,
+        const uint8_t priority)
+{
+    /*
+     * Create Link Message:
+     * - Type
+     * - Their LinkID
+     */
+    // link id
+    message.push_front(desc.remoteLink.serialize());
+    // type
+    memcpy(message.push_front(sizeof(uint8_t)).mutable_data(), &type, sizeof(uint8_t));
+    /* [ Create Link Message ] */
+    
+    
+    /* send message */
+    transport_connection::sptr conn = desc.get_connection();
+    if ( ! conn )
+    {
+        cout << "/// MARIO: No connection!!" << endl;  // XXX debug
+        return false;
+    }
+    
+    // * send over connection *
+    return conn->send(message, priority);
+}
+
+void BaseCommunication::send_to_peer(
+        const uint8_t type,
+        const PeerID& peer_id,
+        reboost::message_t message,
+        const EndpointDescriptor& endpoint,
+        const uint8_t priority )
+{
+    /*
+     * Create Peer Message:
+     * - Type
+     * - Their PeerID
+     */
+    // peer id
+    message.push_front(peer_id.serialize());
+    // type
+    memcpy(message.push_front(sizeof(uint8_t)).mutable_data(), &type, sizeof(uint8_t));
+    
+    
+    /* send message */
+    transport->send(endpoint.getEndpoints(), message, priority);
+}
+
+
+
 
 }} // namespace ariba, communication