Index: source/ariba/overlay/BaseOverlay.cpp
===================================================================
--- source/ariba/overlay/BaseOverlay.cpp	(revision 6202)
+++ source/ariba/overlay/BaseOverlay.cpp	(revision 6266)
@@ -51,4 +51,5 @@
 
 #include "ariba/overlay/messages/OverlayMsg.h"
+#include "ariba/overlay/messages/DHTMessage.h"
 #include "ariba/overlay/messages/JoinRequest.h"
 #include "ariba/overlay/messages/JoinReply.h"
@@ -58,4 +59,184 @@
 namespace ariba {
 namespace overlay {
+
+class ValueEntry {
+public:
+	ValueEntry( const Data& value ) : ttl(0), last_update(time(NULL)),
+		last_change(time(NULL)), value(value.clone()) {
+	}
+
+	~ValueEntry()  {
+		value.release();
+	}
+
+	void refresh() {
+		last_update = time(NULL);
+	}
+
+	void set_value( const Data& value ) {
+		this->value.release();
+		this->value = value.clone();
+		this->last_change = time(NULL);
+		this->last_update = time(NULL);
+	}
+
+	Data get_value() const {
+		return value;
+	}
+
+	uint16_t get_ttl() const {
+		return ttl;
+	}
+
+	void set_ttl( uint16_t ttl ) {
+		this->ttl = ttl;
+	}
+
+	bool is_ttl_elapsed() const {
+		// is persistent? yes-> always return false
+		if (ttl==0) return false;
+
+		// return true, if ttl is elapsed
+		return ( difftime( time(NULL), this->last_update ) >= ttl );
+	}
+
+private:
+	uint16_t ttl;
+	time_t last_update;
+	time_t last_change;
+	Data value;
+};
+
+class DHTEntry {
+public:
+	Data key;
+	vector<ValueEntry> values;
+
+	vector<Data> get_values() {
+		vector<Data> vect;
+		BOOST_FOREACH( ValueEntry& e, values )
+			vect.push_back( e.get_value() );
+		return vect;
+	}
+};
+
+class DHT {
+public:
+	typedef vector<DHTEntry> Entries;
+	typedef vector<ValueEntry> Values;
+	Entries entries;
+
+	static bool equals( const Data& lhs, const Data& rhs ) {
+		if (rhs.getLength()!=lhs.getLength()) return false;
+		for (int i=0; i<lhs.getLength()/8; i++)
+			if (lhs.getBuffer()[i] != rhs.getBuffer()[i]) return false;
+		return true;
+	}
+
+	void put( const Data& key, const Data& value, uint16_t ttl = 0 ) {
+
+		// find entry
+		for (size_t i=0; i<entries.size(); i++) {
+			DHTEntry& entry = entries.at(i);
+
+			// check if key is already known
+			if ( equals(entry.key, key) ) {
+
+				// check if value is already in values list
+				for (size_t j=0; j<entry.values.size(); j++) {
+					// found value already? yes-> refresh ttl
+					if ( equals(entry.values[j].get_value(), value) ) {
+						entry.values[j].refresh();
+						std::cout << "DHT: Republished value. Refreshing TTL."
+							<< std::endl;
+						return;
+					}
+				}
+
+				// new value-> add to entry
+				std::cout << "DHT: Added value to "
+					<< " key=" << key << " with value=" << value << std::endl;
+				entry.values.push_back( ValueEntry( value ) );
+				entry.values.back().set_ttl(ttl);
+				return;
+			}
+		}
+
+		// key is unknown-> add key value pair
+		std::cout << "DHT: New key value pair "
+			<< " key=" << key << " with value=" << value << std::endl;
+
+		// add new entry
+		entries.push_back( DHTEntry() );
+		DHTEntry& entry = entries.back();
+		entry.key = key.clone();
+		entry.values.push_back( ValueEntry(value) );
+		entry.values.back().set_ttl(ttl);
+	}
+
+	vector<Data> get( const Data& key ) {
+		// find entry
+		for (size_t i=0; i<entries.size(); i++) {
+			DHTEntry& entry = entries.at(i);
+			if ( equals(entry.key,key) )
+				return entry.get_values();
+		}
+		return vector<Data>();
+	}
+
+	bool remove( const Data& key ) {
+		// find entry
+		for (Entries::iterator i = entries.begin(); i != entries.end(); i++) {
+			DHTEntry& entry = *i;
+
+			// found? yes-> delete entry
+			if ( equals(entry.key, key) ) {
+				i = entries.erase(i);
+				return true;
+			}
+		}
+		return false;
+	}
+
+	bool remove( const Data& key, const Data& value ) {
+		// find entry
+		for (Entries::iterator i = entries.begin(); i != entries.end(); i++) {
+			DHTEntry& entry = *i;
+
+			// found? yes-> try to find value
+			if ( equals(entry.key, key) ) {
+				for (Values::iterator j = entry.values.begin();
+						j != entry.values.end(); j++) {
+
+					// value found? yes-> delete
+					if (equals(j->get_value(), value)) {
+						j = entry.values.erase(j);
+						return true;
+					}
+				}
+			}
+		}
+		return false;
+	}
+
+	void cleanup() {
+		// find entry
+		for (Entries::iterator i = entries.begin(); i != entries.end(); i++) {
+			DHTEntry& entry = *i;
+
+			for (Values::iterator j = entry.values.begin();
+					j != entry.values.end(); j++) {
+
+				// value found? yes-> delete
+				if (j->is_ttl_elapsed())
+					j = entry.values.erase(j);
+			}
+
+			if (entry.values.size()==0) i = entries.erase(i);
+		}
+	}
+};
+
+// ----------------------------------------------------------------------------
 
 /* *****************************************************************************
@@ -545,7 +726,9 @@
 	spovnetId(SpoVNetID::UNSPECIFIED), state(BaseOverlayStateInvalid),
 	sideport(&SideportListener::DEFAULT), started(false), counter(0) {
+	dht = new DHT();
 }
 
 BaseOverlay::~BaseOverlay() {
+	delete dht;
 }
 
@@ -1582,4 +1765,8 @@
 	overlayMsg->addRouteRecord(nodeId);
 
+	// handle dht messages (do not route)
+	if (overlayMsg->isDHTMessage())
+		return handleDHTMessage(overlayMsg);
+
 	// handle signaling messages (do not route!)
 	if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart &&
@@ -1601,4 +1788,12 @@
 		return true;
 	}
+
+	// handle DHT response messages
+	if (overlayMsg->hasTypeMask( OverlayMsg::maskDHTResponse )) {
+		bool ret = handleDHTMessage(overlayMsg);
+		delete overlayMsg;
+		return ret;
+	}
+
 
 	// handle base overlay message
@@ -1699,3 +1894,124 @@
 }
 
+
+// ----------------------------------------------------------------------------
+
+
+
+/// stabilize DHT state
+void BaseOverlay::stabilizeDHT() {
+
+}
+
+// handle DHT messages
+bool BaseOverlay::handleDHTMessage( OverlayMsg* msg ) {
+
+	// decapsulate message
+	logging_debug("received DHT message");
+	DHTMessage* dhtMsg = msg->decapsulate<DHTMessage>();
+
+	// handle DHT data message
+	if (msg->getType()==OverlayMsg::typeDHTData) {
+		const ServiceID& service = msg->getService();
+		logging_debug( "Received DHT data for service " << service.toString() );
+
+		// delegate data message
+		getListener(service)->onKeyValue(dhtMsg->getKey(), dhtMsg->getValues() );
+		return true;
+	}
+
+	// route message to closest node
+	if (!overlayInterface->isClosestNodeTo(msg->getDestinationNode())) {
+		logging_debug("Routing DHT message to closest node "
+			<< " from " << msg->getSourceNode()
+			<< " to " << msg->getDestinationNode()
+		);
+		route( msg );
+		delete msg;
+		return true;
+	}
+
+	// now, we are the closest node...
+	switch (msg->getType()) {
+	case OverlayMsg::typeDHTPut: {
+		BOOST_FOREACH( Data value, dhtMsg->getValues() )
+			dht->put(dhtMsg->getKey(), value, dhtMsg->getTTL() );
+		break;
+	}
+
+	case OverlayMsg::typeDHTGet: {
+		vector<Data> vect = dht->get(dhtMsg->getKey());
+		OverlayMsg omsg(*msg);
+		omsg.swapRoles();
+		omsg.setType(OverlayMsg::typeDHTData);
+		DHTMessage dhtmsg(dhtMsg->getKey(), vect);
+		omsg.encapsulate(&dhtmsg);
+		this->send(&omsg, omsg.getDestinationNode());
+		break;
+	}
+
+	case OverlayMsg::typeDHTRemove: {
+		if (dhtMsg->hasValues()) {
+			BOOST_FOREACH( Data value, dhtMsg->getValues() )
+					dht->remove(dhtMsg->getKey(), value );
+		} else
+			dht->remove( dhtMsg->getKey() );
+		break;
+	}
+
+	default:
+		logging_error("DHT Message type unknown.");
+		return false;
+	}
+	delete msg;
+	return true;
+}
+
+/// put a value to the DHT with a ttl given in seconds
+void BaseOverlay::dhtPut( const Data& key, const Data& value, int ttl ) {
+	// calculate hash
+	NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+	OverlayMsg msg(OverlayMsg::typeDHTPut);
+	DHTMessage dhtmsg(key,value);
+	dhtmsg.setTTL(ttl);
+	msg.setDestinationNode(dest);
+	msg.encapsulate( &dhtmsg );
+	send(&msg, dest);
+}
+
+/// removes a key value pair from the DHT
+void BaseOverlay::dhtRemove( const Data& key, const Data& value ) {
+	// calculate hash
+	NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+	OverlayMsg msg(OverlayMsg::typeDHTRemove);
+	DHTMessage dhtmsg(key,value);
+	msg.setDestinationNode(dest);
+	msg.encapsulate( &dhtmsg );
+	send(&msg, dest);
+}
+
+/// removes all data stored at the given key
+void BaseOverlay::dhtRemove( const Data& key ) {
+	// calculate hash
+	NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+	OverlayMsg msg(OverlayMsg::typeDHTRemove);
+	DHTMessage dhtmsg(key);
+	msg.setDestinationNode(dest);
+	msg.encapsulate( &dhtmsg );
+	send(&msg, dest);
+}
+
+/// requests data stored using key
+void BaseOverlay::dhtGet( const Data& key, const ServiceID& service ) {
+	// calculate hash
+	NodeID dest = NodeID::sha1(key.getBuffer(), key.getLength() / 8);
+	OverlayMsg msg(OverlayMsg::typeDHTRemove);
+	DHTMessage dhtmsg(key);
+	msg.setDestinationNode(dest);
+	msg.setService(service);
+	msg.encapsulate( &dhtmsg );
+	send(&msg, dest);
+}
+
+
 }} // namespace ariba, overlay
