Index: source/ariba/overlay/modules/chord/Chord.cpp
===================================================================
--- source/ariba/overlay/modules/chord/Chord.cpp	(revision 10572)
+++ source/ariba/overlay/modules/chord/Chord.cpp	(revision 12060)
@@ -43,5 +43,5 @@
 #include "detail/chord_routing_table.hpp"
 
-#include "messages/Discovery.h"
+//#include "messages/Discovery.h"  // XXX DEPRECATED
 
 namespace ariba {
@@ -55,5 +55,60 @@
 typedef chord_routing_table::item route_item;
 
+using ariba::transport::system_priority;
+
 use_logging_cpp( Chord );
+
+
+////// Messages
+struct DiscoveryMessage
+{
+    /**
+     * DiscoveryMessage
+     * - type
+     * - data
+     * - Endpoint
+     */
+
+    // type enum
+    enum type_ {
+        invalid = 0,
+        normal = 1,
+        successor = 2,
+        predecessor = 3
+    };
+
+    
+    // data
+    uint8_t type;
+    uint8_t ttl;
+    EndpointDescriptor endpoint;
+
+    // serialize
+    reboost::message_t serialize()
+    {
+        // serialize endpoint
+        reboost::message_t msg = endpoint.serialize();
+        
+        // serialize type and ttl
+        uint8_t* buff1 = msg.push_front(2*sizeof(uint8_t)).mutable_data();
+        buff1[0] = type;
+        buff1[1] = ttl;
+        
+        return msg;
+    }
+    
+    //deserialize
+    reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff)
+    {
+        // deserialize type and ttl
+        const uint8_t* bytes = buff.data();
+        type = bytes[0];
+        ttl = bytes[1];
+        
+        // deserialize endpoint
+        return endpoint.deserialize(buff(2*sizeof(uint8_t)));
+    }
+};
+
 
 Chord::Chord(BaseOverlay& _baseoverlay, const NodeID& _nodeid,
@@ -102,7 +157,22 @@
 
 /// helper: sends a message using the "base overlay"
-seqnum_t Chord::send( OverlayMsg* msg, const LinkID& link ) {
-	if (link.isUnspecified()) return 0;
-	return baseoverlay.send_link( msg, link );
+void Chord::send( OverlayMsg* msg, const LinkID& link ) {
+	if (link.isUnspecified())
+        return;
+    
+	baseoverlay.send_link( msg, link, system_priority::OVERLAY );
+}
+
+void Chord::send_node( OverlayMsg* message, const NodeID& remote )
+{
+    try
+    {
+        baseoverlay.send( message, remote, system_priority::OVERLAY );
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Chord: Could not send message to " << remote
+                << ": " << e.what());
+    }
 }
 
@@ -116,20 +186,44 @@
 	OverlayMsg msg( typeDiscovery );
 	msg.setRegisterRelay(true);
-	Discovery dmsg( Discovery::normal, (uint8_t)ttl, baseoverlay.getEndpointDescriptor() );
-	msg.encapsulate(&dmsg);
+	
+	// create DiscoveryMessage
+	DiscoveryMessage dmsg;
+	dmsg.type = DiscoveryMessage::normal;
+	dmsg.ttl = ttl;
+	dmsg.endpoint = baseoverlay.getEndpointDescriptor();
+
+	msg.set_payload_message(dmsg.serialize());
 
 	// send to node
-	baseoverlay.send_node( &msg, remote );
+    try
+    {
+        baseoverlay.send_node( &msg, remote, system_priority::OVERLAY );
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Chord: Could not send message to " << remote
+                << ": " << e.what());
+    }
 }
 
 void Chord::discover_neighbors( const LinkID& link ) {
 	uint8_t ttl = 1;
+	
+    // FIXME try-catch for the send operations
+    
+    // create DiscoveryMessage
+    DiscoveryMessage dmsg;
+    dmsg.ttl = ttl;
+    dmsg.endpoint = baseoverlay.getEndpointDescriptor();
 	{
 		// send predecessor discovery
 		OverlayMsg msg( typeDiscovery );
 		msg.setRegisterRelay(true);
-		Discovery dmsg( Discovery::predecessor, ttl,
-			baseoverlay.getEndpointDescriptor() );
-		msg.encapsulate(&dmsg);
+
+		// set type
+	    dmsg.type = DiscoveryMessage::predecessor;
+
+	    // send
+	    msg.set_payload_message(dmsg.serialize());
 		send(&msg, link);
 	}
@@ -137,9 +231,12 @@
 		// send successor discovery
 		OverlayMsg msg( typeDiscovery );
-		msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );
+//		msg.setSourceEndpoint( baseoverlay.getEndpointDescriptor() );  // XXX this was redundand, wasn't it?
 		msg.setRegisterRelay(true);
-		Discovery dmsg( Discovery::successor, ttl,
-			baseoverlay.getEndpointDescriptor() );
-		msg.encapsulate(&dmsg);
+
+		// set type
+        dmsg.type = DiscoveryMessage::successor;
+
+        // send
+        msg.set_payload_message(dmsg.serialize());
 		send(&msg, link);
 	}
@@ -163,5 +260,6 @@
 
 	// timer for stabilization management
-	Timer::setInterval(1000);
+//	Timer::setInterval(1000);  // TODO find an appropriate interval!
+	Timer::setInterval(10000);  // XXX testing...
 	Timer::start();
 }
@@ -200,4 +298,48 @@
 	return item->info;
 }
+
+std::vector<const LinkID*> Chord::getSortedLinkIdsTowardsNode(
+    const NodeID& id, int num ) const
+{
+    std::vector<const LinkID*> ret;
+    
+    switch ( num )
+    {
+        // special case: just call Â»getNextLinkIdÂ«
+        case 1:
+        {
+            ret.push_back(&getNextLinkId(id));
+            
+            break;
+        }
+        
+        // * calculate top 2 *
+        case 0:
+        case 2:
+        {
+            std::vector<const route_item*> items = table->get_next_2_hops(id);
+            
+            ret.reserve(items.size());
+            
+            BOOST_FOREACH( const route_item* item, items )
+            {
+                ret.push_back(&item->info);
+            }
+            
+            break;
+        }
+        
+        // NOTE: implement real sorting, if needed (and handle "case 0" properly, then)
+        default:
+        {
+            throw std::runtime_error("Not implemented. (Chord::getSortedLinkIdsTowardsNode with num != 2)");
+            
+            break;
+        }
+    }
+    
+    return ret;
+}
+
 
 /// @see OverlayInterface.h
@@ -253,4 +395,5 @@
 	if (remote==nodeid) {
 		logging_warn("dropping link that has been established to myself (nodes have same nodeid?)");
+		logging_warn("NodeID: " << remote);
 		baseoverlay.dropLink(lnk);
 		return;
@@ -290,5 +433,6 @@
 /// @see CommunicationListener.h or @see OverlayInterface.h
 void Chord::onLinkDown(const LinkID& lnk, const NodeID& remote) {
-	logging_debug("link_down: link=" << lnk.toString() << " remote=" <<
+    // XXX logging_debug
+	logging_info("link_down (Chord): link=" << lnk.toString() << " remote=" <<
 			remote.toString() );
 
@@ -303,23 +447,24 @@
 /// @see CommunicationListener.h
 /// @see OverlayInterface.h
-void Chord::onMessage(const DataMessage& msg, const NodeID& remote,
+void Chord::onMessage(OverlayMsg* msg,
+        reboost::shared_buffer_t sub_msg,
+        const NodeID& remote,
 		const LinkID& link) {
 
-	// decode message
-	OverlayMsg* m = dynamic_cast<OverlayMsg*>(msg.getMessage());
-	if (m == NULL) return;
-
 	// handle messages
-	switch ((signalMessageTypes)m->getType()) {
+	switch ((signalMessageTypes) msg->getType()) {
 
 	// discovery request
-	case typeDiscovery: {
-		// decapsulate message
-		Discovery* dmsg = m->decapsulate<Discovery> ();
+	case typeDiscovery:
+	{
+		// deserialize discovery message
+		DiscoveryMessage dmsg;
+		dmsg.deserialize(sub_msg);
+		
 		logging_debug("Received discovery message with"
-			    << " src=" << m->getSourceNode().toString()
-				<< " dst=" << m->getDestinationNode().toString()
-				<< " ttl=" << (int)dmsg->getTTL()
-				<< " type=" << (int)dmsg->getType()
+			    << " src=" << msg->getSourceNode().toString()
+				<< " dst=" << msg->getDestinationNode().toString()
+				<< " ttl=" << (int)dmsg.ttl
+				<< " type=" << (int)dmsg.type
 		);
 
@@ -327,108 +472,125 @@
 		bool found = false;
 		BOOST_FOREACH( NodeID& value, discovery )
-			if (value == m->getSourceNode()) {
+			if (value == msg->getSourceNode()) {
 				found = true;
 				break;
 			}
-		if (!found) discovery.push_back(m->getSourceNode());
+		if (!found) discovery.push_back(msg->getSourceNode());
 
 		// check if source node can be added to routing table and setup link
-		if (m->getSourceNode() != nodeid)
-			setup( dmsg->getEndpoint(), m->getSourceNode() );
+		if (msg->getSourceNode() != nodeid)
+			setup( dmsg.endpoint, msg->getSourceNode() );
 
 		// process discovery message -------------------------- switch start --
-		switch (dmsg->getType()) {
-
-		// normal: route discovery message like every other message
-		case Discovery::normal: {
-			// closest node? yes-> split to follow successor and predecessor
-			if ( table->is_closest_to(m->getDestinationNode()) ) {
-				logging_debug("Discovery split:");
-				if (!table->get_successor()->isUnspecified()) {
-					OverlayMsg omsg(*m);
-					dmsg->setType(Discovery::successor);
-					omsg.encapsulate(dmsg);
-					logging_debug("* Routing to successor "
-							<< table->get_successor()->toString() );
-					baseoverlay.send( &omsg, *table->get_successor() );
-				}
-
-				// send predecessor message
-				if (!table->get_predesessor()->isUnspecified()) {
-					OverlayMsg omsg(*m);
-					dmsg->setType(Discovery::predecessor);
-					omsg.encapsulate(dmsg);
-					logging_debug("* Routing to predecessor "
-							<< table->get_predesessor()->toString() );
-					baseoverlay.send( &omsg, *table->get_predesessor() );
-				}
-			}
-			// no-> route message
-			else {
-				baseoverlay.route( m );
-			}
-			break;
-		}
-
-		// successor mode: follow the successor until TTL is zero
-		case Discovery::successor:
-		case Discovery::predecessor: {
-			// reached destination? no->forward!
-			if (m->getDestinationNode() != nodeid) {
-				OverlayMsg omsg(*m);
-				omsg.encapsulate(dmsg);
-				omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
-				baseoverlay.route( &omsg );
-				break;
-			}
-
-			// time to live ended? yes-> stop routing
-			if (dmsg->getTTL() == 0 || dmsg->getTTL() > 10) break;
-
-			// decrease time-to-live
-			dmsg->setTTL(dmsg->getTTL() - 1);
-
-			const route_item* item = NULL;
-			if (dmsg->getType() == Discovery::successor &&
-					table->get_successor() != NULL) {
-				item = table->get(*table->get_successor());
-			} else {
-				if (table->get_predesessor()!=NULL)
-					item = table->get(*table->get_predesessor());
-			}
-			if (item == NULL)
-				break;
-
-			logging_debug("Routing discovery message to succ/pred "
-				<< item->id.toString() );
-			OverlayMsg omsg(*m);
-			omsg.encapsulate(dmsg);
-			omsg.setDestinationNode(item->id);
-			omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
-			baseoverlay.send(&omsg, omsg.getDestinationNode());
-			break;
-		}
-		case Discovery::invalid:
-			break;
-
-		default:
-			break;
-		}
-		// process discovery message ---------------------------- switch end --
-
-		delete dmsg;
-		break;
-	}
-
-	// leave
-	case typeLeave: {
-		if (link!=LinkID::UNSPECIFIED) {
-			route_item* item = table->get(remote);
-			if (item!=NULL) item->info = LinkID::UNSPECIFIED;
-			table->remove(remote);
-			baseoverlay.dropLink(link);
-		}
-		break;
-	}}
+		switch ( dmsg.type )
+		{
+            // normal: route discovery message like every other message
+            case DiscoveryMessage::normal:
+            {
+                // closest node? yes-> split to follow successor and predecessor
+                if ( table->is_closest_to(msg->getDestinationNode()) )
+                {
+                    logging_debug("Discovery split:");
+                    if (!table->get_successor()->isUnspecified())
+                    {
+                        OverlayMsg omsg(*msg);
+                        
+                        dmsg.type = DiscoveryMessage::successor;
+                        omsg.set_payload_message(dmsg.serialize());
+
+                        logging_debug("* Routing to successor "
+                                << table->get_successor()->toString() );
+                        send_node( &omsg, *table->get_successor() );
+                    }
+    
+                    // send predecessor message
+                    if (!table->get_predesessor()->isUnspecified())
+                    {
+                        OverlayMsg omsg(*msg);
+                        
+                        dmsg.type = DiscoveryMessage::predecessor;
+                        omsg.set_payload_message(dmsg.serialize());
+
+                        logging_debug("* Routing to predecessor "
+                                << table->get_predesessor()->toString() );
+                        send_node( &omsg, *table->get_predesessor() );
+                    }
+                }
+                // no-> route message
+                else
+                {
+                    baseoverlay.route( msg );
+                }
+                break;
+            }
+    
+            // successor mode: follow the successor until TTL is zero
+            case DiscoveryMessage::successor:
+            case DiscoveryMessage::predecessor:
+            {
+                // reached destination? no->forward!
+                if (msg->getDestinationNode() != nodeid)
+                {
+                    OverlayMsg omsg(*msg);
+                    omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
+                    
+                    omsg.set_payload_message(dmsg.serialize());
+                    
+                    baseoverlay.route( &omsg );
+                    break;
+                }
+    
+                // time to live ended? yes-> stop routing
+                if (dmsg.ttl == 0 || dmsg.ttl > 10) break;
+    
+                // decrease time-to-live
+                dmsg.ttl--;
+    
+                const route_item* item = NULL;
+                if (dmsg.type == DiscoveryMessage::successor &&
+                        table->get_successor() != NULL)
+                {
+                    item = table->get(*table->get_successor());
+                }
+                else if (table->get_predesessor() != NULL)
+                {
+                        item = table->get(*table->get_predesessor());
+                }
+                if (item == NULL)
+                    break;
+    
+                logging_debug("Routing discovery message to succ/pred "
+                    << item->id.toString() );
+                OverlayMsg omsg(*msg);
+                omsg.setService(OverlayInterface::OVERLAY_SERVICE_ID);
+                omsg.setDestinationNode(item->id);
+                
+                omsg.set_payload_message(dmsg.serialize());
+                
+                send_node( &omsg, omsg.getDestinationNode() );
+                break;
+            }
+            case DiscoveryMessage::invalid:
+                break;
+    
+            default:
+                break;
+            }
+            // process discovery message ---------------------------- switch end --
+    
+            break;
+        }
+    
+        // leave
+        case typeLeave: {
+            if (link!=LinkID::UNSPECIFIED) {
+                route_item* item = table->get(remote);
+                if (item!=NULL) item->info = LinkID::UNSPECIFIED;
+                table->remove(remote);
+                baseoverlay.dropLink(link);
+            }
+            break;
+        }
+	}
 }
 
