Changeset 12060 for source/ariba/communication

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.

Location:

source/ariba/communication

Files:

• BaseCommunication.cpp (modified) (23 diffs)
• BaseCommunication.h (modified) (16 diffs)
• CommunicationEvents.cpp (modified) (1 diff)
• CommunicationEvents.h (modified) (6 diffs)
• EndpointDescriptor.cpp (modified) (2 diffs)
• EndpointDescriptor.h (modified) (8 diffs)
• messages/AribaBaseMsg.h (modified) (4 diffs)
• messages/CMakeLists.txt (modified) (1 diff)
• networkinfo/AddressDiscovery.cpp (modified) (3 diffs)
• networkinfo/AddressDiscovery.h (modified) (1 diff)

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

source/ariba/communication/BaseCommunication.h

@@ -50 +50 @@
 #include <boost/foreach.hpp>
 
+#ifdef ECLIPSE_PARSER
+    #define foreach(a, b) for(a : b)
+#else
+    #define foreach(a, b) BOOST_FOREACH(a, b)
+#endif
+
 // utilities
 #include "ariba/utility/types.h"
-#include "ariba/utility/messages.h"
+#include "ariba/utility/messages/MessageReceiver.h"
 #include "ariba/utility/logging/Logging.h"
 #include "ariba/utility/misc/Demultiplexer.hpp"
@@ -58 +64 @@
 
 // new transport and addressing
-#include "ariba/utility/addressing/addressing.hpp"
-#include "ariba/utility/transport/transport.hpp"
-#include "ariba/utility/transport/transport_connection.hpp"
+#include "ariba/utility/transport/transport_peer.hpp"
+#include "ariba/utility/transport/interfaces/transport_connection.hpp"
+#include "ariba/utility/transport/interfaces/transport_listener.hpp"
+#include "ariba/utility/addressing2/endpoint.hpp"
 
 // communication
@@ -72 +79 @@
 #include "ariba/communication/networkinfo/NetworkInformation.h"
 
-// disabled
-//#ifndef UNDERLAY_OMNET
-//  #include "ariba/communication/modules/transport/tcp/TCPTransport.h"
-//  #include "ariba/communication/modules/network/ip/IPv4NetworkProtocol.h"
-//  using ariba::communication::IPv4NetworkProtocol;
-//  using ariba::communication::TCPTransport;
-//#endif
-
 namespace ariba {
-  class SideportListener;
+    class SideportListener;
 }
 
@@ -87 +86 @@
 namespace communication {
 
+
+class communication_message_not_sent: public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    explicit communication_message_not_sent(const string& __arg)  :
+        std::runtime_error(__arg)
+    {
+    }
+    
+    virtual ~communication_message_not_sent() throw() {}
+};
+
+
+
 using namespace std;
-using namespace ariba::addressing;
 using namespace ariba::transport;
 using namespace ariba::utility;
@@ -102 +115 @@
  * protocols and addressing schemes.
  *
- * @author Sebastian Mies, Christoph Mayer
+ * @author Sebastian Mies, Christoph Mayer, Mario Hock
  */
 class BaseCommunication:
         public NetworkChangeInterface,
-        public SystemEventListener,
         public transport_listener {
 
@@ -120 +132 @@
 
         /// Startup the base communication, start modules etc.
-        void start();
+        void start(addressing2::EndpointSetPtr listen_on);
 
         /// Stops the base communication, stop modules etc.
         void stop();
-
-        /// Sets the endpoints
-        void setEndpoints( string& endpoints );
 
         /// Check whether the base communication has been started up
@@ -147 +156 @@
          * @return A sequence number for this message
          */
-        seqnum_t sendMessage(const LinkID lid, const Message* message);
+        seqnum_t sendMessage(const LinkID& lid,
+                reboost::message_t message,
+                uint8_t priority,
+                bool bypass_overlay = false) throw(communication_message_not_sent);
 
         /**
@@ -164 +176 @@
          * @return List of LinkID
          */
-        LinkIDs getLocalLinks(const address_v* addr) const;
+//      LinkIDs getLocalLinks(const address_v* addr) const;  // XXX aktuell
 
         /**
@@ -187 +199 @@
 
         void unregisterEventListener(CommunicationEvents* _events);
-
-        /// called when a system event is emitted by system queue
-        virtual void handleSystemEvent(const SystemEvent& event);
 
         /**
@@ -196 +205 @@
          */ 
         virtual void receive_message(transport_connection::sptr connection,
-                reboost::message_t msg);
-
+                reboost::shared_buffer_t msg);
+
+    /**
+     * called within the ASIO thread
+     * when a connection is terminated (e.g. TCP close)
+     */ 
+    virtual void connection_terminated(transport_connection::sptr connection);
+
+    addressing2::EndpointPtr get_local_endpoint_of_link(const LinkID& linkid);
+    addressing2::EndpointPtr get_remote_endpoint_of_link(const LinkID& linkid);
+
+        
 protected:
 
@@ -205 +224 @@
          */ 
         void receiveMessage(transport_connection::sptr connection,
-                reboost::message_t msg);
+                reboost::shared_buffer_t message);
+        
+    /**
+     * called within the ARIBA thread (System Queue)
+     * when a connection is terminated (e.g. TCP close)
+     */ 
+    void connectionTerminated(transport_connection::sptr connection);
+        
 
         /// called when a network interface change happens
@@ -221 +247 @@
                 /// default constructor
                 LinkDescriptor() :
-                        localLink(LinkID::UNSPECIFIED), localLocator(NULL),
-                        remoteLink(LinkID::UNSPECIFIED), remoteLocator(NULL),
+                        localLink(LinkID::UNSPECIFIED),
+                        remoteLink(LinkID::UNSPECIFIED),
                         up(false) {
                 }
 
-                ~LinkDescriptor() {
-                        if (localLocator!=NULL)  delete localLocator;
-                        if (remoteLocator!=NULL) delete remoteLocator;
+                ~LinkDescriptor()
+                {
+                        if ( connection )
+                        {
+                            connection->unregister_communication_link(&localLink);
+                        }
                 }
 
@@ -240 +269 @@
                         return *unspec;
                 }
+                
+                
+                transport_connection::sptr get_connection() const
+                {
+                    return connection;
+                }
+                
+                void set_connection(const transport_connection::sptr& conn)
+                {
+                    // unregister from old connection,
+                    // if any (but normally there shouldn't..)
+                    if ( connection )
+                    {
+                        connection->unregister_communication_link(&localLink);
+                    }
+
+                    // * set_connection *
+                    connection = conn;
+                    
+                    // register this link with the connection
+                    conn->register_communication_link(&localLink);
+                }
 
                 bool unspecified;
@@ -245 +296 @@
                 /// link identifiers
                 LinkID localLink;
-                const address_v* localLocator;
+                addressing2::EndpointPtr localLocator;
 
                 /// used underlay addresses for the link
                 LinkID remoteLink;
-                const address_v* remoteLocator;
+                addressing2::EndpointPtr remoteLocator;
 
                 /// the remote end-point descriptor
-                EndpointDescriptor remoteEndpoint;
+                EndpointDescriptor remoteDescriptor;
 
                 /// flag, whether this link is up
                 bool up;
                 
+                
+        private:
                 /// connection if link is up
                 transport_connection::sptr connection;
@@ -281 +334 @@
         /// The local end-point descriptor
         EndpointDescriptor localDescriptor;
-
-#ifndef UNDERLAY_OMNET
+        
+        /**
+         * endpoint_set holding the addresses of the "server"-sockets,
+         * ---> that should be opened
+         * 
+         * (e.g. 0.0.0.0:41322)
+         */
+        addressing2::EndpointSetPtr listenOn_endpoints;
+        
+    /**
+     * endpoint_set holding the addresses of the "server"-sockets,
+     * ---> that are actually open
+     * 
+     * (e.g. 0.0.0.0:41322)
+     * 
+     * XXX should only be in transport_peer
+     */
+        addressing2::EndpointSetPtr active_listenOn_endpoints;
+        
+    /**
+     * endpoint_set holding the addresses of the "server"-sockets,
+     * ---> here the discovered "addressable" addresses are stored
+     * 
+     * (e.g. 192.168.0.5:41322)
+     * 
+     * XXX should only be in localDescriptor
+     */
+        addressing2::EndpointSetPtr local_endpoints;
+
         /// network change detector
         NetworkChangeDetection networkMonitor;
-#endif
 
         /// list of all remote addresses of links to end-points
-        class endpoint_reference {
-        public:
-                int count; ///< the number of open links to this end-point
-                const address_v* endpoint; ///< the end-point itself
-        };
-        vector<endpoint_reference> remote_endpoints;
-
-        /// adds an end-point to the list
-        void add_endpoint( const address_v* endpoint );
-
-        /// removes an end-point from the list
-        void remove_endpoint( const address_v* endpoint );
+        // XXX DEPRECATED
+//      class endpoint_reference {
+//      public:
+//              int count; ///< the number of open links to this end-point
+//              const address_v* endpoint; ///< the end-point itself
+//      };
+//      vector<endpoint_reference> remote_endpoints;
+
+        // XXX DEPRECATED
+//      /// adds an end-point to the list
+//      void add_endpoint( const address_v* endpoint );
+//
+//      /// removes an end-point from the list
+//      void remove_endpoint( const address_v* endpoint );
 
         /// event listener
@@ -314 +395 @@
         MessageReceiver* messageReceiver;
 
-        /// convenience: send message to peer
-        void send( Message* message, const EndpointDescriptor& endpoint );
-        void send( Message* message, const LinkDescriptor& descriptor );
+        
+        /* 
+         * Sends a message over an existing link.
+         *   ---> Adds »Link Message« Header
+         */
+        bool send_over_link(
+                const uint8_t type,
+                reboost::message_t message,
+                const LinkDescriptor& desc,
+                const uint8_t priority);
+
+    /* 
+     * Sends a message to a known peer. (To all known endpoints.)
+     *   ---> Adds »Peer Message« Header
+     */
+        void send_to_peer(
+                const uint8_t type,
+                const PeerID& peer_id,
+                reboost::message_t message,
+                const EndpointDescriptor& endpoint,
+                const uint8_t priority );
+        
 
         /// state of the base communication

source/ariba/communication/CommunicationEvents.cpp

@@ -49 +49 @@
 
 bool CommunicationEvents::onLinkRequest(const LinkID& id,
-        const address_v* local, const address_v* remote) {
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
         return true;
 }
 
-void CommunicationEvents::onLinkUp(const LinkID& id, const address_v* local,
-        const address_v* remote) {
+void CommunicationEvents::onLinkUp(const LinkID& id,
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote)
+{
 }
 
-void CommunicationEvents::onLinkDown(const LinkID& id, const address_v* local,
-        const address_v* remote) {
+void CommunicationEvents::onLinkDown(const LinkID& id,
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote)
+{
 }
 
 void CommunicationEvents::onLinkChanged(const LinkID& id,
-        const address_v* oldlocal, const address_v* newlocal,
-        const address_v* oldremote, const address_v* newremote) {
+        const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+        const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote)
+{
 }
 
-void CommunicationEvents::onLinkFail(const LinkID& id, const address_v* local,
-        const address_v* remote) {
+void CommunicationEvents::onLinkFail(const LinkID& id,
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote)
+{
 }
 
 void CommunicationEvents::onLinkQoSChanged(const LinkID& id,
-        const address_v* local, const address_v* remote, const QoSParameterSet& qos) {
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+        const QoSParameterSet& qos)
+{
 }
 

source/ariba/communication/CommunicationEvents.h

@@ -42 +42 @@
 #include "ariba/utility/types/LinkID.h"
 #include "ariba/utility/types/QoSParameterSet.h"
-#include "ariba/utility/addressing/addressing.hpp"
+#include "ariba/utility/addressing2/endpoint.hpp"
 
 namespace ariba {
@@ -49 +49 @@
 using ariba::utility::LinkID;
 using ariba::utility::QoSParameterSet;
-using namespace ariba::addressing;
 
 class CommunicationEvents {
@@ -68 +67 @@
          * @return True, if the link should be established
          */
-        virtual bool onLinkRequest(const LinkID& id, const address_v* local,
-                const address_v* remote);
+        virtual bool onLinkRequest(const LinkID& id,
+                const addressing2::EndpointPtr local,
+                const addressing2::EndpointPtr remote);
 
         /**
@@ -77 +77 @@
          * @param id The link id of the established link
          */
-        virtual void onLinkUp(const LinkID& id, const address_v* local,
-                const address_v* remote);
+        virtual void onLinkUp(const LinkID& id,
+                const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
 
         /**
@@ -85 +85 @@
          * @param id The link identifier of the dropped link
          */
-        virtual void onLinkDown(const LinkID& id, const address_v* local,
-                const address_v* remote);
+        virtual void onLinkDown(const LinkID& id,
+                const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
 
         /**
@@ -97 +97 @@
          */
         virtual void onLinkChanged(const LinkID& id,
-                const address_v* oldlocal,  const address_v* newlocal,
-                const address_v* oldremote, const address_v* newremote
-        );
+                const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+                const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote);
 
-        virtual void onLinkFail(const LinkID& id, const address_v* local,
-                const address_v* remote);
+        virtual void onLinkFail(const LinkID& id,
+                const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
 
-        virtual void onLinkQoSChanged(const LinkID& id, const address_v* local,
-                const address_v* remote, const QoSParameterSet& qos);
+        virtual void onLinkQoSChanged(const LinkID& id,
+                const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+                const QoSParameterSet& qos);
 };
 

source/ariba/communication/EndpointDescriptor.cpp

@@ -42 +42 @@
 namespace communication {
 
-vsznDefault(EndpointDescriptor);
+//vsznDefault(EndpointDescriptor);
 
-EndpointDescriptor::EndpointDescriptor(){
+EndpointDescriptor::EndpointDescriptor()  :
+        endpoints(new addressing2::endpoint_set())
+{
 }
 
@@ -55 +57 @@
 }
 
-EndpointDescriptor::EndpointDescriptor(const endpoint_set& endpoints ) :
-        endpoints(endpoints){
+EndpointDescriptor::EndpointDescriptor(
+        const PeerID& peer_id,
+        addressing2::EndpointSetPtr endpoints ) :
+    peerId(peer_id),
+        endpoints(endpoints)
+{
 }
 
-EndpointDescriptor::EndpointDescriptor(const string& str) : endpoints(str){
+EndpointDescriptor::EndpointDescriptor(const string& str)  :
+        endpoints(new addressing2::endpoint_set())
+{
+    cout << "ERROR: Construction of EndpointDescriptor from String is not functional!!" << endl;
+    assert( false );
 }
 
+
+reboost::message_t EndpointDescriptor::serialize() const
+{
+    reboost::message_t msg;
+    msg.push_back(peerId.serialize());
+    msg.push_back(endpoints->serialize());
+    
+    return msg;
+}
+
+reboost::shared_buffer_t EndpointDescriptor::deserialize(reboost::shared_buffer_t buff)
+{
+    buff = peerId.deserialize(buff);
+    buff = endpoints->deserialize(buff);
+    
+    return buff;
+}
+
+
 }} // namespace ariba, communication

source/ariba/communication/EndpointDescriptor.h

@@ -42 +42 @@
 #include <string>
 #include <set>
-#include "ariba/utility/serialization.h"
+//#include "ariba/utility/serialization.h"
 #include "ariba/utility/types/PeerID.h"
-#include "ariba/utility/addressing/endpoint_set.hpp"
+
+#include "ariba/utility/addressing2/endpoint_set.hpp"
+
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
+
 
 namespace ariba {
@@ -51 +56 @@
 using_serialization;
 using namespace std;
-using namespace ariba::addressing;
 using ariba::utility::PeerID;
 
-class EndpointDescriptor: public VSerializeable { VSERIALIZEABLE
-        friend class BaseCommunication;
+
+/**
+ * This class is used a transitions helper between the old addressing and
+ * serialization to the new addressing2 and the new message classes
+ * 
+ * Maybe it will be replaced, or at least modified in the future.
+ */
+//class EndpointDescriptor: public VSerializeable { VSERIALIZEABLE
+//    friend class BaseCommunication;
+class EndpointDescriptor
+{
+    friend class BaseCommunication;
 
 public:
@@ -68 +82 @@
 
         /// construct end-points from an endpoint set
-        EndpointDescriptor(const endpoint_set& endpoints );
+        EndpointDescriptor(const PeerID& peer_id, addressing2::EndpointSetPtr endpoints );
 
+        // FIXME NOT WORKING !!
         /// construct end-points from a string
         EndpointDescriptor(const string& str);
@@ -75 +90 @@
         /// convert end-points to string
         string toString() const {
-                return endpoints.to_string();
+                return endpoints->to_string();
         }
 
@@ -90 +105 @@
         }
 
-        /// create endpoint
-        static EndpointDescriptor* fromString(string str) {
-                return new EndpointDescriptor(str);
-        }
+//      /// create endpoint
+//      static EndpointDescriptor* fromString(string str) {
+//              return new EndpointDescriptor(str);
+//      }
 
         bool operator==(const EndpointDescriptor& rh) const {
@@ -113 +128 @@
 
         /// returns the end-points of this descriptor
-        endpoint_set& getEndpoints() {
+        addressing2::const_EndpointSetPtr getEndpoints() const {
                 return endpoints;
         }
-
-        /// returns the end-points of this descriptor
-        const endpoint_set& getEndpoints() const {
-                return endpoints;
+        
+        void replace_endpoint_set(addressing2::EndpointSetPtr new_endpoints)
+        {
+            endpoints = new_endpoints;
         }
-
+        
         /// returns a reference to the peer id
         PeerID& getPeerId() {
@@ -132 +147 @@
                 return peerId;
         }
+        
+        /// returns a message with peerId and endpoints in it
+        reboost::message_t serialize() const;
+        
+        /// deserialite peerId and endpoints
+        reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff);
+        
 private:
-        endpoint_set endpoints;
+        addressing2::EndpointSetPtr endpoints;
         PeerID peerId;
 };
@@ -139 +161 @@
 }} // namespace ariba, communication
 
-sznBeginDefault( ariba::communication::EndpointDescriptor, X ){
-
-        // serialize peer id
-        X && &peerId;
-
-        // serialize end-points
-        uint16_t len = endpoints.to_bytes_size();
-        X && len;
-        uint8_t* buffer = X.bytes( len );
-        if (buffer!=NULL) {
-                if (X.isDeserializer()) endpoints.assign(buffer,len);
-                else endpoints.to_bytes(buffer);
-        }
-}sznEnd();
+//sznBeginDefault( ariba::communication::EndpointDescriptor, X ){
+//
+//    // TODO
+//    assert(false);
+//    
+//      // serialize peer id
+//      X && &peerId;
+//
+//      // serialize end-points
+//      uint16_t len = endpoints.to_bytes_size();
+//      X && len;
+//      uint8_t* buffer = X.bytes( len );
+//      if (buffer!=NULL) {
+//              if (X.isDeserializer()) endpoints.assign(buffer,len);
+//              else endpoints.to_bytes(buffer);
+//      }
+//}sznEnd();
 
 #endif /*ENDPOINTDESCRIPTOR_H_*/

source/ariba/communication/messages/AribaBaseMsg.h

@@ -42 +42 @@
 #include <string>
 #include <boost/cstdint.hpp>
-#include "ariba/utility/messages.h"
+//#include "ariba/utility/messages.h"
+#include "ariba/utility/messages/Message.h"
 #include "ariba/utility/serialization.h"
 #include "ariba/utility/types/LinkID.h"
@@ -61 +62 @@
 using_serialization;
 
+// XXX This whole message is DEPRECATED
 class AribaBaseMsg : public Message {
         VSERIALIZEABLE;
@@ -69 +71 @@
                 typeLinkReply = 2,
                 typeLinkClose = 3,
-                typeLinkUpdate = 4
+                typeLinkUpdate = 4,
+                typeDirectData = 5
         };
 
@@ -115 +118 @@
 
 sznBeginDefault( ariba::communication::AribaBaseMsg, X ) {
-        X && type && &localLink && &remoteLink;
+        X && type && &remoteLink;
         if (type == typeLinkReply || type == typeLinkRequest)
-                X && localDescriptor && remoteDescriptor;
-        X && Payload();
+                X && &localLink && localDescriptor && remoteDescriptor;
+//      X && Payload();
 } sznEnd();
 

source/ariba/communication/messages/CMakeLists.txt

@@ -37 +37 @@
 # [License]
 
-add_headers(AribaBaseMsg.h)
+#add_headers(AribaBaseMsg.h)
 
-add_sources(AribaBaseMsg.cpp)
+#add_sources(AribaBaseMsg.cpp)

source/ariba/communication/networkinfo/AddressDiscovery.cpp

@@ -49 +49 @@
 #include <ifaddrs.h>
 
+#include <string>
+#include <boost/asio/ip/address.hpp>
+#include <boost/foreach.hpp>
+
+#include "ariba/utility/addressing2/tcpip_endpoint.hpp"
+#include "ariba/utility/addressing/mac_address.hpp"
+
 #ifdef HAVE_LIBBLUETOOTH
   #include <bluetooth/bluetooth.h>
@@ -58 +65 @@
 namespace communication {
 
-mac_address AddressDiscovery::getMacFromIF( const char* name ) {
+
+using namespace std;
+using namespace addressing2;
+using namespace boost::asio::ip;
+
+using ariba::addressing::mac_address;
+
+mac_address getMacFromIF( const char* name )
+{
         mac_address addr;
 #ifdef HAVE_LIBBLUETOOTH
@@ -73 +88 @@
 }
 
-int AddressDiscovery::dev_info(int s, int dev_id, long arg) {
-#ifdef HAVE_LIBBLUETOOTH
-        endpoint_set* set = (endpoint_set*)arg;
-        struct hci_dev_info di;
-        memset(&di, 0, sizeof(struct hci_dev_info));
-        di.dev_id = dev_id;
-        if (ioctl(s, HCIGETDEVINFO, (void *) &di)) return 0;
-        mac_address mac;
-        mac.bluetooth( di.bdaddr );
-        address_vf vf = mac;
-        set->add(vf);
+int dev_info(int s, int dev_id, long arg)
+{
+#ifdef HAVE_LIBBLUETOOTH
+//      endpoint_set* set = (endpoint_set*)arg;
+//      struct hci_dev_info di;
+//      memset(&di, 0, sizeof(struct hci_dev_info));
+//      di.dev_id = dev_id;
+//      if (ioctl(s, HCIGETDEVINFO, (void *) &di)) return 0;
+//      mac_address mac;
+//      mac.bluetooth( di.bdaddr );
+//      address_vf vf = mac;
+//      set->add(vf);
 #endif
         return 0;
 }
 
-void AddressDiscovery::discover_bluetooth( endpoint_set& endpoints ) {
-#ifdef HAVE_LIBBLUETOOTH
-        hci_for_each_dev(HCI_UP, &AddressDiscovery::dev_info, (long)&endpoints );
-#endif
-}
-
-void AddressDiscovery::discover_ip_addresses( endpoint_set& endpoints ) {
-        struct ifaddrs* ifaceBuffer = NULL;
-        void*           tmpAddrPtr  = NULL;
-
-        int ret = getifaddrs( &ifaceBuffer );
-        if( ret != 0 ) return;
-
-        for( struct ifaddrs* i=ifaceBuffer; i != NULL; i=i->ifa_next ) {
-
-                // ignore devices that are disabled or have no ip
-                if(i == NULL) continue;
-                struct sockaddr* addr = i->ifa_addr;
-                if (addr==NULL) continue;
-
-                // ignore tun devices
-                string device = string(i->ifa_name);
-                if(device.find_first_of("tun") == 0) continue;
-
-                if (addr->sa_family == AF_INET) {
-                        // look for ipv4
-                        char straddr[INET_ADDRSTRLEN];
-                        tmpAddrPtr= &((struct sockaddr_in*)addr)->sin_addr;
-                        inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
-                        ip_address ip = straddr;
-                        if (ip.is_loopback()) continue;
-                        address_vf vf = ip;
-                        endpoints.add( vf );
-                } else
-                if (addr->sa_family == AF_INET6) {
-                        // look for ipv6
-                        char straddr[INET6_ADDRSTRLEN];
-                        tmpAddrPtr= &((struct sockaddr_in6*)addr)->sin6_addr;
-                        inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
-                        ip_address ip = straddr;
-                        if (ip.is_loopback()) continue;
-//                      if (ip.is_link_local()) continue;
-                        address_vf vf = ip;
-                        endpoints.add( vf );
-                }
-        }
-
-        freeifaddrs(ifaceBuffer);
-}
-
-void AddressDiscovery::discover_endpoints( endpoint_set& endpoints ) {
-        discover_ip_addresses( endpoints );
-        discover_bluetooth( endpoints );
+void discover_bluetooth(
+        EndpointSetPtr listenOn_endpoints,
+        EndpointSetPtr discovered_endpoints )
+{
+#ifdef HAVE_LIBBLUETOOTH
+    // FIXME aktuell bluetooth
+//      hci_for_each_dev(HCI_UP, &AddressDiscovery::dev_info, (long)&endpoints );
+#endif
+}
+
+void discover_ip_addresses(
+        EndpointSetPtr listenOn_endpoints,
+        EndpointSetPtr discovered_endpoints )
+{
+    bool discover_ipv4 = false;
+    bool discover_ipv6 = false;
+    vector<uint16_t> ipv4_ports;
+    vector<uint16_t> ipv6_ports;
+    
+    /* analyze listenOn_endpoints */
+    BOOST_FOREACH( TcpIP_EndpointPtr endp, listenOn_endpoints->get_tcpip_endpoints() )
+    {
+        // BRANCH: IPv4 any [0.0.0.0]
+        if ( endp->to_asio().address() == address_v4::any() )
+        {
+            // add port
+            ipv4_ports.push_back(endp->to_asio().port());
+            
+            discover_ipv4 = true;
+        }
+
+        // BRANCH: IPv6 any [::]
+        else if ( endp->to_asio().address() == address_v6::any() )
+        {
+            // add port
+            ipv6_ports.push_back(endp->to_asio().port());
+            
+            discover_ipv6 = true;
+
+            
+            // NOTE: on linux the ipv6-any address [::] catches ipv4 as well
+            ipv4_ports.push_back(endp->to_asio().port());
+            discover_ipv4 = true;
+        }
+        
+        // BRANCH: explicit ip address
+        else
+        {
+            // ---> don't discover anything, just add it directly
+            discovered_endpoints->add_endpoint(endp);
+        }
+    }
+    
+    
+    /* discover addresses */
+    if ( discover_ipv4 || discover_ipv6 )
+    {
+        struct ifaddrs* ifaceBuffer = NULL;
+        void*           tmpAddrPtr  = NULL;
+    
+        int ret = getifaddrs( &ifaceBuffer );
+        if( ret != 0 ) return;
+    
+        for( struct ifaddrs* i=ifaceBuffer; i != NULL; i=i->ifa_next )
+        {
+            // ignore devices that are disabled or have no ip
+            if(i == NULL) continue;
+            struct sockaddr* addr = i->ifa_addr;
+            if (addr==NULL) continue;
+    
+    //          // ignore tun devices  // XXX why?
+    //          string device = string(i->ifa_name);
+    //          if(device.find_first_of("tun") == 0) continue;
+
+            // IPv4
+            if ( discover_ipv4 && addr->sa_family == AF_INET )
+            {
+                char straddr[INET_ADDRSTRLEN];
+                tmpAddrPtr= &((struct sockaddr_in*)addr)->sin_addr;
+                inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
+                
+                address ip_addr = address::from_string(straddr);
+    
+                // skip loopback address
+                if ( ip_addr.to_v4() == address_v4::loopback() )
+                    continue;
+
+                // add endpoint for this address and every given ipv4 port
+                BOOST_FOREACH( uint16_t port, ipv4_ports )
+                {
+                    tcp::endpoint tcpip_endp(ip_addr, port);
+                    TcpIP_EndpointPtr endp(new tcpip_endpoint(tcpip_endp));
+                    
+                    discovered_endpoints->add_endpoint(endp);
+                }
+            }
+            
+            // IPv6
+            else if ( discover_ipv6 && addr->sa_family == AF_INET6 )
+            {
+                // look for ipv6
+                char straddr[INET6_ADDRSTRLEN];
+                tmpAddrPtr= &((struct sockaddr_in6*)addr)->sin6_addr;
+                inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
+                
+                address ip_addr = address::from_string(straddr);
+                
+                // skip loopback address
+                if ( ip_addr.to_v6() == address_v6::loopback() )
+                    continue;
+
+                // add endpoint for this address and every given ipv4 port
+                BOOST_FOREACH( uint16_t port, ipv6_ports )
+                {
+                    tcp::endpoint tcpip_endp(ip_addr, port);
+                    TcpIP_EndpointPtr endp(new tcpip_endpoint(tcpip_endp));
+                    
+                    discovered_endpoints->add_endpoint(endp);
+                }
+            }
+        }
+    
+        freeifaddrs(ifaceBuffer);
+    }
+}
+
+
+
+EndpointSetPtr AddressDiscovery::discover_endpoints(EndpointSetPtr listenOn_endpoints)
+{
+    EndpointSetPtr discovered_endpoints(new addressing2::endpoint_set());
+    
+        discover_ip_addresses( listenOn_endpoints, discovered_endpoints );
+        discover_bluetooth( listenOn_endpoints, discovered_endpoints );
+        
+        return discovered_endpoints;
 }
 

source/ariba/communication/networkinfo/AddressDiscovery.h

@@ -40 +40 @@
 #define __ADDRESS_DISCOVERY_H
 
-#include "ariba/utility/addressing/addressing.hpp"
-
-using namespace ariba::addressing;
+#include "ariba/utility/addressing2/endpoint_set.hpp"
 
 namespace ariba {
 namespace communication {
 
+using addressing2::EndpointSetPtr;
+
 class AddressDiscovery {
 public:
-        static void discover_endpoints( endpoint_set& endpoints );
+        static EndpointSetPtr discover_endpoints(EndpointSetPtr listenOn_endpoints);
 
 private:
-        static mac_address getMacFromIF( const char* name );
-        static int dev_info(int s, int dev_id, long arg);
-        static void discover_bluetooth( endpoint_set& endpoints );
-        static void discover_ip_addresses( endpoint_set& endpoints );
+        // TODO aktuell weg damit..
+//      static mac_address getMacFromIF( const char* name );
+//      static int dev_info(int s, int dev_id, long arg);
+//      static void discover_bluetooth( EndpointSetPtr listenOn_endpoints, EndpointSetPtr discovered_endpoints );
+//      static void discover_ip_addresses( EndpointSetPtr listenOn_endpoints, EndpointSetPtr discovered_endpoints );
 };