Index: source/services/CMakeLists.txt =================================================================== --- source/services/CMakeLists.txt (revision 10700) +++ source/services/CMakeLists.txt (revision 10700) @@ -0,0 +1,39 @@ +# [License] +# The Ariba-Underlay Copyright +# +# Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +# +# Institute of Telematics +# UniversitÀt Karlsruhe (TH) +# Zirkel 2, 76128 Karlsruhe +# Germany +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OF TELEMATICS OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +# +# The views and conclusions contained in the software and documentation +# are those of the authors and should not be interpreted as representing +# official policies, either expressed or implied, of the Institute of +# Telematics. +# [License] + +add_subdirectory(ariba_dht) Index: source/services/Makefile.am =================================================================== --- source/services/Makefile.am (revision 10688) +++ (revision ) @@ -1,1 +1,0 @@ -SUBDIRS = dht Index: source/services/ariba_dht/CMakeLists.txt =================================================================== --- source/services/ariba_dht/CMakeLists.txt (revision 10700) +++ source/services/ariba_dht/CMakeLists.txt (revision 10700) @@ -0,0 +1,97 @@ +# [License] +# The Ariba-Underlay Copyright +# +# Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +# +# Institute of Telematics +# UniversitÀt Karlsruhe (TH) +# Zirkel 2, 76128 Karlsruhe +# Germany +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OF TELEMATICS OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +# +# The views and conclusions contained in the software and documentation +# are those of the authors and should not be interpreted as representing +# official policies, either expressed or implied, of the Institute of +# Telematics. +# [License] + +###################################################### +### Increment this whenever the interface changes! ### +###################################################### +set(ariba_dht_SOVERSION 1) +###################################################### + +# Find Boost +find_package(BoostAdditionalVersions QUIET) +if(NOT DEFINED BoostAdditionalVersions_FOUND) + message(WARNING "Could not find FindBoostAdditionalVersions.cmake. " + "This might cause the Boost detection to fail") +endif() +find_package(Boost 1.42.0 REQUIRED COMPONENTS system) +mark_as_advanced(Boost_DIR) + +include_directories(${Boost_INCLUDE_DIRS}) + + +# Include ariba header files +include_directories("${ariba_SOURCE_DIR}/source/" "${ariba_BINARY_DIR}/source/") + + +include(BuildSharedAndStaticLib) + +set(ariba_dht_HEADERS + Dht.h + DhtAnswerInterface.h + ) + +set(ariba_dht_SOURCES + Dht.cpp + messages/DhtMessage.cpp + messages/DhtMessage.h + ${ariba_dht_HEADERS} + ) + +set(ariba_dht_LINK_LIBRARIES ariba ${Boost_LIBRARIES}) +set(ariba_dht_VERSION ${ariba_VERSION}) +# ariba_dht_SOVERSION already defined above + +build_shared_and_static_libs(ariba_dht) + + +# Installation stuff +install(TARGETS ariba_dht ${ariba_dht_STATIC_TARGET} EXPORT ariba-targets + LIBRARY DESTINATION lib COMPONENT Runtime + ARCHIVE DESTINATION lib COMPONENT Development + RUNTIME DESTINATION bin COMPONENT Runtime + ) + +install(FILES ${ariba_dht_HEADERS} + DESTINATION include/ariba_dht + COMPONENT Development + ) + +# Make libariba_dht usable from build tree. +export(TARGETS ariba_dht ${ariba_dht_STATIC_TARGET} + APPEND FILE "${ariba_BINARY_DIR}/ariba-exports.cmake" + ) Index: source/services/ariba_dht/Dht.cpp =================================================================== --- source/services/ariba_dht/Dht.cpp (revision 10700) +++ source/services/ariba_dht/Dht.cpp (revision 10700) @@ -0,0 +1,553 @@ +/* + * Dht.cpp + * + * Created on: 20.06.2012 + * Author: mario + */ + +#include "Dht.h" +#include "messages/DhtMessage.h" +#include + +namespace ariba_service { +namespace dht { + +use_logging_cpp(Dht) + +using namespace std; +using boost::date_time::second_clock; +using boost::posix_time::ptime; + +SystemEventType DhtRepublishEvent("DhtRepublishEvent"); +SystemEventType DhtCleanupEvent("DhtCleanupEvent"); + + +Dht::Dht(ariba::ServiceID serviceID, ariba::Node* node) : + serviceID(serviceID), + node(node), + cleanup_running(false), + listener(NULL) +{ + this->node->bind(this, serviceID); +} + +Dht::~Dht() +{ + this->node->unbind(this, serviceID); +} + + + +void Dht::put(const std::string& key, const std::string& value, uint16_t ttl) +{ + DhtMessage msg(DhtMessage::DhtPut, key, value, ttl); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::get(const std::string& key) +{ + DhtMessage msg(DhtMessage::DhtGet, key); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::atomic_put_and_get(const std::string& key, const std::string& value, uint16_t ttl) +{ + DhtMessage msg(DhtMessage::DhtPutAndGet, key, value, ttl); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::meet(const std::string& key, const std::string& value, uint16_t ttl_in_sec) +{ + // insert into meet_store + insert_into_table(meet_store, + key, + std::vector(1, value), + ttl_in_sec); + + // send message (and program republishing) + send_meet_message(key, value); +} + +void Dht::stop_meet(const std::string& key, const std::string& value) +{ + remove_from_table(meet_store, + key, + std::vector(1, value)); +} + +void Dht::remove(const std::string& key, const std::string& value) +{ + // send delete message + DhtMessage msg(DhtMessage::DhtRemove, key, value); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + + + + +bool Dht::add_listener(DhtAnswerInterface* new_listener) +{ + if ( listener == NULL ) + { + listener = new_listener; + + return true; + } + else + { + return false; + } +} + +bool Dht::remove_listener(DhtAnswerInterface* new_listener) +{ + if (listener == new_listener) { + listener = NULL; + return true; + + } else { + return false; + } +} + + + + +//** PRIVATE FUNCTIONS **// + +void Dht::handle_dht_message(const DhtMessage& message, const NodeID& source) +{ + // send message closer to hashed key + NodeID addr = message.getHashedKey(); + + logging_debug("Processing DHT message..."); + + logging_debug("Dest Addr: " << addr.toString()); + + // * send closer, if possible * + const ariba::NodeID dest = node->sendMessageCloserToNodeID(message, addr, this->serviceID); + + logging_debug("Closer Node: " << dest.toString()); + + // couldn't send closer, so we are the closest node + // ---> * handle dht request * (store value, etc.) + if ( dest == NodeID::UNSPECIFIED ) + { + logging_debug("DHT: We are the closest node!"); + + switch (message.getType()) + { + case DhtMessage::DhtPut: + { + insert_into_table( + table, + message.getKey(), + message.getValues(), + message.getTTL()); + + break; + } + + case DhtMessage::DhtGet: + { + answer_dht_request(message.getKey(), source); + + break; + } + + case DhtMessage::DhtPutAndGet: + { + insert_into_table( + table, + message.getKey(), + message.getValues(), + message.getTTL()); + answer_dht_request(message.getKey(), source); + + break; + } + + case DhtMessage::DhtRemove: + { + remove_from_table(table, message.getKey(), message.getValues()); + + break; + } + } + } +} + + +void Dht::insert_into_table(DhtTableType& table, + const std::string& key, + const vector& values, + uint16_t ttl) +{ + DhtTableType::mapped_type& value_entries = table[key]; + + BOOST_FOREACH(const std::string& value, values) { + + // Debug output + logging_info("DHT: Inserting (" << key << ", " << value << ")"); + + // push the value for the given key (into the vector) + bool entry_updated = false; + for ( + DhtTableType::mapped_type::iterator position = value_entries.begin(); + position != value_entries.end(); + ++position) + { + if (position->get_value() == value) { + position->set_ttl(ttl); + entry_updated = true; + break; + } + } + + if (!entry_updated) { + value_entries.push_back(ValueEntry(value, ttl)); + } + } + + schedule_cleanup_event(); +} + + +void Dht::remove_from_table(DhtTableType& table, + const std::string& key, + const vector& values) +{ + logging_debug("DHT: trying to delete some values for key " << key); + // find key + DhtTableType::iterator key_position = table.find(key); + if (key_position == table.end()) { + return; + } + + // delete values from set of values + DhtTableType::mapped_type& entries = key_position->second; + BOOST_FOREACH(const std::string& value, values) { + for ( + DhtTableType::mapped_type::iterator entry = entries.begin(); + entry != entries.end(); + ++entry) + { + if (entry->get_value() == value) { + logging_info("DHT: Deleting " + "(" <get_value() << ")"); + entries.erase(entry); + break; + } + } + } + + // the key could empty now + // ---> remove it + if ( entries.size() == 0 ) + { + table.erase(key_position); + } +} + + +void Dht::cleanup_table(DhtTableType& table) +{ + logging_debug("DHT: cleaning up table"); + + vector to_be_deleted; + + for ( + DhtTableType::iterator position = table.begin(); + position != table.end(); + ++position) + { + cleanup_entries(position->second); + + // mark entry container for removal if empty + if (position->second.size() == 0) { + to_be_deleted.push_back(position->first); + } + } + + BOOST_FOREACH(const std::string& key, to_be_deleted) { + table.erase(key); + } +} + +void Dht::cleanup_entries(DhtTableType::mapped_type& entries) +{ + DhtTableType::mapped_type::iterator position = entries.begin(); + while (position != entries.end()) { + + if (position->is_ttl_elapsed()) { + // remove stale entry + position = entries.erase(position); + + } else { + // move on otherwise + ++position; + } + } +} + + +void Dht::answer_dht_request(const std::string& key, const NodeID& source) +{ + // get entries from table + const DhtTableType::mapped_type& entries = table[key]; + + // need to convert value entries to strings + vector values; + values.reserve(entries.size()); + BOOST_FOREACH(const ValueEntry& entry, entries) { + + if (!entry.is_ttl_elapsed()) { + values.push_back(entry.get_value()); + } + + } + + // BRANCH: request comes from another node + // ---> send answer message + if ( source != NodeID::UNSPECIFIED ) + { + // create answer message + DhtMessage msg(DhtMessage::DhtAnswer, key, values); + + // * send answer * + node->sendMessage(msg, source, serviceID); + } + + // BRANCH: local request + // ---> inform listeners directly (TODO code duplicates...) + else + { + logging_debug("DHT: Answering request for key '" << key << "' locally"); + + // * inform listeners * + if ( listener ) + { + listener->handle_dht_answer(key, values); + } + } + + + // an empty key could have been created + // ---> remove it + if ( entries.size() == 0 ) + { + table.erase(key); + } +} + + +void Dht::send_meet_message(const std::string& key, const std::string& value) +{ + // send put&get message + DhtMessage msg(DhtMessage::DhtPutAndGet, key, value, MEET_DHT_TTL); + + handle_dht_message(msg, NodeID::UNSPECIFIED); + + // program timer for republish (or deletion) + Key_Value* kv = new Key_Value; + kv->key = key; + kv->value = value; + + SystemQueue::instance().scheduleEvent( + SystemEvent( this, DhtRepublishEvent, kv), + MEET_REPUBLISH_INTERVAL * 1000 ); +} + + +void Dht::meet_update_event(const std::string& key, const std::string& value) +{ + // get entries from table + DhtTableType::mapped_type& entries = meet_store[key]; + + cleanup_entries(entries); + + // find right entry + BOOST_FOREACH(const ValueEntry& entry, entries) { + if (entry.get_value() == value) { + + // republish value + logging_debug("DHT: Republishing " + "(" << key << ", " << entry.get_value() << ")"); + send_meet_message(key, entry.get_value()); + } + } + + // an empty key could have been created + // ---> remove it + if ( entries.size() == 0 ) + { + meet_store.erase(key); + } +} + +void Dht::schedule_cleanup_event(bool reschedule) +{ + if (reschedule || !cleanup_running) { + SystemQueue::instance().scheduleEvent( + SystemEvent(this, DhtCleanupEvent), + CLEANUP_INTERVAL * 1000); + cleanup_running = true; + } +} + + +void Dht::print_dht() +{ + logging_debug("======== DHT ========"); + for ( DhtTableType::iterator dht_it = table.begin(); dht_it != table.end(); dht_it++) + { + logging_debug("Key: " << dht_it->first); + + for ( DhtTableType::mapped_type::iterator value_it = dht_it->second.begin(); + value_it != dht_it->second.end(); + value_it++ ) + { + logging_debug("--> " << value_it->get_value()); + } + + logging_debug("- - - - -"); + } + + logging_debug("======== [DHT] ========"); +} + + + +void Dht::onMessage(const ariba::DataMessage& msg, const ariba::NodeID& source, + const ariba::LinkID& lnk) +{ + logging_debug("DHT: Incoming message..."); + + DhtMessage* mess = msg.getMessage()->convert (); + + // handle message + switch (mess->getType()) + { + // BRANCH: Message is an Answer for our request + case DhtMessage::DhtAnswer: + { + logging_debug("DHT: Got answer for key '" << mess->getKey() << "'"); + + BOOST_FOREACH(string str, mess->getValues()) + { + logging_debug("--> Value: '" << str << "'"); + } + + // * inform listeners * + if ( listener ) + { + listener->handle_dht_answer(mess->getKey(), mess->getValues()); + } + + break; + } + + // BRANCH: Message is a Request + // ---> route or handle + default: + { + handle_dht_message(*mess, source); + + break; + } + } + + delete mess; +} + + +void Dht::handleSystemEvent( const SystemEvent& event ) +{ + + if (event.getType() == DhtRepublishEvent) { + logging_debug("DHT: Meet republish event!"); + + // republish meet entry + Key_Value* kv = event.getData(); + meet_update_event(kv->key, kv->value); + delete kv; + + } else if (event.getType() == DhtCleanupEvent) { + logging_debug("DHT: Cleanup event!"); + + cleanup_table(table); + schedule_cleanup_event(true); + } +} + + +/************** + * ValueEntry * + **************/ + +Dht::ValueEntry::ValueEntry( + const std::string& value, + uint16_t ttl) : + ttl(ttl), + last_update(second_clock::universal_time()), + value(value) +{ +} + + +void Dht::ValueEntry::refresh() { + last_update = second_clock::universal_time(); +} + + +const std::string& Dht::ValueEntry::get_value() const { + return value; +} + +uint16_t Dht::ValueEntry::get_age() const +{ + boost::posix_time::time_duration diff = + second_clock::universal_time() - last_update; + + return diff.total_seconds(); +} + +uint16_t Dht::ValueEntry::get_ttl() const { + return ttl; +} + +void Dht::ValueEntry::set_ttl(uint16_t ttl) { + this->refresh(); + this->ttl = ttl; +} + +bool Dht::ValueEntry::is_ttl_elapsed() const { + // ttl == 0 signals infinite lifetime + if (ttl == 0) { + return false; + } + + return second_clock::universal_time() >= + (last_update + boost::posix_time::seconds(ttl)); +} + +uint16_t Dht::ValueEntry::get_remaining_ttl() const +{ + if ( ttl == 0 ) + return -1; + + if ( is_ttl_elapsed() ) + return 0; + + boost::posix_time::time_duration diff = + (last_update + boost::posix_time::seconds(ttl)) - + second_clock::universal_time(); + + return ttl - get_age(); +} + +}} /* namespace ariba_service::dht */ Index: source/services/ariba_dht/Dht.h =================================================================== --- source/services/ariba_dht/Dht.h (revision 10700) +++ source/services/ariba_dht/Dht.h (revision 10700) @@ -0,0 +1,213 @@ +/* + * Dht.h + * + * Created on: 20.06.2012 + * Author: mario + */ + +#ifndef DHT_H_ +#define DHT_H_ + +#include "ariba/ariba.h" +#include "ariba/utility/system/SystemQueue.h" +#include "ariba/utility/logging/Logging.h" +#include "DhtAnswerInterface.h" +#include +#include +#include + +namespace ariba_service { +namespace dht { + +using ariba::utility::SystemQueue; +using ariba::utility::SystemEvent; +using ariba::utility::SystemEventType; +using ariba::utility::SystemEventListener; + +// Forward declarations to avoid adding messages/*.h to the public interface +class DhtMessage; + +#define MEET_REPUBLISH_INTERVAL 10 +#define MEET_DHT_TTL 30 +#define CLEANUP_INTERVAL (5 * 60) + +class Dht : + public ariba::CommunicationListener, + public ariba::utility::SystemEventListener, + public boost::noncopyable +{ +use_logging_h(Dht) +public: + Dht(ariba::ServiceID serviceID, ariba::Node* node); + virtual ~Dht(); + + /** + * Put the value into the DHT under the specified key + * + * @param key + * Key to put the value under + * @param value + * The value which is put + * @param ttl + * The lifetime of the entry in seconds. The value will be removed + * automatically when it expires + */ + void put( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Get the values specified by the key + * + * @param key + * Key of the values which should be fetched + */ + void get(const std::string& key); + + /** + * Put and get in one single operation + * + * @param key + * The key the value will be put under and retrieved from + * @param value + * The value is first put then all values for that key, including the + * one just inserted will be sent back + * @param ttl + * The lifetime of the entry in seconds. The value will be removed + * automatically when it expires + */ + void atomic_put_and_get( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Periodically put and get the value + * + * @param key + * The key the value will be put under and retrieved from + * @param value + * The value that will be periodically put into the DHT. The value is + * first put then all values for that key, including the one just + * inserted will be sent back + * @param ttl + * How long should we try to put the value periodically (measured in + * seconds). 0 means putting the value until stop_meet() is called + */ + void meet( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Stop periodically pushing the value + */ + void stop_meet(const std::string& key, const std::string& value); + + /** + * Remove the value under the specified key + */ + void remove(const std::string& key, const std::string& value); + + /** + * Register a listener which is called when an answer is received + */ + bool add_listener(DhtAnswerInterface* new_listener); + + /** + * Unregister a listener + * + * @returns true if the handler was successfully unregistered, false if the + * listener was not registered + */ + bool remove_listener(DhtAnswerInterface* new_listener); + +protected: + /*** CommunicationListener interface ***/ + + /** + * Called when a message is incoming + * @param msg The data message that is received + * @param remote The remote node that sent the message + * @param lnk The link id of the link where the message is received + */ + virtual void onMessage(const ariba::DataMessage& msg, const ariba::NodeID& source, + const ariba::LinkID& lnk = ariba::LinkID::UNSPECIFIED); + + + /*** SystemEventListener interface ***/ + virtual void handleSystemEvent( const SystemEvent& event ); + + +private: + class ValueEntry { + public: + ValueEntry(const std::string& value, uint16_t ttl = 0); + + void refresh(); + + const std::string& get_value() const; + + uint16_t get_age() const; + + uint16_t get_ttl() const; + void set_ttl(uint16_t ttl); + bool is_ttl_elapsed() const; + uint16_t get_remaining_ttl() const; + + bool operator<(const ValueEntry& rhs) const; + + private: + uint16_t ttl; + boost::posix_time::ptime last_update; + std::string value; + }; + + struct Key_Value + { + string key; + string value; + }; + + +private: + void handle_dht_message(const DhtMessage& message, const NodeID& source); + + void answer_dht_request(const std::string& key, const NodeID& source); + void send_meet_message(const std::string& key, const std::string& value); + void meet_update_event(const std::string& key, const std::string& value); + + // just for debug purpose + void print_dht(); + + + ariba::ServiceID serviceID; + ariba::Node* node; + + typedef std::map< std::string, std::vector > DhtTableType; + DhtTableType table; + DhtTableType meet_store; + + void insert_into_table(DhtTableType& table, + const std::string& key, + const vector& values, + uint16_t ttl); + void remove_from_table(DhtTableType& table, + const std::string& key, + const vector& values); + void cleanup_table(DhtTableType& table); + void cleanup_entries(DhtTableType::mapped_type& entries); + + void schedule_cleanup_event(bool reschedule = false); + bool cleanup_running; + + // AnswerListener + DhtAnswerInterface* listener; +}; + + +}} /* namespace ariba_service::dht */ + + +#endif /* DHT_H_ */ Index: source/services/ariba_dht/DhtAnswerInterface.h =================================================================== --- source/services/ariba_dht/DhtAnswerInterface.h (revision 10700) +++ source/services/ariba_dht/DhtAnswerInterface.h (revision 10700) @@ -0,0 +1,17 @@ +#ifndef DHT_ANSWER_INTERFACE_H_ +#define DHT_ANSWER_INTERFACE_H_ + +namespace ariba_service { +namespace dht { + +class DhtAnswerInterface +{ +public: + virtual void handle_dht_answer(const std::string& key, const std::vector& values) = 0; + + virtual ~DhtAnswerInterface() {} +}; + +}} /* namespace ariba_service::dht */ + +#endif /* DHT_ANSWER_INTERFACE_H_ */ Index: source/services/ariba_dht/messages/DhtMessage.cpp =================================================================== --- source/services/ariba_dht/messages/DhtMessage.cpp (revision 10700) +++ source/services/ariba_dht/messages/DhtMessage.cpp (revision 10700) @@ -0,0 +1,48 @@ +#include "DhtMessage.h" + +#include + +namespace ariba_service { +namespace dht { + +vsznDefault(DhtMessage); + +DhtMessage::DhtMessage() : + ttl( 0 ), + replace( false ) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key ) : + type( static_cast(type) ), + ttl( 0 ), + replace( false ), + key( key ) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key, + const std::string& value, uint16_t ttl ) : + type( static_cast(type) ), + ttl( ttl ), + replace( false ), + key( key ), + values(1, value) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key, + const vector& values, uint16_t ttl ) : + type( static_cast(type) ), + ttl( ttl ), + replace( false ), + key( key ) +{ + // preallocate enough room so we don't need to copy a lot + this->values.reserve(values.size()); + BOOST_FOREACH(const std::string value, values ) + this->values.push_back( value ); +} + +DhtMessage::~DhtMessage() { + // empty +} + +}} Index: source/services/ariba_dht/messages/DhtMessage.h =================================================================== --- source/services/ariba_dht/messages/DhtMessage.h (revision 10700) +++ source/services/ariba_dht/messages/DhtMessage.h (revision 10700) @@ -0,0 +1,121 @@ +#ifndef DHTMESSAGE_H_ +#define DHTMESSAGE_H_ + +#include "ariba/utility/messages.h" +#include "ariba/utility/serialization.h" +#include "ariba/Name.h" + +namespace ariba_service { +namespace dht { + +using ariba::utility::Message; +using_serialization; + +class DhtMessage : public Message { VSERIALIZEABLE +public: + typedef enum { + DhtInvalid = 0, + DhtGet = 1, + DhtPut = 2, + DhtPutAndGet = 3, + DhtRemove = 4, + DhtRepublish = 5, + DhtAnswer = 8 + } DhtMessageType; + + DhtMessage(); + DhtMessage( DhtMessageType type, const std::string& key ); + DhtMessage( DhtMessageType type, const std::string& key, + const std::string& value, uint16_t ttl = 0 ); + + DhtMessage( DhtMessageType type, const std::string& key, + const vector& values, uint16_t ttl = 0 ); + + virtual ~DhtMessage(); + + DhtMessageType getType() const { + return static_cast(type); + } + + NodeID getHashedKey() const { + return ariba::Name(key).toNodeId(); + } + + const std::string& getKey() const { + return key; + } + + /// returns the first element of the key vector + const std::string& getValue() const { + return values.at(0); + } + + /// return all values for the key + const vector& getValues() const { + return values; + } + + /// return all values for the key + vector& getValues() { + return values; + } + + bool hasValues() const { + return values.size() != 0; + } + + uint16_t getTTL() const { + return ttl; + } + + void setTTL( uint16_t ttl ) { + this->ttl = ttl; + } + + void setReplace( bool replace ) { + this->replace = replace; + } + + bool doReplace() const { + return replace; + } + + +private: + uint8_t type; + uint16_t ttl; + bool replace; + std::string key; + vector values; +}; + +}} // namespace ariba_service::dht + +sznBeginDefault( ariba_service::dht::DhtMessage, X ) { + X && type; + + // serialize tll + X && ttl; + + // key serialization + X && T(key); + + // store number of values + uint16_t num_values = values.size(); + X && num_values; + + // value serialization + for (size_t i=0; i