Changeset 12060 for source

source/ariba/CMakeLists.txt

-              r11885
+              r12060
 ### Increment this whenever the interface changes! ###
 ######################################################
 set(ariba_SOVERSION 1)
+set(ariba_SOVERSION 2)
 ######################################################
 …
 add_headers(
     ariba.h
     AribaModule.h
+#    AribaModule.h
     CommunicationListener.h
     DataMessage.h
 …
 add_sources(
     AribaModule.cpp
+#    AribaModule.cpp
     CommunicationListener.cpp
     DataMessage.cpp

source/ariba/CommunicationListener.cpp

-              r7468
+              r12060
 void CommunicationListener::onLinkFail(const LinkID& l, const NodeID& r) {
+    onLinkDown(l, r);
+}
 …
+}
+// this implementation provides backward compatibility
+// overwrite it to use the new interface
+void CommunicationListener::onMessage(reboost::shared_buffer_t message,
+        const NodeID& remote,
+        const LinkID& lnk,
+        const SequenceNumber& seqnum,
+        const ariba::overlay::OverlayMsg* overlay_msg)
+{
+    // copy data
+    Data data;
+    data.setLength(message.size() * 8);
+    memcpy(data.getBuffer(), message.data(), message.size());
+    // and prepare old-style overlay message
+    Message legacy_msg;
+    legacy_msg.setPayload(data);
+    // * call legacy handler *
+    this->onMessage(legacy_msg, remote, lnk);
+}
 void CommunicationListener::onKeyValue( const Data& key, const vector<Data>& value ) {
+}
+bool CommunicationListener::onPing(const NodeID& remote)
+{
+    return true;
+}
+void CommunicationListener::onPingLost(const NodeID& remote)
+{
+}
+void CommunicationListener::onPong(const NodeID& remote)
+{
+}
 } // namespace ariba

source/ariba/CommunicationListener.h

-              r9684
+              r12060
 #include "LinkProperties.h"
 #include "DataMessage.h"
+#include "ariba/overlay/SequenceNumber.h"
 namespace ariba {
+typedef ariba::overlay::SequenceNumber SequenceNumber;
 // forward decl
 namespace overlay {
         class BaseOverlay;
+        class OverlayMsg;
+}
 …
         /**
+         * @DEPRECATED
+         *
          * Called when a message is incoming
          * @param msg The data message that is received
 …
         virtual void onMessage(const DataMessage& msg, const NodeID& remote,
                         const LinkID& lnk = LinkID::UNSPECIFIED);
+    /**
+     * NOTE: This interface is still unstable and may change in future releases.
+     *   ---> Especially the parameter Â«overlay_msgÂ» may be replaced or removed.
+     *
+     * Called when a message is incoming
+     * @param msg A shared buffer with the (serialized) payload
+     * @param remote The remote node that sent the message
+     * @param lnk The link id of the link where the message is received
+     * @param overlay_msg A pointer to the ariba internal Overlay Message
+     */
+    virtual void onMessage(reboost::shared_buffer_t message,
+            const NodeID& remote,
+            const LinkID& lnk,
+            const SequenceNumber& seqnum,
+            const ariba::overlay::OverlayMsg* overlay_msg);
         // --- dht functionality ---
 …
         virtual void onKeyValue( const Data& key, const vector<Data>& value );
+    // --- ping pong (routed) ---
+    /**
+     * Called when a ping message was received.
+     *
+     * @param remote Node which sent the ping message
+     * @return bool Allows to send a pong message (default: true)
+     */
+    virtual bool onPing(const NodeID& remote);
+    /**
+     * Called when a ping message is lost.
+     *
+     * NOTE: A ping message could also get lost without notice.
+     *
+     * @param remote Node to which the ping message was sent
+     */
+    virtual void onPingLost(const NodeID& remote);
+    /**
+     * Called when a pong message was received.
+     *
+     * @param remote Node which sent the pong message
+     */
+    virtual void onPong(const NodeID& remote);
 };

source/ariba/DataMessage.h

-              r9684
+              r12060
 // use message utility
 #ifdef USE_MESSAGE_UTILITY
+  #include "ariba/utility/messages.h"
+//  #include "ariba/utility/messages.h"
+  #include "ariba/utility/messages/Message.h"
   namespace ariba {
     typedef utility::Message Message;

source/ariba/Message.h

-              r9684
+              r12060
 #include <inttypes.h>
+#include "ariba/utility/messages.h"
+// legacy messages
+#include "ariba/utility/messages/Message.h"
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
 /** \addtogroup public

source/ariba/Node.cpp

-              r10653
+              r12060
 #include "Node.h"
+#include <boost/foreach.hpp>
 #include "ariba/overlay/BaseOverlay.h"
+#include "ariba/communication/BaseCommunication.h"
 #include "ariba/utility/types/OverlayParameterSet.h"
 #include "ariba/communication/EndpointDescriptor.h"
+#include <boost/property_tree/exceptions.hpp>
+using namespace std;
 using ariba::communication::EndpointDescriptor;
+using boost::property_tree::ptree;
 namespace ariba {
+Node::Node(AribaModule& ariba_mod, const Name& node_name) :
+        name(node_name), ariba_mod(ariba_mod) {
+        base_overlay = new BaseOverlay();
+//Node::Node(AribaModule& ariba_mod, const Name& node_name) :
+//      name(node_name), ariba_mod(ariba_mod) {
+//      base_overlay = new BaseOverlay();
+//}
+Node::Node()  :
+        name(Name::UNSPECIFIED),
+        base_communication(NULL),
+        base_overlay(NULL)
+{
+    base_communication = new BaseCommunication();
+    base_overlay = new BaseOverlay();
+}
 …
         delete base_overlay;
         base_overlay = NULL;
+}
+void Node::join(const Name& vnetname) {
+        spovnetId = vnetname.toSpoVNetId();
+        nodeId = generateNodeId(name);
+        // start base comm if not started
+        if( !ariba_mod.base_comm->isStarted() )
+                ariba_mod.base_comm->start();
+        // start base overlay if not started
+        // join against ourselfs
+        if( !base_overlay->isStarted() )
+                base_overlay->start( *ariba_mod.base_comm, nodeId );
+        base_overlay->joinSpoVNet( spovnetId );
+        // join against static bootstrap points and
+        // start automatic bootstrapping modules
+        vector<AribaModule::BootstrapMechanism> mechanisms
+                = ariba_mod.getBootstrapMechanisms(vnetname);
+        vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
+        BOOST_FOREACH(AribaModule::BootstrapMechanism m, mechanisms){
+                switch(m){
+                        case AribaModule::BootstrapMechanismStatic:
+                        {
+                                const communication::EndpointDescriptor* ep =
+                                                        ariba_mod.getBootstrapNode(vnetname, m);
+                                        if( ep != NULL && ep->isUnspecified() == false )
+                                                base_overlay->joinSpoVNet( spovnetId, *ep);
+                                break;
+                        }
+                        case AribaModule::BootstrapMechanismBroadcast:
+                                internalmodules.push_back(make_pair(
+                                                BootstrapManager::BootstrapTypePeriodicBroadcast,
+                                                ariba_mod.getBootstrapInfo(vnetname, m)));
+                                break;
+                        case AribaModule::BootstrapMechanismMulticastDNS:
+                                internalmodules.push_back(make_pair(
+                                                BootstrapManager::BootstrapTypeMulticastDns,
+                                                ariba_mod.getBootstrapInfo(vnetname, m)));
+                                break;
+                        case AribaModule::BootstrapMechanismSDP:
+                                internalmodules.push_back(make_pair(
+                                                BootstrapManager::BootstrapTypeBluetoothSdp,
+                                                ariba_mod.getBootstrapInfo(vnetname, m)));
+                                break;
+                        default:
+                                break;
+                }
+        }
+        // start automatic overlay bootstrapping modules
+        base_overlay->startBootstrapModules(internalmodules);
+        // done
+}
+void Node::initiate(const Name& vnetname, const SpoVNetProperties& parm) {
+        utility::OverlayParameterSet ovrpset;
+        ovrpset.setOverlayStructure(
+                        (utility::OverlayParameterSet::_OverlayStructure)
+                        parm.getBaseOverlayType()
+                        );
+        spovnetId = vnetname.toSpoVNetId();
+        nodeId = generateNodeId(name);
+        // start base comm if not started
+        if( !ariba_mod.base_comm->isStarted() )
+                ariba_mod.base_comm->start();
+        // start base overlay if not started
+        if( !base_overlay->isStarted() )
+                base_overlay->start( *ariba_mod.base_comm, nodeId );
+        base_overlay->createSpoVNet( spovnetId, ovrpset );
+}
+        delete base_communication;
+        base_communication = NULL;
+}
+void Node::connect(const ptree& config)
+{
+    using namespace boost::property_tree;
+    using namespace addressing2;
+    assert( ! base_communication->isStarted() );
+    assert( ! base_overlay->isStarted() );
+    // XXX needed since Â»empty_ptree<ptree>()Â« is not working
+    //   ---> see: http://stackoverflow.com/questions/5003549/where-is-boost-property-treeempty-ptree
+    static const ptree empty_pt;
+    // SpovNet ID
+    Name spovnet_name(config.get<string>("spovnet_name"));
+    spovnetId = spovnet_name.toSpoVNetId();
+    // Node ID
+    try
+    {
+        name = config.get<string>("node_name");
+    }
+    catch ( ptree_bad_path& e )
+    {
+        name = Name::UNSPECIFIED;
+    }
+    nodeId = generateNodeId(name);
+    /* Base Communication */
+    EndpointSetPtr listen_on;
+    try
+    {
+        listen_on = endpoint_set::create_EndpointSet(
+            config.get_child("listen_on"));
+    }
+    catch ( ptree_bad_path& e )
+    {
+        /* no endpoints specified, using default: Â»[::]:41322+Â« */
+        ptree default_listen_on;
+        default_listen_on.put("endp.category", "TCPIP");
+        default_listen_on.put("endp.addr", "::");
+        default_listen_on.put("endp.port", 0);  // defaults to 41322 (or higher)
+        listen_on = endpoint_set::create_EndpointSet(default_listen_on);
+//        logging_warn("No endpoints specified in config. ---> Using default.");
+        cout << "No endpoints specified in config. ---> Using default." << endl;
+    }
+    base_communication->start(listen_on);
+    // TODO maybe notify the upper layer whether we have any active endpoints
+    /* Base Overlay */
+    base_overlay->start( base_communication, nodeId );
+    base_overlay->createSpoVNet( spovnetId );
+    base_overlay->joinSpoVNet( spovnetId );
+    /* Bootstrap */
+    const ptree& bootstrap_pt = config.get_child("bootstrap", empty_pt);
+    // Static Bootstrap
+    try
+    {
+        // read endpoint_set from config
+        EndpointSetPtr ep_set = endpoint_set::create_EndpointSet(
+                bootstrap_pt.get_child("direct"));
+        EndpointDescriptor ep = EndpointDescriptor::UNSPECIFIED();
+        ep.replace_endpoint_set(ep_set);
+        // try to connect
+        base_overlay->joinSpoVNet( spovnetId, ep);
+    }
+    catch ( ptree_bad_path& e )
+    {
+//        logging_info("No direct bootstrap info in config.");
+        cout << "No direct bootstrap info in config." << endl;
+    }
+    /* Bootstrap modules */
+    vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
+    // Bootstrap: Broadcast
+    if ( bootstrap_pt.get("broadcast", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypePeriodicBroadcast,""));
+    }
+    // Bootstrap: MDNS
+    if ( bootstrap_pt.get("mdns", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypeMulticastDns,""));
+    }
+    // Bootstrap: SDP
+    if ( bootstrap_pt.get("sdp", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypeBluetoothSdp,""));
+    }
+    // start automatic overlay bootstrapping modules
+    base_overlay->startBootstrapModules(internalmodules);
+}
+//void Node::join(const Name& vnetname) {
+//      spovnetId = vnetname.toSpoVNetId();
+//      nodeId = generateNodeId(name);
+//
+//      // start base comm if not started
+//      if( !ariba_mod.base_comm->isStarted() )
+//              ariba_mod.base_comm->start();
+//
+//      // start base overlay if not started
+//      // join against ourselfs
+//      if( !base_overlay->isStarted() )
+//              base_overlay->start( *ariba_mod.base_comm, nodeId );
+//      base_overlay->joinSpoVNet( spovnetId );
+//
+//      // join against static bootstrap points and
+//      // start automatic bootstrapping modules
+//      vector<AribaModule::BootstrapMechanism> mechanisms
+//              = ariba_mod.getBootstrapMechanisms(vnetname);
+//
+//      vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
+//
+//      BOOST_FOREACH(AribaModule::BootstrapMechanism m, mechanisms){
+//              switch(m){
+//                      case AribaModule::BootstrapMechanismStatic:
+//                      {
+//                              const communication::EndpointDescriptor* ep =
+//                                                      ariba_mod.getBootstrapNode(vnetname, m);
+//                                      if( ep != NULL && ep->isUnspecified() == false )
+//                                              base_overlay->joinSpoVNet( spovnetId, *ep);
+//                              break;
+//                      }
+//                      case AribaModule::BootstrapMechanismBroadcast:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypePeriodicBroadcast,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      case AribaModule::BootstrapMechanismMulticastDNS:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypeMulticastDns,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      case AribaModule::BootstrapMechanismSDP:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypeBluetoothSdp,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      default:
+//                              break;
+//              }
+//      }
+//
+//      // start automatic overlay bootstrapping modules
+//      base_overlay->startBootstrapModules(internalmodules);
+//
+//      // done
+//}
+//
+//void Node::initiate(const Name& vnetname, const SpoVNetProperties& parm) {
+//      utility::OverlayParameterSet ovrpset;
+//      ovrpset.setOverlayStructure(
+//                      (utility::OverlayParameterSet::_OverlayStructure)
+//                      parm.getBaseOverlayType()
+//                      );
+//
+//      spovnetId = vnetname.toSpoVNetId();
+//      nodeId = generateNodeId(name);
+//
+//      // start base comm if not started
+//      if( !ariba_mod.base_comm->isStarted() )
+//              ariba_mod.base_comm->start();
+//
+//      // start base overlay if not started
+//      if( !base_overlay->isStarted() )
+//              base_overlay->start( *ariba_mod.base_comm, nodeId );
+//
+//      base_overlay->createSpoVNet( spovnetId, ovrpset );
+//}
 void Node::leave() {
         base_overlay->stopBootstrapModules();
         base_overlay->leaveSpoVNet();
         ariba_mod.base_comm->stop();
+        base_communication->stop();  // XXX before Â»base_overlay->stop()Â« ??
         base_overlay->stop();
+}
 …
+}
+bool Node::isLinkDirect(const ariba::LinkID& lnk) const
+{
+    return base_overlay->isLinkDirect(lnk);
+}
+int Node::getHopCount(const ariba::LinkID& lnk) const
+{
+    return base_overlay->getHopCount(lnk);
+}
+/* +++++ Message sending +++++ */
+void Node::check_send_priority(uint8_t priority)
+{
+    if ( priority < send_priority::HIGHEST || priority > send_priority::LOWEST )
+        throw std::invalid_argument("Illegal priority");
+}
+// +++ new interface +++
+const SequenceNumber& Node::sendMessage(reboost::message_t msg, const LinkID& lnk, uint8_t priority)
+{
+    // check priority
+    check_send_priority(priority);
+    // * call base overlay *
+    return base_overlay->sendMessage(msg, lnk, priority);
+}
+// +++ legacy interface +++
+seqnum_t Node::sendMessage(const DataMessage& msg, const LinkID& lnk)
+{
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    try
+    {
+        base_overlay->sendMessage(message, lnk, send_priority::NORMAL);
+        return 0;
+    }
+    catch ( ariba::overlay::message_not_sent& e )
+    {
+        logging_warn("Message could not be sent. Dropped.");
+        return -1;
+    }
+}
+// +++ new interface +++
+const SequenceNumber& Node::sendMessage(reboost::message_t msg, const NodeID& nid,
+        const ServiceID& sid, uint8_t priority, const LinkProperties& req) {
+    // check priority
+    check_send_priority(priority);
+    // * call base overlay *
+    return base_overlay->sendMessage(msg, nid, priority, sid);
+}
+// +++ legacy interface +++
 seqnum_t Node::sendMessage(const DataMessage& msg, const NodeID& nid,
                 const ServiceID& sid, const LinkProperties& req) {
+        return base_overlay->sendMessage((Message*) msg, nid, sid);
+}
+//    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    reboost::message_t message;
+    message.push_back( ((Message*) msg)->serialize_into_shared_buffer() );
+    try
+    {
+        sendMessage(message, nid, sid, send_priority::NORMAL, req);
+        return 0;
+    }
+    catch ( ariba::overlay::message_not_sent& e )
+    {
+        logging_warn("Message could not be sent. Dropped.");
+        return -1;
+    }
+}
+// +++ new interface +++
+NodeID Node::sendMessageCloserToNodeID(reboost::message_t msg, const NodeID& nid, const ServiceID& sid,
+        uint8_t priority, const LinkProperties& req) {
+    // check priority
+    check_send_priority(priority);
+    // * call base overlay *
+    return base_overlay->sendMessageCloserToNodeID(msg, nid, priority, sid);
+}
+// +++ legacy interface +++
 NodeID Node::sendMessageCloserToNodeID(const DataMessage& msg, const NodeID& nid, const ServiceID& sid,
         const LinkProperties& req) {
+    return base_overlay->sendMessageCloserToNodeID((Message*) msg, nid, sid);
+}
+seqnum_t Node::sendMessage(const DataMessage& msg, const LinkID& lnk) {
+        return base_overlay->sendMessage((Message*) msg, lnk);
+}
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    return sendMessageCloserToNodeID(message, nid, sid, send_priority::NORMAL, req);
+}
+// +++ new interface +++
+void Node::sendBroadcastMessage(reboost::message_t msg, const ServiceID& sid, uint8_t priority) {
+    // check priority
+    check_send_priority(priority);
+    // * call base overlay *
+    return base_overlay->broadcastMessage(msg, sid, priority);
+}
+// +++ legacy interface +++
 void Node::sendBroadcastMessage(const DataMessage& msg, const ServiceID& sid) {
+        return base_overlay->broadcastMessage((Message*)msg, sid);
+}
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+        return sendBroadcastMessage(message, sid);
+}
+/* +++++ [Message sending] +++++ */
 bool Node::bind(NodeListener* listener) {
 …
         bool ret = base_overlay->bind(listener, sid);
         // now that we have a listener, we can ask if sniffing is ok
         if( ariba_mod.sideport_sniffer != NULL ){
                 base_overlay->registerSidePort(ariba_mod.sideport_sniffer);
+        }
+//      // now that we have a listener, we can ask if sniffing is ok
+//      if( ariba_mod.sideport_sniffer != NULL ){
+//              base_overlay->registerSidePort(ariba_mod.sideport_sniffer);
+//      }
         return ret;

source/ariba/Node.h

-              r10653
+              r12060
 namespace ariba {
         class Node;
+        namespace overlay {
+                class BaseOverlay;
+        }
+    namespace communication {
+        class BaseCommunication;
+    }
+    namespace overlay {
+        class BaseOverlay;
+    }
+}
 #include <vector>
 #include <iostream>
+#include <boost/foreach.hpp>
+#include <exception>
 #include "Module.h"
 #include "Identifiers.h"
 …
 #include "NodeListener.h"
 #include "Name.h"
 #include "AribaModule.h"
+//#include "AribaModule.h"
 #include "CommunicationListener.h"
 #include "DataMessage.h"
 #include "SideportListener.h"
+#include "ariba/overlay/SequenceNumber.h"
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
+#include <boost/property_tree/ptree.hpp>
 using std::vector;
 …
 namespace ariba {
+// typedef ariba::overlay::SequenceNumber SequenceNumber; // XXX see CommunicationListener
+using boost::property_tree::ptree;
+// sendMessage-Priorities
+struct send_priority
+{
+    enum SEND_PRIORITY
+    {
+        HIGHEST = 2,
+        HIGHER = 3,
+        NORMAL = 4,
+        LOWER = 5,
+        LOWEST = 6
+    };
+};
 /**
 …
  * @author Christoph Mayer <mayer@tm.uka.de>
  */
+// TODO do we really want to inherit from Module.. ?
 class Node: public Module {
 public:
         /**
          * Constructs a new node using a given ariba module
 …
          *   is a zero-terminated char-string.
          */
+        Node(AribaModule& ariba_mod, const Name& node_name = Name::UNSPECIFIED);
+//      Node(AribaModule& ariba_mod, const Name& node_name = Name::UNSPECIFIED);
+        // XXX EXPERIMENTAL
+        Node();
         /**
 …
         //--- node control ---
+    /**
+     * XXX EXPERIMENTAL
+     *
+     * Replaces initialte & join
+     */
+    void connect(const ptree& config);
+    // XXX DEPRECATED
         /**
          * This method instructs the node to join a particular spovnet.
 …
          * @param vnetId The SpoVNet name
          */
         void join(const Name& name);
+//      void join(const Name& name);
         /**
 …
          * @param param The SpoVNet properties
          */
         void initiate(const Name& name, const SpoVNetProperties& parm =
                         SpoVNetProperties::DEFAULT);
+//      void initiate(const Name& name, const SpoVNetProperties& parm =
+//                      SpoVNetProperties::DEFAULT);
         /**
 …
         void dropLink(const LinkID& lnk);
+        // message sending
+        /**
+         * Sends a one-shot message to a service. If link properties are specified,
+         * the node tries to fulfill those requirements. This may cause the node
+         * to first establish a temporary link, second sending the message and last
+         * dropping the link. This would result in a small amount of extra latency
+         * until the message is delivered. If reliable transport was selected,
+         * the method returns a sequence number and a communication event is
+         * triggered on message delivery or loss.
+         *
+         * @param msg The message to be sent
+         * @param nid The remote node identifier
+         * @param sid The remote service identifier
+         * @param req The requirements associated with the message
+         * @return A sequence number
+         */
+        /**
+         * Returns whether a link is direct or relayed over other nodes
+         * @param lnk LinkID of the link
+         * @return true if link is direct; false otherwise
+         */
+        bool isLinkDirect(const ariba::LinkID& lnk) const;
+        /**
+         * Returns the latest measured hop count on this link.
+         * NOTE: This is not guaranteed to be up to date.
+         *
+         * @param lnk LinkID of the link
+         * @return overlay hop count on this link
+         */
+        int getHopCount(const ariba::LinkID& lnk) const;
+        /* +++++ Message sending +++++ */
+    /**
+     * Sends a message via an established link. If reliable transport was
+     * selected, the method returns a sequence number and a communication event
+     * is triggered on message delivery or loss.
+     *
+     * +++ New interface, using efficient zero-copy reboost messages. +++
+     *
+     * @param msg The message to be sent
+     * @param lnk The link to be used for sending the message
+     */
+    const SequenceNumber& sendMessage(reboost::message_t msg, const LinkID& lnk, uint8_t priority=send_priority::NORMAL);
+    /**
+     * +++ Legacy interface, converts old ariba messages into new reboost messages. +++
+     */
+    seqnum_t sendMessage(const DataMessage& msg, const LinkID& lnk);
+    /**
+     * Sends a one-shot message to a service. If link properties are specified,
+     * the node tries to fulfill those requirements. This may cause the node
+     * to first establish a temporary link, second sending the message and last
+     * dropping the link. This would result in a small amount of extra latency
+     * until the message is delivered. If reliable transport was selected,
+     * the method returns a sequence number and a communication event is
+     * triggered on message delivery or loss.
+     *
+     * +++ New interface, using efficient zero-copy reboost messages. +++
+     *
+     * @param msg The message to be sent
+     * @param nid The remote node identifier
+     * @param sid The remote service identifier
+     * @param req The requirements associated with the message
+     * @return A sequence number
+     */
+    const SequenceNumber& sendMessage(reboost::message_t msg, const NodeID& nid, const ServiceID& sid,
+            uint8_t priority=send_priority::NORMAL, const LinkProperties& req = LinkProperties::DEFAULT);
+    /**
+     * +++ Legacy interface, converts old ariba messages into new reboost messages. +++
+     */
         seqnum_t sendMessage(const DataMessage& msg, const NodeID& nid, const ServiceID& sid,
                         const LinkProperties& req = LinkProperties::DEFAULT);
+        /**
+         * like the above function, but sends the message to the closest directly known node
+         * to the specified address
+         */
+    /**
+     * like the above function, but sends the message to the closest directly known node
+     * to the specified address
+     *
+     * +++ New interface, using efficient zero-copy reboost messages. +++
+     *
+     */
+    NodeID sendMessageCloserToNodeID(reboost::message_t msg, const NodeID& nid, const ServiceID& sid,
+            uint8_t priority=send_priority::NORMAL, const LinkProperties& req = LinkProperties::DEFAULT);
+    /**
+     * +++ Legacy interface, converts old ariba messages into new reboost messages. +++
+     */
     NodeID sendMessageCloserToNodeID(const DataMessage& msg, const NodeID& nid, const ServiceID& sid,
             const LinkProperties& req = LinkProperties::DEFAULT);
+        /**
+         * Sends a message via an established link. If reliable transport was
+         * selected, the method returns a sequence number and a communication event
+         * is triggered on message delivery or loss.
+         *
+         * @param msg The message to be sent
+         * @param lnk The link to be used for sending the message
+         */
+        seqnum_t sendMessage(const DataMessage& msg, const LinkID& lnk);
+        /**
+         * Sends a message to all known hosts in the overlay structure
+         * the nodes that are reached here depend on the overlay structure.
+         *
+         * @param msg The message to be send
+         * @param sid The id of the service that should receive the message
+         * @see getNeighborNodes
+         */
+    /**
+     * Sends a message to all known hosts in the overlay structure
+     * the nodes that are reached here depend on the overlay structure.
+     *
+     * +++ New interface, using efficient zero-copy reboost messages. +++
+     *
+     * @param msg The message to be send
+     * @param sid The id of the service that should receive the message
+     * @see getNeighborNodes
+     */
+        void sendBroadcastMessage(reboost::message_t msg, const ServiceID& sid, uint8_t priority=send_priority::NORMAL);
+    /**
+     * +++ Legacy interface, converts old ariba messages into new reboost messages. +++
+     */
         void sendBroadcastMessage(const DataMessage& msg, const ServiceID& sid);
+        /* +++++ [Message sending] +++++ */
         // --- communication listeners ---
 …
         /** @see Module.h */
         string getName() const;
+private:
+        inline void check_send_priority(uint8_t priority);
 protected:
         // friends
 …
         // member variables
         Name name;                             //< node name
         AribaModule& ariba_mod;                //< ariba module
+//      AribaModule* ariba_mod;                //< ariba module
         SpoVNetID spovnetId;                   //< current spovnet id
         NodeID nodeId;                             //< current node id
+        communication::BaseCommunication* base_communication;
         overlay::BaseOverlay* base_overlay;    //< the base overlay

source/ariba/SideportListener.cpp

-              r7468
+              r12060
 #include "ariba/overlay/BaseOverlay.h"
 #include "ariba/overlay/LinkDescriptor.h"
 #include "ariba/utility/addressing/endpoint_set.hpp"
+#include "ariba/utility/addressing2/tcpip_endpoint.hpp"
 using ariba::overlay::LinkDescriptor;
 …
         if( overlay == NULL ) return (Protocol)ret;
-        using namespace ariba::addressing;
         LinkDescriptor* link = NULL;
         BOOST_FOREACH( LinkDescriptor* lnk, overlay->links ){
 …
         if(bclink.isUnspecified() || bclink.remoteLocator == NULL) return (Protocol)ret;
         const address_v* locator = bclink.remoteLocator;
+        addressing2::EndpointPtr locator = bclink.remoteLocator;
+        if( locator->instanceof<tcpip_endpoint>() ){
+                tcpip_endpoint tcpip = *locator;
+                if( tcpip.address().is_v4() || tcpip.address().is_v4_mapped() ){
+        if( locator->get_category() == addressing2::endpoint_category::TCPIP )
+        {
+        boost::shared_ptr<addressing2::tcpip_endpoint> endp =
+                boost::dynamic_pointer_cast<addressing2::tcpip_endpoint>(locator);
+                if( endp->to_asio().address().is_v4() )
+                {
                         ret = SideportListener::ipv4;
                 }else if( tcpip.address().is_v6() ){
+                }else {
                         ret = SideportListener::ipv6;
+                }
         }else if( locator->instanceof<rfcomm_endpoint>() ){
+        }else if( locator->get_category() == addressing2::endpoint_category::BLUETOOTH ){
                 ret = SideportListener::rfcomm;
+        }

source/ariba/ariba.h

r3374	r12060
44	44	*/
45	45
46		~~#include "AribaModule.h"~~
47	46	#include "CommunicationListener.h"
48	47	#include "DataMessage.h"

source/ariba/communication/BaseCommunication.cpp

-              r10767
+              r12060
 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;
+                }
+        }
+}
 …
           transport( NULL ),
           messageReceiver( NULL ),
+          started( false )
+          started( false ),
+          listenOn_endpoints(new addressing2::endpoint_set())
+{
+}
 …
+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
 …
 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());
+}
 …
         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;
 …
         // 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
 …
+}
+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;
+}
 …
                 LinkDescriptor& linkDesc = queryLocalLink(link);
                 if (linkDesc.isUnspecified()) return EndpointDescriptor::UNSPECIFIED();
                 return linkDesc.remoteEndpoint;
+                return linkDesc.remoteDescriptor;
+        }
+}
 …
+}
+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,
 …
                     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;
 …
                         /// 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()
 …
                         // 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(),
 …
                         if( !allowlink ){
                                 logging_warn( "Overlay denied creation of link" );
-                                delete msg;
                                 return;
+                        }
 …
                         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
 …
                         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);
+                        }
 …
                 // 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 "
 …
                         // inform lisneters about link up event
                         BOOST_FOREACH( CommunicationEvents* i, eventListener ){
+                        foreach( CommunicationEvents* i, eventListener ){
                                 i->onLinkUp( ld.localLink, ld.localLocator, ld.remoteLocator );
+                        }
 …
                 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 );
 …
                         // remove the link descriptor
                         removeLink( localLink );
+                        removeLink( link_id );
                         // done
 …
                 // ---------------------------------------------------------------------
                 // 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;
+        }
+        }
+}
 …
         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 );
 …
+}
 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!
 …
+}
+/// 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

-              r10653
+              r12060
 #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"
 …
 // 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
 …
 #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;
+}
 …
 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;
 …
  * 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 {
 …
         /// 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
 …
          * @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);
         /**
 …
          * @return List of LinkID
          */
+        LinkIDs getLocalLinks(const address_v* addr) const;
+//      LinkIDs getLocalLinks(const address_v* addr) const;  // XXX aktuell
         /**
 …
         void unregisterEventListener(CommunicationEvents* _events);
-        /// called when a system event is emitted by system queue
-        virtual void handleSystemEvent(const SystemEvent& event);
         /**
 …
          */
         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:
 …
          */
         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
 …
                 /// 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);
+                        }
+                }
 …
                         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;
 …
                 /// 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;
 …
         /// 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
 …
         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

-              r5284
+              r12060
 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

-              r5284
+              r12060
 #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 {
 …
 using ariba::utility::LinkID;
 using ariba::utility::QoSParameterSet;
-using namespace ariba::addressing;
 class CommunicationEvents {
 …
          * @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);
         /**
 …
          * @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);
         /**
 …
          * @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);
         /**
 …
          */
         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

-              r5624
+              r12060
 namespace communication {
 vsznDefault(EndpointDescriptor);
+//vsznDefault(EndpointDescriptor);
+EndpointDescriptor::EndpointDescriptor(){
+EndpointDescriptor::EndpointDescriptor()  :
+        endpoints(new addressing2::endpoint_set())
+{
+}
 …
+}
+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

-              r7744
+              r12060
 #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 {
 …
 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:
 …
         /// 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);
 …
         /// convert end-points to string
         string toString() const {
                 return endpoints.to_string();
+                return endpoints->to_string();
+        }
 …
+        }
         /// 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 {
 …
         /// 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() {
 …
                 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;
 };
 …
 }} // 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

-              r9694
+              r12060
 #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"
 …
 using_serialization;
+// XXX This whole message is DEPRECATED
 class AribaBaseMsg : public Message {
         VSERIALIZEABLE;
 …
                 typeLinkReply = 2,
                 typeLinkClose = 3,
+                typeLinkUpdate = 4
+                typeLinkUpdate = 4,
+                typeDirectData = 5
         };
 …
 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

-              r10700
+              r12060
 # [License]
 add_headers(AribaBaseMsg.h)
+#add_headers(AribaBaseMsg.h)
 add_sources(AribaBaseMsg.cpp)
+#add_sources(AribaBaseMsg.cpp)

source/ariba/communication/networkinfo/AddressDiscovery.cpp

-              r10700
+              r12060
 #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>
 …
 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
 …
+}
+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

-              r5639
+              r12060
 #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 );
 };

source/ariba/overlay/BaseOverlay.cpp

-              r10653
+              r12060
 #include "ariba/utility/visual/DddVis.h"
 #include "ariba/utility/visual/ServerVis.h"
+#include <ariba/utility/misc/sha1.h>
 namespace ariba {
 namespace overlay {
+using namespace std;
+using ariba::transport::system_priority;
 #define visualInstance          ariba::utility::DddVis::instance()
 #define visualIdOverlay         ariba::utility::ServerVis::NETWORK_ID_BASE_OVERLAY
 #define visualIdBase            ariba::utility::ServerVis::NETWORK_ID_BASE_COMMUNICATION
+// time constants (in seconds)
+#define KEEP_ALIVE_TIME         60                      // send keep-alive message after link is not used for #s
+#define LINK_ESTABLISH_TIME_OUT 10                      // timeout: link requested but not up
+#define KEEP_ALIVE_TIME_OUT     KEEP_ALIVE_TIME + LINK_ESTABLISH_TIME_OUT     // timeout: no data received on this link (incl. keep-alive messages)
+#define AUTO_LINK_TIME_OUT      KEEP_ALIVE_TIME_OUT                    // timeout: auto link not used for #s
 …
 LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) {
         BOOST_FOREACH( LinkDescriptor* lp, links )
+        foreach( LinkDescriptor* lp, links )
                                 if ((communication ? lp->communicationId : lp->overlayId) == link)
                                         return lp;
 …
 const LinkDescriptor* BaseOverlay::getDescriptor( const LinkID& link, bool communication ) const {
         BOOST_FOREACH( const LinkDescriptor* lp, links )
+        foreach( const LinkDescriptor* lp, links )
                                 if ((communication ? lp->communicationId : lp->overlayId) == link)
                                         return lp;
 …
 /// returns a auto-link descriptor
+LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service ) {
+LinkDescriptor* BaseOverlay::getAutoDescriptor( const NodeID& node, const ServiceID& service )
+{
         // search for a descriptor that is already up
+        BOOST_FOREACH( LinkDescriptor* lp, links )
+                                if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->up && lp->keepAliveMissed == 0)
+                                        return lp;
+        foreach( LinkDescriptor* lp, links )
+    {
+        if (lp->autolink && lp->remoteNode == node && lp->service == service && isLinkVital(lp) )
+            return lp;
+    }
         // if not found, search for one that is about to come up...
+        BOOST_FOREACH( LinkDescriptor* lp, links )
+        if (lp->autolink && lp->remoteNode == node && lp->service == service && lp->keepAliveMissed == 0 )
+                return lp;
+        foreach( LinkDescriptor* lp, links )
+        {
+            time_t now = time(NULL);
+        if (lp->autolink && lp->remoteNode == node && lp->service == service
+                && difftime( now, lp->keepAliveReceived ) <= LINK_ESTABLISH_TIME_OUT )
+            return lp;
+        }
         return NULL;
+}
 …
 /// stabilizes link information
 void BaseOverlay::stabilizeLinks() {
+        // send keep-alive messages over established links
+        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+    time_t now = time(NULL);
+    // send keep-alive messages over established links
+        foreach( LinkDescriptor* ld, links )
+    {
                 if (!ld->up) continue;
+                OverlayMsg msg( OverlayMsg::typeLinkAlive,
+                                OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
+                if (ld->relayed) msg.setRouteRecord(true);
+                send_link( &msg, ld->overlayId );
+                if ( difftime( now, ld->keepAliveSent ) >= KEEP_ALIVE_TIME )
+                {
+                    logging_debug("[BaseOverlay] Sending KeepAlive over "
+                            << ld->to_string()
+                            << " after "
+                            << difftime( now, ld->keepAliveSent )
+                            << "s");
+            OverlayMsg msg( OverlayMsg::typeKeepAlive,
+                    OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
+            msg.setRouteRecord(true);
+            ld->keepAliveSent = now;
+            send_link( &msg, ld->overlayId, system_priority::OVERLAY );
+                }
+        }
         // iterate over all links and check for time boundaries
         vector<LinkDescriptor*> oldlinks;
+        time_t now = time(NULL);
+        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+                // keep alives and not up? yes-> link connection request is stale!
+                if ( !ld->up && difftime( now, ld->keepAliveTime ) >= 2 ) {
+                        // increase counter
+                        ld->keepAliveMissed++;
+                        // missed more than four keep-alive messages (10 sec)? -> drop link
+                        if (ld->keepAliveMissed > 4) {
+                                logging_info( "Link connection request is stale, closing: " << ld );
+                                oldlinks.push_back( ld );
+                                continue;
+                        }
+        foreach( LinkDescriptor* ld, links ) {
+                // link connection request stale?
+                if ( !ld->up && difftime( now, ld->keepAliveReceived ) >= LINK_ESTABLISH_TIME_OUT )  // NOTE: keepAliveReceived == now, on connection request
+                {
+            logging_info( "Link connection request is stale, closing: " << ld );
+            ld->failed = true;
+            oldlinks.push_back( ld );
+            continue;
+                }
                 if (!ld->up) continue;
                 // check if link is relayed and retry connecting directly
+                // TODO Mario: What happens here?  --> There are 3 attempts to replace a relayed link with a direct one. see: handleLinkReply
                 if ( ld->relayed && !ld->communicationUp && ld->retryCounter > 0) {
                         ld->retryCounter--;
 …
                 // remote used as relay flag
                 if ( ld->relaying && difftime( now, ld->timeRelaying ) > 10)
+                if ( ld->relaying && difftime( now, ld->timeRelaying ) > KEEP_ALIVE_TIME_OUT)  // TODO is this a reasonable timeout ??
                         ld->relaying = false;
 …
                 // auto-link time exceeded?
                 if ( ld->autolink && difftime( now, ld->lastuse ) > 30 ) {
+                if ( ld->autolink && difftime( now, ld->lastuse ) > AUTO_LINK_TIME_OUT ) {
                         oldlinks.push_back( ld );
                         continue;
 …
                 // keep alives missed? yes->
+                if ( difftime( now, ld->keepAliveTime ) > 4 ) {
+                        // increase counter
+                        ld->keepAliveMissed++;
+                        // missed more than four keep-alive messages (4 sec)? -> drop link
+                        if (ld->keepAliveMissed >= 2) {
+                                logging_info( "Link is stale, closing: " << ld );
+                                oldlinks.push_back( ld );
+                                continue;
+                        }
+                if ( difftime( now, ld->keepAliveReceived ) >= KEEP_ALIVE_TIME_OUT )
+                {
+            logging_info( "Link is stale, closing: " << ld );
+            ld->failed = true;
+            oldlinks.push_back( ld );
+            continue;
+                }
+        }
         // drop links
         BOOST_FOREACH( LinkDescriptor* ld, oldlinks ) {
+        foreach( LinkDescriptor* ld, oldlinks ) {
                 logging_info( "Link timed out. Dropping " << ld );
                 ld->relaying = false;
 …
+        }
+        // show link state
+        counter++;
+        if (counter>=4) showLinks();
+        if (counter>=4 || counter<0) counter = 0;
+        // show link state  (debug output)
+        if (counter>=10 || counter<0)
+    {
+            showLinks();
+            counter = 0;
+    }
+        else
+        {
+            counter++;
+        }
+}
 …
                 int i=0;
                 BOOST_FOREACH( LinkDescriptor* ld, links ) {
                         if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue;
+                foreach( LinkDescriptor* ld, links ) {
+                        if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID) continue;
                         bool found = false;
                         BOOST_FOREACH(NodeID& id, nodes)
+                        foreach(NodeID& id, nodes)
                         if (id  == ld->remoteNode) found = true;
                         if (found) continue;
 …
         int i=0;
         logging_info("--- link state -------------------------------");
         BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 string epd = "";
+                if (ld->isDirectVital())
+                        epd = getEndpointDescriptor(ld->remoteNode).toString();
+                if (isLinkDirectVital(ld))
+                {
+//                      epd = getEndpointDescriptor(ld->remoteNode).toString();
+                    epd = "Connection: ";
+                    epd += bc->get_local_endpoint_of_link(ld->communicationId)->to_string();
+                    epd += " <---> ";
+                    epd += bc->get_remote_endpoint_of_link(ld->communicationId)->to_string();
+                }
                 logging_info("LINK_STATE: " << i << ": " << ld << " " << epd);
 …
 // internal message delivery ---------------------------------------------------
+seqnum_t BaseOverlay::send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority )
+{
+    // set priority
+    message->setPriority(priority);
+    // wrap old OverlayMsg into reboost message
+    reboost::message_t wrapped_message = message->wrap_up_for_sending();
+    // send down to BaseCommunication
+    try
+    {
+        // * send *
+        return bc->sendMessage(bc_link, wrapped_message, priority, false);
+    }
+    catch ( communication::communication_message_not_sent& e )
+    {
+        ostringstream out;
+        out << "Communication message not sent: " << e.what();
+        throw message_not_sent(out.str());
+    }
+    throw logic_error("This should never happen!");
+}
 /// routes a message to its destination node
 void BaseOverlay::route( OverlayMsg* message ) {
+void BaseOverlay::route( OverlayMsg* message, const NodeID& last_hop )
+{
         // exceeded time-to-live? yes-> drop message
+        if (message->getNumHops() > message->getTimeToLive()) {
+                logging_warn("Message exceeded TTL. Dropping message and relay routes"
+                                "for recovery.");
+        if (message->getNumHops() > message->getTimeToLive())
+        {
+                logging_warn("Message exceeded TTL. Dropping message and relay routes "
+            << "for recovery. Hop count: " << (int) message->getNumHops());
                 removeRelayNode(message->getDestinationNode());
                 return;
 …
         // no-> forward message
+        else {
+        else
+        {
                 // destinastion myself? yes-> handle message
+                if (message->getDestinationNode() == nodeId) {
+                        logging_warn("Usually I should not route messages to myself!");
+                        Message msg;
+                        msg.encapsulate(message);
+                        handleMessage( &msg, NULL );
+                } else {
+                        // no->send message to next hop
+                        send( message, message->getDestinationNode() );
+                }
+        }
+                if (message->getDestinationNode() == nodeId)
+                {
+                        logging_warn("Usually I should not route messages to myself. And I won't!");
+                }
+                // no->send message to next hop
+                else
+                {
+                    try
+                    {
+                /* (deep) packet inspection to determine priority */
+                // BRANCH: typeData  -->  send with low priority
+                if ( message->getType() == OverlayMsg::typeData )
+                {
+                    // TODO think about implementing explicit routing queue (with active queue management??)
+                    send( message,
+                          message->getDestinationNode(),
+                          message->getPriority(),
+                          last_hop );
+                }
+                // BRANCH: internal message  -->  send with higher priority
+                else
+                {
+                    send( message,
+                          message->getDestinationNode(),
+                          system_priority::HIGH,
+                          last_hop );
+                }
+                    }
+                    catch ( message_not_sent& e )
+                    {
+                        logging_warn("Unable to route message of type "
+                                << message->getType()
+                                << " to "
+                                << message->getDestinationNode()
+                                << ". Reason: "
+                                << e.what());
+                        // inform sender
+                if ( message->getType() != OverlayMsg::typeMessageLost )
+                {
+                    report_lost_message(message);
+                }
+                    }
+                }
+        }
+}
+void BaseOverlay::report_lost_message( const OverlayMsg* message )
+{
+    OverlayMsg reply(OverlayMsg::typeMessageLost);
+    reply.setSeqNum(message->getSeqNum());
+    /**
+     * MessageLost-Message
+     *
+     * - Type of lost message
+     * - Hop count of lost message
+     * - Source-LinkID  of lost message
+     */
+    reboost::shared_buffer_t b(sizeof(uint8_t)*2);
+    b.mutable_data()[0] = message->getType();
+    b.mutable_data()[1] = message->getNumHops();
+    reply.append_buffer(b);
+    reply.append_buffer(message->getSourceLink().serialize());
+    try
+    {
+        send_node(&reply, message->getSourceNode(),
+                system_priority::OVERLAY,
+                OverlayInterface::OVERLAY_SERVICE_ID);
+    }
+    catch ( message_not_sent& e )
+    {
+        logging_warn("Tried to inform another node that we could'n route their message. But we were not able to send this error-message, too.");
+    }
+}
 /// sends a message to another node, delivers it to the base overlay class
+seqnum_t BaseOverlay::send( OverlayMsg* message, const NodeID& destination ) {
+seqnum_t BaseOverlay::send( OverlayMsg* message,
+        const NodeID& destination,
+        uint8_t priority,
+        const NodeID& last_hop ) throw(message_not_sent)
+{
         LinkDescriptor* next_link = NULL;
         // drop messages to unspecified destinations
+        if (destination.isUnspecified()) return -1;
+        // send messages to myself -> handle message and drop warning!
+        if (destination == nodeId) {
+                logging_warn("Sent message to myself. Handling message.")
+                Message msg;
+                msg.encapsulate(message);
+                handleMessage( &msg, NULL );
+                return -1;
+        if (destination.isUnspecified())
+            throw message_not_sent("No destination specified. Drop!");
+        // send messages to myself -> drop!
+    // TODO maybe this is not what we want. why not just deliver this message?
+    //   There is a similar check in the route function, there it should be okay.
+        if (destination == nodeId)
+        {
+            logging_warn("Sent message to myself. Drop!");
+            throw message_not_sent("Sent message to myself. Drop!");
+        }
         // use relay path?
+        if (message->isRelayed()) {
+        if (message->isRelayed())
+        {
                 next_link = getRelayLinkTo( destination );
+                if (next_link != NULL) {
+                if (next_link != NULL)
+                {
                         next_link->setRelaying();
+                        return bc->sendMessage(next_link->communicationId, message);
+                } else {
+                        logging_warn("Could not send message. No relay hop found to "
+                                        << destination << " -- trying to route over overlay paths ...")
+//                      logging_error("ERROR: " << debugInformation() );
+                //                      return -1;
+                }
+        }
+                        // * send message over relayed link *
+                        return send_overlaymessage_down(message, next_link->communicationId, priority);
+                }
+                else
+                {
+                        logging_warn("No relay hop found to " << destination
+                                << " -- trying to route over overlay paths ...")
+                }
+        }
         // last resort -> route over overlay path
         LinkID next_id = overlayInterface->getNextLinkId( destination );
+        if (next_id.isUnspecified()) {
+                logging_warn("Could not send message. No next hop found to " <<
+                                destination );
+                logging_error("ERROR: " << debugInformation() );
+                return -1;
+        }
+        // get link descriptor, up and running? yes-> send message
+        if ( next_id.isUnspecified() )
+        {
+        // apperently we're the closest node --> try second best node
+        //   NOTE: This is helpful if our routing table is not up-to-date, but
+        //   may lead to circles. So we have to be careful.
+        std::vector<const LinkID*> next_ids =
+            overlayInterface->getSortedLinkIdsTowardsNode( destination );
+        for ( int i = 0; i < next_ids.size(); i++ )
+        {
+            const LinkID& link = *next_ids[i];
+            if ( ! link.isUnspecified() )
+            {
+                next_id = link;
+                break;
+            }
+        }
+        // still no next hop found. drop.
+        if ( next_id.isUnspecified() )
+        {
+            logging_warn("Could not send message. No next hop found to " <<
+                destination );
+            logging_error("ERROR: " << debugInformation() );
+            throw message_not_sent("No next hop found.");
+        }
+        }
+        /* get link descriptor, do some checks and send message */
         next_link = getDescriptor(next_id);
+        if (next_link != NULL && next_link->up) {
+                // send message over relayed link
+                return send(message, next_link);
+        }
+        // no-> error, dropping message
+        else {
+                logging_warn("Could not send message. Link not known or up");
+                logging_error("ERROR: " << debugInformation() );
+                return -1;
+        }
+        // not reached-> fail
+        return -1;
+}
+    // check pointer
+    if ( next_link == NULL )
+    {
+        // NOTE: this shuldn't happen
+        throw message_not_sent("Could not send message. Link not known.");
+    }
+    // avoid circles
+    if ( next_link->remoteNode == last_hop )
+    {
+        // XXX logging_debug
+        logging_info("Possible next hop would create a circle: "
+            << next_link->remoteNode);
+        throw message_not_sent("Could not send message. Possible next hop would create a circle.");
+    }
+    // check if link is up
+        if ( ! next_link->up)
+        {
+        logging_warn("Could not send message. Link not up");
+        logging_error("ERROR: " << debugInformation() );
+        throw message_not_sent("Could not send message. Link not up");
+        }
+        // * send message over overlay link *
+        return send(message, next_link, priority);
+}
 /// send a message using a link descriptor, delivers it to the base overlay class
+seqnum_t BaseOverlay::send( OverlayMsg* message, LinkDescriptor* ldr, bool ignore_down ) {
+seqnum_t BaseOverlay::send( OverlayMsg* message,
+        LinkDescriptor* ldr,
+        uint8_t priority ) throw(message_not_sent)
+{
         // check if null
+        if (ldr == NULL) {
+                logging_error("Can not send message to " << message->getDestinationAddress());
+                return -1;
+        if (ldr == NULL)
+        {
+        ostringstream out;
+        out << "Can not send message to " << message->getDestinationAddress();
+        throw message_not_sent(out.str());
+        }
         // check if up
         if (!ldr->up && !ignore_down) {
                 logging_error("Can not send message. Link not up:" << ldr );
+        if ( !ldr->up )
+        {
                 logging_error("DEBUG_INFO: " << debugInformation() );
+                return -1;
+        }
+        LinkDescriptor* ld = NULL;
+        // handle relayed link
+        if (ldr->relayed) {
+        ostringstream out;
+        out << "Can not send message. Link not up:" << ldr->to_string();
+        throw message_not_sent(out.str());
+        }
+        LinkDescriptor* next_hop_ld = NULL;
+        // BRANCH: relayed link
+        if (ldr->relayed)
+        {
                 logging_debug("Resolving direct link for relayed link to "
                                 << ldr->remoteNode);
+                ld = getRelayLinkTo( ldr->remoteNode );
+                if (ld==NULL) {
+                        logging_error("No relay path found to link " << ldr );
+                next_hop_ld = getRelayLinkTo( ldr->remoteNode );
+                if (next_hop_ld==NULL)
+                {
                         logging_error("DEBUG_INFO: " << debugInformation() );
+                        return -1;
+                }
+                ld->setRelaying();
+                ostringstream out;
+                out << "No relay path found to link: " << ldr;
+                throw message_not_sent(out.str());
+                }
+                next_hop_ld->setRelaying();
                 message->setRelayed(true);
+        } else
+                ld = ldr;
+        // handle direct link
+        if (ld->communicationUp) {
+                logging_debug("send(): Sending message over direct link.");
+                return bc->sendMessage( ld->communicationId, message );
+        } else {
+                logging_error("send(): Could not send message. "
+                                "Not a relayed link and direct link is not up.");
+                return -1;
+        }
+        return -1;
+        }
+        // BRANCH: direct link
+        else
+        {
+                next_hop_ld = ldr;
+        }
+        // check next hop-link
+        if ( ! next_hop_ld->communicationUp)
+        {
+            throw message_not_sent( "send(): Could not send message."
+                    " Not a relayed link and direct link is not up." );
+        }
+        // send over next link
+    logging_debug("send(): Sending message over direct link.");
+    return send_overlaymessage_down(message, next_hop_ld->communicationId, priority);
+}
 seqnum_t BaseOverlay::send_node( OverlayMsg* message, const NodeID& remote,
+                const ServiceID& service) {
+        uint8_t priority, const ServiceID& service) throw(message_not_sent)
+{
         message->setSourceNode(nodeId);
         message->setDestinationNode(remote);
         message->setService(service);
+        return send( message, remote );
+}
+seqnum_t BaseOverlay::send_link( OverlayMsg* message, const LinkID& link,bool ignore_down ) {
+        return send( message, remote, priority );
+}
+void BaseOverlay::send_link( OverlayMsg* message,
+        const LinkID& link,
+        uint8_t priority ) throw(message_not_sent)
+{
         LinkDescriptor* ld = getDescriptor(link);
+        if (ld==NULL) {
+                logging_error("Cannot find descriptor to link id=" << link.toString());
+                return -1;
+        }
+        if (ld==NULL)
+        {
+            throw message_not_sent("Cannot find descriptor to link id=" + link.toString());
+        }
         message->setSourceNode(nodeId);
         message->setDestinationNode(ld->remoteNode);
 …
         message->setService(ld->service);
         message->setRelayed(ld->relayed);
+        return send( message, ld, ignore_down );
+    try
+    {
+        // * send message *
+        send( message, ld, priority );
+    }
+    catch ( message_not_sent& e )
+    {
+        // drop failed link
+        ld->failed = true;
+        dropLink(ld->overlayId);
+    }
+}
 …
                 // relay link still used and alive?
                 if (ld==NULL
+                                || !ld->isDirectVital()
+                                || difftime(route.used, time(NULL)) > 8) {
+                                || !isLinkDirectVital(ld)
+                                || difftime(route.used, time(NULL)) > KEEP_ALIVE_TIME_OUT)  // TODO this was set to 8 before.. Is the new timeout better?
+                {
                         logging_info("Forgetting relay information to node "
                                         << route.node.toString() );
 …
         if (message->isRelayed()) {
                 // try to find source node
                 BOOST_FOREACH( relay_route& route, relay_routes ) {
+                foreach( relay_route& route, relay_routes ) {
                         // relay route found? yes->
                         if ( route.node == message->getDestinationNode() ) {
 …
                 // try to find source node
                 BOOST_FOREACH( relay_route& route, relay_routes ) {
+                foreach( relay_route& route, relay_routes ) {
                         // relay route found? yes->
 …
                                 if (route.hops > message->getNumHops()
                                                 || rld == NULL
                                                 || !rld->isDirectVital()) {
+                                                || !isLinkDirectVital(ld)) {
                                         logging_info("Updating relay information to node "
                                                         << route.node.toString()
                                                         << " reducing to " << message->getNumHops() << " hops.");
+                                                        << " reducing to " << (int) message->getNumHops() << " hops.");
                                         route.hops = message->getNumHops();
                                         route.link = ld->overlayId;
 …
 LinkDescriptor* BaseOverlay::getRelayLinkTo( const NodeID& remote ) {
         // try to find source node
         BOOST_FOREACH( relay_route& route, relay_routes ) {
+        foreach( relay_route& route, relay_routes ) {
                 if (route.node == remote ) {
                         LinkDescriptor* ld = getDescriptor( route.link );
                         if (ld==NULL || !ld->isDirectVital()) return NULL; else {
+                        if (ld==NULL || !isLinkDirectVital(ld)) return NULL; else {
                                 route.used = time(NULL);
                                 return ld;
 …
 // ----------------------------------------------------------------------------
 void BaseOverlay::start( BaseCommunication& _basecomm, const NodeID& _nodeid ) {
+void BaseOverlay::start( BaseCommunication* _basecomm, const NodeID& _nodeid ) {
         logging_info("Starting...");
         // set parameters
         bc = &_basecomm;
+        bc = _basecomm;
         nodeId = _nodeid;
 …
         // timer for auto link management
+        Timer::setInterval( 1000 );
+        Timer::setInterval( 1000 ); // XXX
+//      Timer::setInterval( 10000 );
         Timer::start();
 …
                 overlayInterface->joinOverlay();
                 state = BaseOverlayStateCompleted;
                 BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                         i->onJoinCompleted( spovnetId );
 …
         // gather all service links
         vector<LinkID> servicelinks;
+        BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links )
+        {
                 if( ld->service != OverlayInterface::OVERLAY_SERVICE_ID )
                         servicelinks.push_back( ld->overlayId );
 …
         // drop all service links
+        BOOST_FOREACH( LinkID lnk, servicelinks )
+        dropLink( lnk );
+        foreach( LinkID lnk, servicelinks )
+        {
+            logging_debug("Dropping service link " << lnk.toString());
+            dropLink( lnk );
+        }
         // let the node leave the spovnet overlay interface
         logging_debug( "Leaving overlay" );
         if( overlayInterface != NULL )
+        {
                 overlayInterface->leaveOverlay();
+        }
         // drop still open bootstrap links
+        BOOST_FOREACH( LinkID lnk, bootstrapLinks )
+        bc->dropLink( lnk );
+        foreach( LinkID lnk, bootstrapLinks )
+        {
+            logging_debug("Dropping bootstrap link " << lnk.toString());
+            bc->dropLink( lnk );
+        }
         // change to inalid state
 …
         // inform all registered services of the event
+        BOOST_FOREACH( NodeListener* i, nodeListeners ) {
+        foreach( NodeListener* i, nodeListeners )
+        {
                 if( ret ) i->onLeaveCompleted( spovnetId );
                 else i->onLeaveFailed( spovnetId );
 …
                 state = BaseOverlayStateInvalid;
                 BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                 i->onJoinFailed( spovnetId );
 …
                 const ServiceID& service ) {
+    // TODO What if we already have a Link to this node and this service id?
         // do not establish a link to myself!
+        if (remote == nodeId) return LinkID::UNSPECIFIED;
+        if (remote == nodeId) return
+                LinkID::UNSPECIFIED;
         // create a link descriptor
         LinkDescriptor* ld = addDescriptor();
 …
         ld->service = service;
         ld->listener = getListener(ld->service);
+    // initialize sequence numbers
+    ld->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short();
+    logging_debug("Creating new link with initial SeqNum: " << ld->last_sent_seqnum);
         // create link request message
 …
         msg.setRelayed(true);
         msg.setRegisterRelay(true);
+//      msg.setRouteRecord(true);
+    msg.setSeqNum(ld->last_sent_seqnum);
         // debug message
 …
         );
         // sending message to node
+        send_node( &msg, ld->remoteNode, ld->service );
+        try
+        {
+            // * send *
+            seqnum_t seq = send_node( &msg, ld->remoteNode, system_priority::OVERLAY, ld->service );
+        }
+        catch ( message_not_sent& e )
+        {
+            logging_warn("Link request not sent: " << e.what());
+            // Message not sent. Cancel link request.
+            SystemQueue::instance().scheduleCall(
+                    boost::bind(
+                            &BaseOverlay::__onLinkEstablishmentFailed,
+                            this,
+                            ld->overlayId)
+                );
+        }
         return ld->overlayId;
+}
+/// NOTE: "id" is an Overlay-LinkID
+void BaseOverlay::__onLinkEstablishmentFailed(const LinkID& id)
+{
+    // TODO This code redundant. But also it's not easy to aggregate in one function.
+    // get descriptor for link
+    LinkDescriptor* ld = getDescriptor(id, false);
+    if ( ld == NULL ) return; // not found? ->ignore!
+    logging_debug( "__onLinkEstablishmentFaild: " << ld );
+    // removing relay link information
+    removeRelayLink(ld->overlayId);
+    // inform listeners about link down
+    ld->communicationUp = false;
+    if (!ld->service.isUnspecified())
+    {
+        CommunicationListener* lst = getListener(ld->service);
+        if(lst != NULL) lst->onLinkFail( ld->overlayId, ld->remoteNode );
+        sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
+    }
+    // delete all queued messages (auto links)
+    if( ld->messageQueue.size() > 0 ) {
+        logging_warn( "Dropping link " << id.toString() << " that has "
+                << ld->messageQueue.size() << " waiting messages" );
+        ld->flushQueue();
+    }
+    // erase mapping
+    eraseDescriptor(ld->overlayId);
+}
 /// drops an established link
+void BaseOverlay::dropLink(const LinkID& link) {
+        logging_info( "Dropping link (initiated locally):" << link.toString() );
+void BaseOverlay::dropLink(const LinkID& link)
+{
+        logging_info( "Dropping link: " << link.toString() );
         // find the link item to drop
         LinkDescriptor* ld = getDescriptor(link);
+        if( ld == NULL ) {
+        if( ld == NULL )
+        {
                 logging_warn( "Can't drop link, link is unknown!");
                 return;
 …
         // delete all queued messages
+        if( ld->messageQueue.size() > 0 ) {
+        if( ld->messageQueue.size() > 0 )
+        {
                 logging_warn( "Dropping link " << ld->overlayId.toString() << " that has "
                                 << ld->messageQueue.size() << " waiting messages" );
                 ld->flushQueue();
+        }
+        // inform sideport and listener
+        if(ld->listener != NULL)
+                ld->listener->onLinkDown( ld->overlayId, ld->remoteNode );
+        sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId );
+        // do not drop relay links
+        if (!ld->relaying) {
+                // drop the link in base communication
+                if (ld->communicationUp) bc->dropLink( ld->communicationId );
+                // erase descriptor
+                eraseDescriptor( ld->overlayId );
+        } else {
+                ld->dropAfterRelaying = true;
+        }
+        // inform application and remote note (but only once)
+        //   NOTE: If we initiated the drop, this function is called twice, but on
+        //   the second call, there is noting to do.
+        if ( ld->up && ! ld->failed )
+        {
+        // inform sideport and listener
+        if(ld->listener != NULL)
+        {
+            ld->listener->onLinkDown( ld->overlayId, ld->remoteNode );
+        }
+        sideport->onLinkDown(ld->overlayId, this->nodeId, ld->remoteNode, this->spovnetId );
+        // send link-close to remote node
+        logging_info("Sending LinkClose message to remote node.");
+        OverlayMsg close_msg(OverlayMsg::typeLinkClose);
+        send_link(&close_msg, link, system_priority::OVERLAY);
+        // deactivate link
+        ld->up = false;
+//         ld->closing = true;
+        }
+        else if ( ld->failed )
+    {
+        // inform listener
+        if( ld->listener != NULL )
+        {
+            ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
+        }
+        ld->up = false;
+        __removeDroppedLink(ld->overlayId);
+    }
+}
+/// called from typeLinkClose-handler
+void BaseOverlay::__removeDroppedLink(const LinkID& link)
+{
+    // find the link item to drop
+    LinkDescriptor* ld = getDescriptor(link);
+    if( ld == NULL )
+    {
+        return;
+    }
+    // do not drop relay links
+    if (!ld->relaying)
+    {
+        // drop the link in base communication
+        if (ld->communicationUp)
+        {
+            bc->dropLink( ld->communicationId );
+        }
+        // erase descriptor
+        eraseDescriptor( ld->overlayId );
+    }
+    else
+    {
+        ld->dropAfterRelaying = true;
+    }
+}
 …
 /// internal send message, always use this functions to send messages over links
+seqnum_t BaseOverlay::sendMessage( const Message* message, const LinkID& link ) {
+const SequenceNumber& BaseOverlay::sendMessage( reboost::message_t message,
+        const LinkID& link,
+        uint8_t priority ) throw(message_not_sent)
+{
         logging_debug( "Sending data message on link " << link.toString() );
         // get the mapping for this link
         LinkDescriptor* ld = getDescriptor(link);
+        if( ld == NULL ) {
+                logging_error("Could not send message. "
+                                << "Link not found id=" << link.toString());
+                return -1;
+        if( ld == NULL )
+        {
+            throw message_not_sent("Could not send message. Link not found id=" + link.toString());
+        }
         // check if the link is up yet, if its an auto link queue message
+        if( !ld->up ) {
+        if( !ld->up )
+        {
                 ld->setAutoUsed();
+                if( ld->autolink ) {
+                if( ld->autolink )
+                {
                         logging_info("Auto-link " << link.toString() << " not up, queue message");
+                        Data data = data_serialize( message );
+                        const_cast<Message*>(message)->dropPayload();
+                        ld->messageQueue.push_back( new Message(data) );
+                } else {
+                        logging_error("Link " << link.toString() << " not up, drop message");
+                }
+                return -1;
+        }
+        // compile overlay message (has service and node id)
+        OverlayMsg overmsg( OverlayMsg::typeData );
+        overmsg.encapsulate( const_cast<Message*>(message) );
+        // send message over relay/direct/overlay
+        return send_link( &overmsg, ld->overlayId );
+}
+seqnum_t BaseOverlay::sendMessage(const Message* message,
+                const NodeID& node, const ServiceID& service) {
+                        // queue message
+                LinkDescriptor::message_queue_entry msg;
+                msg.message = message;
+                msg.priority = priority;
+                        ld->messageQueue.push_back( msg );
+                        return SequenceNumber::DISABLED;  // TODO what to return if message is queued?
+                }
+                else
+                {
+                    throw message_not_sent("Link " + link.toString() + " not up, drop message");
+                }
+        }
+        // TODO AKTUELL: sequence numbers
+        // TODO seqnum on fast path ?
+        ld->last_sent_seqnum.increment();
+        /* choose fast-path for direct links; normal overlay-path otherwise */
+        // BRANCH: direct link
+        if ( ld->communicationUp && !ld->relayed )
+        {
+            // * send down to BaseCommunication *
+            try
+            {
+                bc->sendMessage(ld->communicationId, message, priority, true);
+        }
+        catch ( communication::communication_message_not_sent& e )
+        {
+            ostringstream out;
+            out << "Communication message on fast-path not sent: " << e.what();
+            throw message_not_sent(out.str());
+        }
+        }
+        // BRANCH: use (slow) overlay-path
+        else
+        {
+        // compile overlay message (has service and node id)
+        OverlayMsg overmsg( OverlayMsg::typeData );
+        overmsg.set_payload_message(message);
+        // set SeqNum
+        if ( ld->transmit_seqnums )
+        {
+            overmsg.setSeqNum(ld->last_sent_seqnum);
+        }
+        logging_debug("Sending Message with SeqNum: " << overmsg.getSeqNum());
+        // send message over relay/direct/overlay
+        send_link( &overmsg, ld->overlayId, priority );
+        }
+        // return seqnum
+        return ld->last_sent_seqnum;
+}
+const SequenceNumber& BaseOverlay::sendMessage(reboost::message_t message,
+                const NodeID& node, uint8_t priority, const ServiceID& service) {
         // find link for node and service
 …
                 if( ld == NULL ) {
                         logging_error( "Failed to establish auto-link.");
                         return -1;
+            throw message_not_sent("Failed to establish auto-link.");
+                }
                 ld->autolink = true;
 …
         // send / queue message
         return sendMessage( message, ld->overlayId );
+}
 NodeID BaseOverlay::sendMessageCloserToNodeID(const Message* message,
         const NodeID& address, const ServiceID& service) {
+        return sendMessage( message, ld->overlayId, priority );
+}
+NodeID BaseOverlay::sendMessageCloserToNodeID(reboost::message_t message,
+        const NodeID& address, uint8_t priority, const ServiceID& service) {
     if ( overlayInterface->isClosestNodeTo(address) )
 …
     if ( closest_node != NodeID::UNSPECIFIED )
+    {
         seqnum_t seqnum = sendMessage(message, closest_node, service);
+        sendMessage(message, closest_node, priority, service);
+    }
     return closest_node;  // XXX return seqnum ?? tuple? closest_node via (non const) reference?
+    return closest_node;  // return seqnum ?? tuple? closest_node via (non const) reference?
+}
 // ----------------------------------------------------------------------------
 …
         // see if we can find the node in our own table
         BOOST_FOREACH(const LinkDescriptor* ld, links){
+        foreach(const LinkDescriptor* ld, links){
                 if(ld->remoteNode != node) continue;
                 if(!ld->communicationUp) continue;
 …
 void BaseOverlay::onLinkUp(const LinkID& id,
+                const address_v* local, const address_v* remote) {
+        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote)
+{
         logging_debug( "Link up with base communication link id=" << id );
 …
         LinkDescriptor* ld = getDescriptor(id, true);
         // handle bootstrap link we initiated
+        // BRANCH: handle bootstrap link we initiated
         if( std::find(bootstrapLinks.begin(), bootstrapLinks.end(), id) != bootstrapLinks.end() ){
                 logging_info(
                                 "Join has been initiated by me and the link is now up. " <<
+                                "LinkID: " << id.toString() <<
                                 "Sending out join request for SpoVNet " << spovnetId.toString()
                 );
 …
                                 OverlayInterface::OVERLAY_SERVICE_ID, nodeId );
                 JoinRequest joinRequest( spovnetId, nodeId );
+                overlayMsg.encapsulate( &joinRequest );
+                bc->sendMessage( id, &overlayMsg );
+                overlayMsg.append_buffer(joinRequest.serialize_into_shared_buffer());
+                send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY);
                 return;
+        }
         // no link found? -> link establishment from remote, add one!
+        // BRANCH: link establishment from remote, add one!
         if (ld == NULL) {
                 ld = addDescriptor( id );
 …
                 // in this case, do not inform listener, since service it unknown
                 // -> wait for update message!
+                // link mapping found? -> send update message with node-id and service id
+        } else {
+        }
+        // BRANCH: We requested this link in the first place
+        else
+        {
                 logging_info( "onLinkUp descriptor (initiated locally):" << ld );
 …
                 ld->fromRemote = false;
+                // if link is a relayed link->convert to direct link
+                if (ld->relayed) {
+                        logging_info( "Converting to direct link: " << ld );
+                // BRANCH: this was a relayed link before --> convert to direct link
+                //   TODO do we really have to send a message here?
+                if (ld->relayed)
+                {
                         ld->up = true;
                         ld->relayed = false;
+                        logging_info( "Converting to direct link: " << ld );
+                        // send message
                         OverlayMsg overMsg( OverlayMsg::typeLinkDirect );
                         overMsg.setSourceLink( ld->overlayId );
                         overMsg.setDestinationLink( ld->remoteLink );
+                        send_link( &overMsg, ld->overlayId );
+                } else {
+                        send_link( &overMsg, ld->overlayId, system_priority::OVERLAY );
+                    // inform listener
+                    if( ld->listener != NULL)
+                        ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
+                }
+        /* NOTE: Chord is opening direct-links in it's setup routine which are
+         *   neither set to "relayed" nor to "up". To activate these links a
+         *   typeLinkUpdate must be sent.
+         *
+         * This branch is would also be taken when we had a working link before
+         *   (ld->up == true). I'm not sure if this case does actually happen
+         *   and whether it's tested.
+         */
+                else
+                {
                         // note: necessary to validate the link on the remote side!
                         logging_info( "Sending out update" <<
 …
                         // compile and send update message
                         OverlayMsg overlayMsg( OverlayMsg::typeLinkUpdate );
-                        overlayMsg.setSourceLink(ld->overlayId);
                         overlayMsg.setAutoLink( ld->autolink );
+                        send_link( &overlayMsg, ld->overlayId, true );
+                    overlayMsg.setSourceNode(nodeId);
+                    overlayMsg.setDestinationNode(ld->remoteNode);
+                    overlayMsg.setSourceLink(ld->overlayId);
+                    overlayMsg.setDestinationLink(ld->remoteLink);
+                    overlayMsg.setService(ld->service);
+                    overlayMsg.setRelayed(false);
+                    // TODO ld->communicationId = id ??
+                    send_overlaymessage_down(&overlayMsg, id, system_priority::OVERLAY);
+                }
+        }
 …
 void BaseOverlay::onLinkDown(const LinkID& id,
+                const address_v* local, const address_v* remote) {
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
         // erase bootstrap links
         vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
 …
+}
+void BaseOverlay::onLinkFail(const LinkID& id,
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
+        logging_debug( "Link fail with base communication link id=" << id );
+//      // erase bootstrap links
+//      vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
+//      if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
+//
+//      // get descriptor for link
+//      LinkDescriptor* ld = getDescriptor(id, true);
+//      if ( ld == NULL ) return; // not found? ->ignore!
+//      logging_debug( "Link failed id=" << ld->overlayId.toString() );
+//
+//      // inform listeners
+//      ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
+//      sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
+        logging_debug( "  ... calling onLinkDown ..." );
+        onLinkDown(id, local, remote);
+}
 void BaseOverlay::onLinkChanged(const LinkID& id,
+                const address_v* oldlocal, const address_v* newlocal,
+                const address_v* oldremote, const address_v* newremote) {
+        // get descriptor for link
+        LinkDescriptor* ld = getDescriptor(id, true);
+        if ( ld == NULL ) return; // not found? ->ignore!
+        logging_debug( "onLinkChanged descriptor: " << ld );
+        // inform listeners
+        ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
+        sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId );
+        // autolinks: refresh timestamp
+        ld->setAutoUsed();
+}
+void BaseOverlay::onLinkFail(const LinkID& id,
+                const address_v* local, const address_v* remote) {
+        logging_debug( "Link fail with base communication link id=" << id );
+        // erase bootstrap links
+        vector<LinkID>::iterator it = std::find( bootstrapLinks.begin(), bootstrapLinks.end(), id );
+        if( it != bootstrapLinks.end() ) bootstrapLinks.erase( it );
+        // get descriptor for link
+        LinkDescriptor* ld = getDescriptor(id, true);
+        if ( ld == NULL ) return; // not found? ->ignore!
+        logging_debug( "Link failed id=" << ld->overlayId.toString() );
+        // inform listeners
+        ld->listener->onLinkFail( ld->overlayId, ld->remoteNode );
+        sideport->onLinkFail( id, this->nodeId, ld->remoteNode, this->spovnetId );
+}
+void BaseOverlay::onLinkQoSChanged(const LinkID& id, const address_v* local,
+                const address_v* remote, const QoSParameterSet& qos) {
+        logging_debug( "Link quality changed with base communication link id=" << id );
+        // get descriptor for link
+        LinkDescriptor* ld = getDescriptor(id, true);
+        if ( ld == NULL ) return; // not found? ->ignore!
+        logging_debug( "Link quality changed id=" << ld->overlayId.toString() );
+}
+bool BaseOverlay::onLinkRequest( const LinkID& id, const address_v* local,
+                const address_v* remote ) {
+        const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+        const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote)
+{
+    // get descriptor for link
+    LinkDescriptor* ld = getDescriptor(id, true);
+    if ( ld == NULL ) return; // not found? ->ignore!
+    logging_debug( "onLinkChanged descriptor: " << ld );
+    // inform listeners
+    ld->listener->onLinkChanged( ld->overlayId, ld->remoteNode );
+    sideport->onLinkChanged( id, this->nodeId, ld->remoteNode, this->spovnetId );
+    // autolinks: refresh timestamp
+    ld->setAutoUsed();
+}
+//void BaseOverlay::onLinkQoSChanged(const LinkID& id,
+//        const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+//        const QoSParameterSet& qos)
+//{
+//      logging_debug( "Link quality changed with base communication link id=" << id );
+//
+//      // get descriptor for link
+//      LinkDescriptor* ld = getDescriptor(id, true);
+//      if ( ld == NULL ) return; // not found? ->ignore!
+//      logging_debug( "Link quality changed id=" << ld->overlayId.toString() );
+//}
+bool BaseOverlay::onLinkRequest(const LinkID& id,
+        const addressing2::EndpointPtr local,
+        const addressing2::EndpointPtr remote)
+{
         logging_debug("Accepting link request from " << remote->to_string() );
+        // TODO ask application..?
         return true;
+}
 /// handles a message from base communication
+bool BaseOverlay::receiveMessage(const Message* message,
+                const LinkID& link, const NodeID& ) {
+bool BaseOverlay::receiveMessage( reboost::shared_buffer_t message,
+                const LinkID& link,
+                const NodeID&,
+                bool bypass_overlay )
+{
         // get descriptor for link
         LinkDescriptor* ld = getDescriptor( link, true );
+        return handleMessage( message, ld, link );
+        /* choose fastpath for direct links; normal overlay-path otherwise */
+        if ( bypass_overlay && ld )
+        {
+        // message received --> link is alive
+        ld->keepAliveReceived = time(NULL);
+        // hop count on this link
+        ld->hops = 0;
+        // hand over to CommunicationListener (aka Application)
+            CommunicationListener* lst = getListener(ld->service);
+            if ( lst != NULL )
+            {
+                lst->onMessage(
+                        message,
+                        ld->remoteNode,
+                        ld->overlayId,
+                    SequenceNumber::DISABLED,
+                        NULL );
+                return true;
+            }
+            return false;
+        }
+        else
+        {
+            return handleMessage( message, ld, link );
+        }
+}
 …
 /// Handle spovnet instance join requests
 bool BaseOverlay::handleJoinRequest( OverlayMsg* overlayMsg, const LinkID& bcLink ) {
+bool BaseOverlay::handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink )
+{
         // decapsulate message
+        JoinRequest* joinReq = overlayMsg->decapsulate<JoinRequest>();
+        JoinRequest joinReq;
+        joinReq.deserialize_from_shared_buffer(message);
         logging_info( "Received join request for spovnet " <<
                         joinReq->getSpoVNetID().toString() );
+                        joinReq.getSpoVNetID().toString() );
         // check spovnet id
         if( joinReq->getSpoVNetID() != spovnetId ) {
+        if( joinReq.getSpoVNetID() != spovnetId ) {
                 logging_error(
                                 "Received join request for spovnet we don't handle " <<
                                 joinReq->getSpoVNetID().toString() );
+                delete joinReq;
+                                joinReq.getSpoVNetID().toString() );
                 return false;
+        }
 …
         logging_info( "Sending join reply for spovnet " <<
                         spovnetId.toString() << " to node " <<
                         overlayMsg->getSourceNode().toString() <<
+                        source.toString() <<
                         ". Result: " << (allow ? "allowed" : "denied") );
         joiningNodes.push_back( overlayMsg->getSourceNode() );
+        joiningNodes.push_back( source );
         // return overlay parameters
 …
                         << getEndpointDescriptor().toString() )
         OverlayParameterSet parameters = overlayInterface->getParameters();
+        // create JoinReplay Message
         OverlayMsg retmsg( OverlayMsg::typeJoinReply,
                         OverlayInterface::OVERLAY_SERVICE_ID, nodeId );
+        JoinReply replyMsg( spovnetId, parameters,
+                        allow, getEndpointDescriptor() );
+        retmsg.encapsulate(&replyMsg);
+        bc->sendMessage( bcLink, &retmsg );
+        delete joinReq;
+        JoinReply replyMsg( spovnetId, parameters, allow );
+        retmsg.append_buffer(replyMsg.serialize_into_shared_buffer());
+        // XXX This is unlovely clash between the old message system and the new one,
+        // but a.t.m. we can't migrate everything to the new system at once..
+        // ---> Consider the EndpointDescriptor as part of the JoinReply..
+        retmsg.append_buffer(getEndpointDescriptor().serialize());
+        // * send *
+        send_overlaymessage_down(&retmsg, bcLink, system_priority::OVERLAY);
         return true;
+}
 /// Handle replies to spovnet instance join requests
+bool BaseOverlay::handleJoinReply( OverlayMsg* overlayMsg, const LinkID& bcLink ) {
+bool BaseOverlay::handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink )
+{
         // decapsulate message
         logging_debug("received join reply message");
+        JoinReply* replyMsg = overlayMsg->decapsulate<JoinReply>();
+        JoinReply replyMsg;
+        EndpointDescriptor endpoints;
+        reboost::shared_buffer_t buff = replyMsg.deserialize_from_shared_buffer(message);
+        buff = endpoints.deserialize(buff);
         // correct spovnet?
         if( replyMsg->getSpoVNetID() != spovnetId ) { // no-> fail
+        if( replyMsg.getSpoVNetID() != spovnetId ) { // no-> fail
                 logging_error( "Received SpoVNet join reply for " <<
                                 replyMsg->getSpoVNetID().toString() <<
+                                replyMsg.getSpoVNetID().toString() <<
                                 " != " << spovnetId.toString() );
+                delete replyMsg;
                 return false;
+        }
         // access granted? no -> fail
         if( !replyMsg->getJoinAllowed() ) {
+        if( !replyMsg.getJoinAllowed() ) {
                 logging_error( "Our join request has been denied" );
 …
                 // inform all registered services of the event
                 BOOST_FOREACH( NodeListener* i, nodeListeners )
+                foreach( NodeListener* i, nodeListeners )
                 i->onJoinFailed( spovnetId );
-                delete replyMsg;
                 return true;
+        }
 …
         logging_debug( "Using bootstrap end-point "
                         << replyMsg->getBootstrapEndpoint().toString() );
+                        << endpoints.toString() );
         // create overlay structure from spovnet parameter set
 …
                 overlayInterface = OverlayFactory::create(
                                 *this, replyMsg->getParam(), nodeId, this );
+                                *this, replyMsg.getParam(), nodeId, this );
                 // overlay structure supported? no-> fail!
 …
                         // inform all registered services of the event
                         BOOST_FOREACH( NodeListener* i, nodeListeners )
+                        foreach( NodeListener* i, nodeListeners )
                         i->onJoinFailed( spovnetId );
-                        delete replyMsg;
                         return true;
+                }
 …
                 overlayInterface->createOverlay();
                 overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );
                 overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() );
+                overlayInterface->joinOverlay( endpoints );
+                overlayBootstrap.recordJoin( endpoints );
                 // update ovlvis
 …
                 // inform all registered services of the event
+                BOOST_FOREACH( NodeListener* i, nodeListeners )
+                i->onJoinCompleted( spovnetId );
+                delete replyMsg;
+        } else {
+                foreach( NodeListener* i, nodeListeners )
+                    i->onJoinCompleted( spovnetId );
+        }
+        else
+        {
                 // this is not the first bootstrap, just join the additional node
                 logging_debug("not first-time bootstrapping");
+                overlayInterface->joinOverlay( replyMsg->getBootstrapEndpoint() );
+                overlayBootstrap.recordJoin( replyMsg->getBootstrapEndpoint() );
+                delete replyMsg;
+                overlayInterface->joinOverlay( endpoints );
+                overlayBootstrap.recordJoin( endpoints );
         } // if( overlayInterface == NULL )
 …
+bool BaseOverlay::handleData( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
+bool BaseOverlay::handleData( reboost::shared_buffer_t message, OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
         // get service
+        const ServiceID& service = overlayMsg->getService();
+        const ServiceID& service = ld->service; //overlayMsg->getService();
         logging_debug( "Received data for service " << service.toString()
                         << " on link " << overlayMsg->getDestinationLink().toString() );
 …
         if(lst != NULL){
                 lst->onMessage(
+                                overlayMsg,
+                                overlayMsg->getSourceNode(),
+                                overlayMsg->getDestinationLink()
+                                message,
+//                              overlayMsg->getSourceNode(),
+//                              overlayMsg->getDestinationLink(),
+                                ld->remoteNode,
+                                ld->overlayId,
+                overlayMsg->getSeqNum(),
+                                overlayMsg
                 );
+        }
 …
+}
+bool BaseOverlay::handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg )
+{
+    /**
+     * Deserialize MessageLost-Message
+     *
+     * - Type of lost message
+     * - Hop count of lost message
+     * - Source-LinkID  of lost message
+     */
+    const uint8_t* buff = message(0, sizeof(uint8_t)*2).data();
+    uint8_t type = buff[0];
+    uint8_t hops = buff[1];
+    LinkID linkid;
+    linkid.deserialize(message(sizeof(uint8_t)*2));
+    logging_warn("Node " << msg->getSourceNode()
+            << " informed us, that our message of type " << (int) type
+            << " is lost after traveling " << (int) hops << " hops."
+            << " (LinkID: " << linkid.toString());
+    // TODO switch-case ?
+    // BRANCH: LinkRequest --> link request failed
+    if ( type == OverlayMsg::typeLinkRequest )
+    {
+        __onLinkEstablishmentFailed(linkid);
+    }
+    // BRANCH: Data --> link disrupted. Drop link.
+    //   (We could use something more advanced here. e.g. At least send a
+    //    keep-alive message and wait for a keep-alive reply.)
+    if ( type == OverlayMsg::typeData )
+    {
+        LinkDescriptor* link_desc = getDescriptor(linkid);
+        if ( link_desc )
+        {
+            link_desc->failed = true;
+        }
+        dropLink(linkid);
+    }
+    // BRANCH: ping lost
+    if ( type == OverlayMsg::typePing )
+    {
+        CommunicationListener* lst = getListener(msg->getService());
+        if( lst != NULL )
+        {
+            lst->onPingLost(msg->getSourceNode());
+        }
+    }
+    return true;
+}
+bool BaseOverlay::handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
+    // TODO AKTUELL: implement interfaces: Node::ping(node); BaseOverlay::ping(node)
+    bool send_pong = false;
+    // inform application and ask permission to send a pong message
+    CommunicationListener* lst = getListener(overlayMsg->getService());
+    if( lst != NULL )
+    {
+        send_pong = lst->onPing(overlayMsg->getSourceNode());
+    }
+    // send pong message if allowed
+    if ( send_pong )
+    {
+        OverlayMsg pong_msg(OverlayMsg::typePong);
+        pong_msg.setSeqNum(overlayMsg->getSeqNum());
+        // send message
+        try
+        {
+            send_node( &pong_msg,
+                overlayMsg->getSourceNode(),
+                system_priority::OVERLAY,
+                overlayMsg->getService() );
+        }
+        catch ( message_not_sent& e )
+        {
+            logging_info("Could not send Pong-Message to node: " <<
+                overlayMsg->getSourceNode());
+        }
+    }
+}
+bool BaseOverlay::handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
+    // inform application
+    CommunicationListener* lst = getListener(overlayMsg->getService());
+    if( lst != NULL )
+    {
+        lst->onPong(overlayMsg->getSourceNode());
+    }
+}
 bool BaseOverlay::handleLinkUpdate( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
 …
                 overlayMsg->setSourceLink(ld->overlayId);
                 overlayMsg->setService(ld->service);
                 send( overlayMsg, ld );
+                send( overlayMsg, ld, system_priority::OVERLAY );
+        }
 …
         if( ld->messageQueue.size() > 0 ) {
                 logging_info( "Sending out queued messages on link " << ld );
+                BOOST_FOREACH( Message* msg, ld->messageQueue ) {
+                        sendMessage( msg, ld->overlayId );
                         delete msg;
+                }
+        foreach( LinkDescriptor::message_queue_entry msg, ld->messageQueue )
+        {
+            sendMessage( msg.message, ld->overlayId, msg.priority );
+        }
                 ld->messageQueue.clear();
+        }
 …
 /// handle a link request and reply
 bool BaseOverlay::handleLinkRequest( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
-        logging_info( "Link request received from node id=" << overlayMsg->getSourceNode() );
         //TODO: Check if a request has already been sent using getSourceLink() ...
 …
         ldn->remoteNode = overlayMsg->getSourceNode();
         ldn->remoteLink = overlayMsg->getSourceLink();
+        ldn->hops = overlayMsg->getNumHops();
+    // initialize sequence numbers
+    ldn->last_sent_seqnum = SequenceNumber::createRandomSeqNum_Short();
+    logging_debug("Creating new link with initial SeqNum: " << ldn->last_sent_seqnum);
         // update time-stamps
         ldn->setAlive();
         ldn->setAutoUsed();
+        logging_info( "Link request received from node id="
+                << overlayMsg->getSourceNode()
+                << " LINK: "
+                << ldn);
         // create reply message and send back!
         overlayMsg->swapRoles(); // swap source/destination
         overlayMsg->setType(OverlayMsg::typeLinkReply);
         overlayMsg->setSourceLink(ldn->overlayId);
-        overlayMsg->setSourceEndpoint( bc->getEndpointDescriptor() );
         overlayMsg->setRelayed(true);
+        send( overlayMsg, ld ); // send back to link
+//      overlayMsg->setRouteRecord(true);
+    overlayMsg->setSeqNum(ld->last_sent_seqnum);
+        // TODO aktuell do the same thing in the typeLinkRequest-Message, too. But be careful with race conditions!!
+        // append our endpoints (for creation of a direct link)
+        overlayMsg->set_payload_message(bc->getEndpointDescriptor().serialize());
+        send( overlayMsg, ld, system_priority::OVERLAY ); // send back to link
         // inform listener
 …
+}
+bool BaseOverlay::handleLinkReply( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
+bool BaseOverlay::handleLinkReply(
+        OverlayMsg* overlayMsg,
+        reboost::shared_buffer_t sub_message,
+        LinkDescriptor* ld )
+{
+    // deserialize EndpointDescriptor
+    EndpointDescriptor endpoints;
+    endpoints.deserialize(sub_message);
         // find link request
         LinkDescriptor* ldn = getDescriptor(overlayMsg->getDestinationLink());
 …
         // debug message
         logging_debug( "Link request reply received. Establishing link"
+        logging_info( "Link request reply received. Establishing link"
                         << " for service " << overlayMsg->getService().toString()
                         << " with local id=" << overlayMsg->getDestinationLink()
                         << " and remote link id=" << overlayMsg->getSourceLink()
+                        << " to " << overlayMsg->getSourceEndpoint().toString()
+                        << " to " << endpoints.toString()
+                        << " hop count: " << overlayMsg->getRouteRecord().size()
         );
 …
                 logging_info( "Sending out queued messages on link " <<
                                 ldn->overlayId.toString() );
                 BOOST_FOREACH( Message* msg, ldn->messageQueue ) {
                         sendMessage( msg, ldn->overlayId );
                         delete msg;
+                foreach( LinkDescriptor::message_queue_entry msg, ldn->messageQueue )
+                {
+                        sendMessage( msg.message, ldn->overlayId, msg.priority );
+                }
                 ldn->messageQueue.clear();
 …
         // try to replace relay link with direct link
         ldn->retryCounter = 3;
         ldn->endpoint = overlayMsg->getSourceEndpoint();
+        ldn->endpoint = endpoints;
         ldn->communicationId =  bc->establishLink( ldn->endpoint );
 …
 /// handle a keep-alive message for a link
+bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld ) {
+bool BaseOverlay::handleLinkAlive( OverlayMsg* overlayMsg, LinkDescriptor* ld )
+{
         LinkDescriptor* rld = getDescriptor(overlayMsg->getDestinationLink());
+        if ( rld != NULL ) {
+                logging_debug("Keep-Alive for " <<
+                                overlayMsg->getDestinationLink() );
+        if ( rld != NULL )
+        {
+                logging_debug("Keep-Alive for " << overlayMsg->getDestinationLink() );
                 if (overlayMsg->isRouteRecord())
+                {
                         rld->routeRecord = overlayMsg->getRouteRecord();
+                }
+                // set alive
                 rld->setAlive();
+                /* answer keep alive */
+                if ( overlayMsg->getType() == OverlayMsg::typeKeepAlive )
+                {
+            time_t now = time(NULL);
+            logging_debug("[BaseOverlay] Answering KeepAlive over "
+                    << ld->to_string()
+                    << " after "
+                    << difftime( now, ld->keepAliveSent )
+                    << "s");
+            OverlayMsg msg( OverlayMsg::typeKeepAliveReply,
+                    OverlayInterface::OVERLAY_SERVICE_ID, nodeId, ld->remoteNode );
+            msg.setRouteRecord(true);
+            ld->keepAliveSent = now;
+            send_link( &msg, ld->overlayId, system_priority::OVERLAY );
+                }
                 return true;
+        } else {
+                logging_error("Keep-Alive for "
+        }
+        else
+        {
+                logging_error("No Keep-Alive for "
                                 << overlayMsg->getDestinationLink() << ": link unknown." );
                 return false;
 …
         // erase the original descriptor
         eraseDescriptor(ld->overlayId);
+    // inform listener
+    if( rld->listener != NULL)
+        rld->listener->onLinkChanged( rld->overlayId, rld->remoteNode );
         return true;
+}
 /// handles an incoming message
+bool BaseOverlay::handleMessage( const Message* message, LinkDescriptor* ld,
+                const LinkID bcLink ) {
+        logging_debug( "Handling message: " << message->toString());
+bool BaseOverlay::handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld,
+                const LinkID bcLink )
+{
         // decapsulate overlay message
+        OverlayMsg* overlayMsg =
+                        const_cast<Message*>(message)->decapsulate<OverlayMsg>();
+        if( overlayMsg == NULL ) return false;
+        OverlayMsg* overlayMsg = new OverlayMsg();
+        reboost::shared_buffer_t sub_buff = overlayMsg->deserialize_from_shared_buffer(message);
+//      // XXX debug
+//      logging_info( "Received overlay message."
+//              << " Hops: " << (int) overlayMsg->getNumHops()
+//              << " Type: " << (int) overlayMsg->getType()
+//              << " Payload size: " << sub_buff.size()
+//             << " SeqNum: " << overlayMsg->getSeqNum() );
         // increase number of hops
         overlayMsg->increaseNumHops();
 …
         // handle signaling messages (do not route!)
         if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart &&
+                        overlayMsg->getType()<=OverlayMsg::typeSignalingEnd ) {
+                overlayInterface->onMessage(overlayMsg, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED);
+                        overlayMsg->getType()<=OverlayMsg::typeSignalingEnd )
+        {
+                overlayInterface->onMessage(overlayMsg, sub_buff, NodeID::UNSPECIFIED, LinkID::UNSPECIFIED);
                 delete overlayMsg;
                 return true;
 …
                                 << " to " << overlayMsg->getDestinationNode()
                 );
+                route( overlayMsg );
+//              // XXX testing AKTUELL
+//        logging_info("MARIO: Routing message "
+//                << " from " << overlayMsg->getSourceNode()
+//                << " to " << overlayMsg->getDestinationNode() );
+//        logging_info( "Type: " << overlayMsg->getType() << " Payload size: " << sub_buff.size());
+                overlayMsg->append_buffer(sub_buff);
+                route( overlayMsg, ld->remoteNode );
                 delete overlayMsg;
                 return true;
+        }
+        // handle base overlay message
+        /* handle base overlay message */
         bool ret = false; // return value
+        switch ( overlayMsg->getType() ) {
+        // data transport messages
+        case OverlayMsg::typeData:
+                ret = handleData(overlayMsg, ld);                       break;
+                // overlay setup messages
+        case OverlayMsg::typeJoinRequest:
+                ret = handleJoinRequest(overlayMsg, bcLink );   break;
+        case OverlayMsg::typeJoinReply:
+                ret = handleJoinReply(overlayMsg, bcLink );     break;
+                // link specific messages
+        case OverlayMsg::typeLinkRequest:
+                ret = handleLinkRequest(overlayMsg, ld );       break;
+        case OverlayMsg::typeLinkReply:
+                ret = handleLinkReply(overlayMsg, ld );         break;
+        case OverlayMsg::typeLinkUpdate:
+                ret = handleLinkUpdate(overlayMsg, ld );        break;
+        case OverlayMsg::typeLinkAlive:
+                ret = handleLinkAlive(overlayMsg, ld );         break;
+        case OverlayMsg::typeLinkDirect:
+                ret = handleLinkDirect(overlayMsg, ld );        break;
+                // handle unknown message type
+        default: {
+                logging_error( "received message in invalid state! don't know " <<
+                                "what to do with this message of type " << overlayMsg->getType() );
+                ret = false;
+                break;
+        }
+        try
+        {
+        switch ( overlayMsg->getType() )
+        {
+            // data transport messages
+            case OverlayMsg::typeData:
+            {
+                // NOTE: On relayed links, Â»ldÂ« does not point to our link, but on the relay link.
+                LinkDescriptor* end_to_end_ld = getDescriptor(overlayMsg->getDestinationLink());
+                if ( ! end_to_end_ld )
+                {
+                    logging_warn("Error: Data-Message claims to belong to a link we don't know.");
+                    ret = false;
+                }
+                else
+                {
+                    // message received --> link is alive
+                    end_to_end_ld->keepAliveReceived = time(NULL);
+                    // hop count on this link
+                    end_to_end_ld->hops = overlayMsg->getNumHops();
+                    // * call handler *
+                    ret = handleData(sub_buff, overlayMsg, end_to_end_ld);
+                }
+                break;
+            }
+            case OverlayMsg::typeMessageLost:
+                ret = handleLostMessage(sub_buff, overlayMsg);
+                break;
+                // overlay setup messages
+            case OverlayMsg::typeJoinRequest:
+                ret = handleJoinRequest(sub_buff, overlayMsg->getSourceNode(), bcLink );        break;
+            case OverlayMsg::typeJoinReply:
+                ret = handleJoinReply(sub_buff, bcLink );       break;
+                // link specific messages
+            case OverlayMsg::typeLinkRequest:
+                ret = handleLinkRequest(overlayMsg, ld );       break;
+            case OverlayMsg::typeLinkReply:
+                ret = handleLinkReply(overlayMsg, sub_buff, ld );       break;
+            case OverlayMsg::typeLinkUpdate:
+                ret = handleLinkUpdate(overlayMsg, ld );        break;
+            case OverlayMsg::typeKeepAlive:
+            case OverlayMsg::typeKeepAliveReply:
+                ret = handleLinkAlive(overlayMsg, ld );         break;
+            case OverlayMsg::typeLinkDirect:
+                ret = handleLinkDirect(overlayMsg, ld );        break;
+            case OverlayMsg::typeLinkClose:
+            {
+                dropLink(overlayMsg->getDestinationLink());
+                __removeDroppedLink(overlayMsg->getDestinationLink());
+                break;
+            }
+            /// ping over overlay path (or similar)
+            case OverlayMsg::typePing:
+            {
+                ret = handlePing(overlayMsg, ld);
+                break;
+            }
+            case OverlayMsg::typePong:
+            {
+                ret = handlePong(overlayMsg, ld);
+                break;
+            }
+                // handle unknown message type
+            default:
+            {
+                logging_error( "received message in invalid state! don't know " <<
+                        "what to do with this message of type " << overlayMsg->getType() );
+                ret = false;
+                break;
+            }
+        }
+        }
+        catch ( reboost::illegal_sub_buffer& e )
+        {
+            logging_error( "Failed to create sub-buffer while reading message: Â»"
+                    << e.what()
+                    << "Â« Message too short? ");
+            assert(false); // XXX
+        }
 …
 // ----------------------------------------------------------------------------
 void BaseOverlay::broadcastMessage(Message* message, const ServiceID& service) {
+void BaseOverlay::broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority) {
         logging_debug( "broadcasting message to all known nodes " <<
                         "in the overlay from service " + service.toString() );
-        if(message == NULL) return;
-        message->setReleasePayload(false);
         OverlayInterface::NodeList nodes = overlayInterface->getKnownNodes(true);
 …
                 NodeID& id = nodes.at(i);
                 if(id == this->nodeId) continue; // don't send to ourselfs
+                if(i+1 == nodes.size()) message->setReleasePayload(true); // release payload on last send
                 sendMessage( message, id, service );
+                sendMessage( message, id, priority, service );
+        }
+}
 …
 vector<LinkID> BaseOverlay::getLinkIDs( const NodeID& nid ) const {
         vector<LinkID> linkvector;
         BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 if( ld->remoteNode == nid || nid == NodeID::UNSPECIFIED ) {
                         linkvector.push_back( ld->overlayId );
 …
         updateVisual();
+}
+/* link status */
+bool BaseOverlay::isLinkDirect(const ariba::LinkID& lnk) const
+{
+    const LinkDescriptor* ld = getDescriptor(lnk);
+    if (!ld)
+        return false;
+    return ld->communicationUp && !ld->relayed;
+}
+int BaseOverlay::getHopCount(const ariba::LinkID& lnk) const
+{
+    const LinkDescriptor* ld = getDescriptor(lnk);
+    if (!ld)
+        return -1;
+    return ld->hops;
+}
+bool BaseOverlay::isLinkVital(const LinkDescriptor* link) const
+{
+    time_t now = time(NULL);
+    return link->up && difftime( now, link->keepAliveReceived ) <= KEEP_ALIVE_TIME_OUT; // TODO is this too long for a "vital" link..?
+}
+bool BaseOverlay::isLinkDirectVital(const LinkDescriptor* link) const
+{
+    return isLinkVital(link) && link->communicationUp && !link->relayed;
+}
+/* [link status] */
 void BaseOverlay::updateVisual(){
 …
         static set<NodeID> linkset;
         set<NodeID> remotenodes;
         BOOST_FOREACH( LinkDescriptor* ld, links ) {
                 if (!ld->isVital() || ld->service != OverlayInterface::OVERLAY_SERVICE_ID)
+        foreach( LinkDescriptor* ld, links ) {
+                if (!isLinkVital(ld) || ld->service != OverlayInterface::OVERLAY_SERVICE_ID)
                         continue;
 …
         do{
                 changed = false;
                 BOOST_FOREACH(NodeID n, linkset){
+                foreach(NodeID n, linkset){
                         if(remotenodes.find(n) == remotenodes.end()){
                                 visualInstance.visDisconnect(visualIdBase, this->nodeId, n, "");
 …
         do{
                 changed = false;
                 BOOST_FOREACH(NodeID n, remotenodes){
+                foreach(NodeID n, remotenodes){
                         if(linkset.find(n) == linkset.end()){
                                 visualInstance.visConnect(visualIdBase, this->nodeId, n, "");
 …
         // dump link state
         s << "--- link state -------------------------------" << endl;
         BOOST_FOREACH( LinkDescriptor* ld, links ) {
+        foreach( LinkDescriptor* ld, links ) {
                 s << "link " << i << ": " << ld << endl;
                 i++;

source/ariba/overlay/BaseOverlay.h

-              r10653
+              r12060
 #include <vector>
 #include <deque>
+#include <stdexcept>
 #include <boost/foreach.hpp>
+#ifdef ECLIPSE_PARSER
+    #define foreach(a, b) for(a : b)
+#else
+    #define foreach(a, b) BOOST_FOREACH(a, b)
+#endif
 #include "ariba/utility/messages.h"
 …
 #include "ariba/overlay/modules/OverlayStructureEvents.h"
 #include "ariba/overlay/OverlayBootstrap.h"
+#include "ariba/overlay/SequenceNumber.h"
 // forward declarations
 …
 using ariba::communication::BaseCommunication;
 using ariba::communication::CommunicationEvents;
+// transport
+//using ariba::transport::system_priority;
 // utilities
 …
 using ariba::utility::Demultiplexer;
 using ariba::utility::MessageReceiver;
-using ariba::utility::MessageSender;
 using ariba::utility::seqnum_t;
 using ariba::utility::Timer;
 …
 namespace overlay {
+using namespace ariba::addressing;
+class message_not_sent: public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    explicit message_not_sent(const string& __arg)  :
+        std::runtime_error(__arg)
+    {
+    }
+    virtual ~message_not_sent() throw() {}
+};
 class LinkDescriptor;
 …
                 protected Timer {
+        friend class OneHop;
+//      friend class OneHop;  // DEPRECATED
         friend class Chord;
         friend class ariba::SideportListener;
 …
 public:
         /**
          * Constructs an empty non-functional base overlay instance
 …
          * Starts the Base Overlay instance
          */
         void start(BaseCommunication& _basecomm, const NodeID& _nodeid);
+        void start(BaseCommunication* _basecomm, const NodeID& _nodeid);
         /**
 …
          * Starts a link establishment procedure to the specfied node
          * for the service with id service
+         *
+         *
          * @param node Destination node id
          * @param service Service to connect to
 …
         void dropLink( const LinkID& link );
+        /* +++++ Message sending +++++ */
         /// sends a message over an existing link
+        seqnum_t sendMessage(const Message* message, const LinkID& link );
+        const SequenceNumber& sendMessage(reboost::message_t message,
+                const LinkID& link,
+                uint8_t priority ) throw(message_not_sent);
         /// sends a message to a node and a specific service
+        seqnum_t sendMessage(const Message* message, const NodeID& remote,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+    const SequenceNumber& sendMessage(reboost::message_t message,
+                const NodeID& remote,
+                uint8_t priority,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
     /**
 …
      *  @return NodeID of the (closest) destination node;
      */
         NodeID sendMessageCloserToNodeID(const Message* message, const NodeID& address,
                 const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+        NodeID sendMessageCloserToNodeID(reboost::message_t message, const NodeID& address,
+                uint8_t priority, const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
         /**
 …
          * Depending on the structure of the overlay, this can be very different.
          */
+        void broadcastMessage(Message* message, const ServiceID& service);
+        void broadcastMessage(reboost::message_t message, const ServiceID& service, uint8_t priority);
+        /* +++++ [Message sending] +++++ */
         /**
          * Returns the end-point descriptor of a link.
 …
          */
         void leaveSpoVNet();
+        /* link status */
+        bool isLinkDirect(const ariba::LinkID& lnk) const;
+        int getHopCount(const ariba::LinkID& lnk) const;
+    bool isLinkVital(const LinkDescriptor* link) const;
+    bool isLinkDirectVital(const LinkDescriptor* link) const;
 protected:
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual void onLinkUp(const LinkID& id, const address_v* local,
+                const address_v* remote);
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual void onLinkDown(const LinkID& id, const address_v* local,
+                const address_v* remote);
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual void onLinkChanged(const LinkID& id,
+                const address_v* oldlocal, const address_v* newlocal,
+                const address_v* oldremote, const address_v* newremote);
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual void onLinkFail(const LinkID& id, const address_v* local,
+                const address_v* remote);
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual void onLinkQoSChanged(const LinkID& id,
+                const address_v* local, const address_v* remote,
+                const QoSParameterSet& qos);
+        /**
+         * @see ariba::communication::CommunicationEvents.h
+         */
+        virtual bool onLinkRequest(const LinkID& id, const address_v* local,
+                const address_v* remote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual bool onLinkRequest(const LinkID& id,
+            const addressing2::EndpointPtr local,
+            const addressing2::EndpointPtr remote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkUp(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkDown(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+    virtual void onLinkChanged(const LinkID& id,
+        const addressing2::EndpointPtr oldlocal,  const addressing2::EndpointPtr newlocal,
+        const addressing2::EndpointPtr oldremote, const addressing2::EndpointPtr newremote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     *
+     * NOTE: Just calls onLinkDown (at the moment..)
+     */
+    virtual void onLinkFail(const LinkID& id,
+            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote);
+    /**
+     * @see ariba::communication::CommunicationEvents.h
+     */
+//    virtual void onLinkQoSChanged(const LinkID& id,
+//            const addressing2::EndpointPtr local, const addressing2::EndpointPtr remote,
+//            const QoSParameterSet& qos);
         /**
          * Processes a received message from BaseCommunication
 …
          * the node the message came from!
          */
+        virtual bool receiveMessage( const Message* message, const LinkID& link,
+                const NodeID& );
+        virtual bool receiveMessage( reboost::shared_buffer_t message,
+                const LinkID& link,
+                const NodeID&,
+                bool bypass_overlay );
         /**
 …
         std::string getLinkHTMLInfo();
+private:
+        /// NOTE: "id" is an Overlay-LinkID
+        void __onLinkEstablishmentFailed(const LinkID& id);
+        /// called from typeLinkClose-handler
+    void __removeDroppedLink(const LinkID& link);
 private:
         /// is the base overlay started yet
 …
         /// demultiplexes a incoming message with link descriptor
         bool handleMessage( const Message* message, LinkDescriptor* ld,
+        bool handleMessage( reboost::shared_buffer_t message, LinkDescriptor* ld,
                 const LinkID bcLink = LinkID::UNSPECIFIED );
         // handle data and signalling messages
+        bool handleData( OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleData( reboost::shared_buffer_t message, OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleLostMessage( reboost::shared_buffer_t message, OverlayMsg* msg );
         bool handleSignaling( OverlayMsg* msg, LinkDescriptor* ld );
         // handle join request / reply messages
         bool handleJoinRequest( OverlayMsg* msg, const LinkID& bcLink );
         bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink );
+        bool handleJoinRequest( reboost::shared_buffer_t message, const NodeID& source, const LinkID& bcLink );
+        bool handleJoinReply( reboost::shared_buffer_t message, const LinkID& bcLink );
         // handle link messages
         bool handleLinkRequest( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkReply( OverlayMsg* msg, LinkDescriptor* ld );
+        bool handleLinkReply( OverlayMsg* msg, reboost::shared_buffer_t sub_message, LinkDescriptor* ld );
         bool handleLinkUpdate( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkDirect( OverlayMsg* msg, LinkDescriptor* ld );
         bool handleLinkAlive( OverlayMsg* msg, LinkDescriptor* ld );
+    // ping-pong over overlaypath/routing
+    bool handlePing( OverlayMsg* overlayMsg, LinkDescriptor* ld );
+    bool handlePong( OverlayMsg* overlayMsg, LinkDescriptor* ld );
 …
         // internal message delivery -----------------------------------------------
+    // Convert OverlayMessage into new format and give it down to BaseCommunication
+    seqnum_t send_overlaymessage_down( OverlayMsg* message, const LinkID& bc_link, uint8_t priority );
         /// routes a message to its destination node
         void route( OverlayMsg* message );
+        void route( OverlayMsg* message, const NodeID& last_hop = NodeID::UNSPECIFIED );
         /// sends a raw message to another node, delivers it to the base overlay class
+        seqnum_t send( OverlayMsg* message, const NodeID& destination );
+        /// may throw "message_not_sent"-exception
+        seqnum_t send( OverlayMsg* message,
+                   const NodeID& destination,
+                   uint8_t priority,
+                const NodeID& last_hop = NodeID::UNSPECIFIED )
+    throw(message_not_sent);
         /// send a raw message using a link descriptor, delivers it to the base overlay class
+        seqnum_t send( OverlayMsg* message, LinkDescriptor* ld,
+                bool ignore_down = false );
+        seqnum_t send( OverlayMsg* message,
+                LinkDescriptor* ld,
+                uint8_t priority ) throw(message_not_sent);
         /// send a message using a node id using overlay routing
         /// sets necessary fields in the overlay message!
+        seqnum_t send_node( OverlayMsg* message, const NodeID& remote,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID);
+        /// may throw "message_not_sent"-exception
+        seqnum_t send_node( OverlayMsg* message, const NodeID& remote, uint8_t priority,
+                const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID) throw(message_not_sent);
         /// send a message using a node id using overlay routing using a link
         /// sets necessary fields in the overlay message!
+        seqnum_t send_link( OverlayMsg* message, const LinkID& link,
+                bool ignore_down = false );
+        void send_link( OverlayMsg* message,
+                const LinkID& link,
+                uint8_t priority ) throw(message_not_sent);
+        /// sends a notification to a sender from whom we just dropped a message
+        void report_lost_message( const OverlayMsg* message );
         // misc --------------------------------------------------------------------

source/ariba/overlay/CMakeLists.txt

-              r10700
+              r12060
     LinkDescriptor.h
     OverlayBootstrap.h
+    SequenceNumber.h
+    )
 …
     LinkDescriptor.cpp
     OverlayBootstrap.cpp
+    SequenceNumber.cpp
+    )

source/ariba/overlay/LinkDescriptor.h

-              r6961
+              r12060
 #include "ariba/communication/EndpointDescriptor.h"
 #include "ariba/CommunicationListener.h"
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
+#include <ariba/utility/misc/sha1.h>
+#include "ariba/overlay/SequenceNumber.h"
 namespace ariba {
 …
 class LinkDescriptor {
 public:
+    struct message_queue_entry
+    {
+        reboost::message_t message;
+        uint8_t priority;
+    };
         // ctor
         LinkDescriptor() {
+            time_t now = time(NULL);
                 // default values
                 this->up = false;
+//         this->closing = false;
+        this->failed = false;
                 this->fromRemote = false;
                 this->remoteNode = NodeID::UNSPECIFIED;
 …
                 this->communicationUp = false;
                 this->communicationId = LinkID::UNSPECIFIED;
                 this->keepAliveTime = time(NULL);
                 this->keepAliveMissed = 0;
+                this->keepAliveReceived = now;
+                this->keepAliveSent = now;
                 this->relaying     = false;
                 this->timeRelaying = time(NULL);
+                this->timeRelaying = now;
                 this->dropAfterRelaying = false;
                 this->service  = ServiceID::UNSPECIFIED;
 …
                 this->remoteLink = LinkID::UNSPECIFIED;
                 this->autolink = false;
                 this->lastuse = time(NULL);
+                this->lastuse = now;
                 this->retryCounter = 0;
+                this->hops = -1;
+        this->transmit_seqnums = false; // XXX
+        }
 …
         // general information about the link --------------------------------------
         bool up;           ///< flag whether this link is up and running
+//      bool closing;      ///< flag, whether this link is in the regular process of closing
+        bool failed;       ///< flag, whether communication is (assumed to be) not/no longer possible on this link
         bool fromRemote;   ///< flag, whether this link was requested from remote
         NodeID remoteNode; ///< remote end-point node
-        bool isVital() {
-                return up && keepAliveMissed == 0;
+        }
-        bool isDirectVital() {
-                return isVital() && communicationUp && !relayed;
+        }
 …
         bool   communicationUp;   ///< flag, whether the communication is up
+        // sequence numbers --------------------------------------------------------
+        SequenceNumber last_sent_seqnum;
+    bool transmit_seqnums;
         // direct link retries -----------------------------------------------------
         EndpointDescriptor endpoint;
 …
         // link alive information --------------------------------------------------
         time_t keepAliveTime; ///< the last time a keep-alive message was received
         int keepAliveMissed;  ///< the number of missed keep-alive messages
+        time_t keepAliveReceived; ///< the last time a keep-alive message was received
+        time_t keepAliveSent;  ///< the number of missed keep-alive messages
         void setAlive() {
                 keepAliveMissed = 0;
                 keepAliveTime = time(NULL);
+//              keepAliveSent = time(NULL);
+                keepAliveReceived = time(NULL);
+        }
 …
         LinkID remoteLink; ///< the remote link id
         vector<NodeID> routeRecord;
+        int hops;
         // relay state -------------------------------------------------------------
 …
         // auto links --------------------------------------------------------------
         bool autolink;  ///< flag, whether this link is a auto-link
         time_t lastuse; ///< time, when the link was last used
         deque<Message*> messageQueue; ///< waiting messages to be delivered
+        time_t lastuse; ///< time, when the link was last used XXX AUTO_LINK-ONLY
+        deque<message_queue_entry> messageQueue; ///< waiting messages to be delivered
         void setAutoUsed() {
                 if (autolink) lastuse = time(NULL);
 …
         /// drops waiting auto-link messages
         void flushQueue() {
+                BOOST_FOREACH( Message* msg, messageQueue )     delete msg;
+//              BOOST_FOREACH( Message* msg, messageQueue )     delete msg;  // XXX MARIO: shouldn't be necessary anymore, since we're using shared pointers
                 messageQueue.clear();
+        }
 …
         // string representation ---------------------------------------------------
         std::string to_string() const {
+            time_t now = time(NULL);
                 std::ostringstream s;
+        if ( relayed )
+            s << "[RELAYED-";
+        else
+            s << "[DIRECT-";
+        s << "LINK] ";
+        s << "id=" << overlayId.toString().substr(0,4) << " ";
+        s << "serv=" << service.toString() << " ";
                 s << "up=" << up << " ";
                 s << "init=" << !fromRemote << " ";
-                s << "id=" << overlayId.toString().substr(0,4) << " ";
-                s << "serv=" << service.toString() << " ";
                 s << "node=" << remoteNode.toString().substr(0,4) << " ";
                 s << "relaying=" << relaying << " ";
                 s << "miss=" << keepAliveMissed << " ";
+                s << "last_received=" << now - keepAliveReceived << "s ";
                 s << "auto=" << autolink << " ";
+                s << "hops=" << hops << " ";
                 if ( relayed ) {
                         s << "| Relayed: ";
 …
                 } else {
                         s << "| Direct: ";
                         s << "using id=" << communicationId.toString().substr(0,4) << " ";
+                        s << "using [COMMUNICATION-LINK] id=" << communicationId.toString().substr(0,4) << " ";
                         s << "(up=" << communicationUp << ") ";
+                }

source/ariba/overlay/messages/JoinReply.cpp

-              r3690
+              r12060
 vsznDefault(JoinReply);
+JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed, const EndpointDescriptor _bootstrapEp)
+        : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed ), bootstrapEp( _bootstrapEp ){
+JoinReply::JoinReply(const SpoVNetID _spovnetid, const OverlayParameterSet _param, bool _joinAllowed)
+        : spovnetid( _spovnetid ), param( _param ), joinAllowed( _joinAllowed )
+{
+}
 …
+}
 const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){
         return bootstrapEp;
+}
+//const EndpointDescriptor& JoinReply::getBootstrapEndpoint(){
+//      return bootstrapEp;
+//}
 }} // ariba::overlay

source/ariba/overlay/messages/JoinReply.h

-              r5870
+              r12060
 #define JOIN_REPLY_H__
+#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/SpoVNetID.h"
 #include "ariba/utility/types/NodeID.h"
 #include "ariba/utility/types/OverlayParameterSet.h"
 #include "ariba/communication/EndpointDescriptor.h"
+//#include "ariba/communication/EndpointDescriptor.h"
 using ariba::utility::OverlayParameterSet;
 …
 using ariba::utility::SpoVNetID;
 using ariba::utility::NodeID;
 using ariba::communication::EndpointDescriptor;
+//using ariba::communication::EndpointDescriptor;
 namespace ariba {
 …
         OverlayParameterSet param; //< overlay parameters
         bool joinAllowed; //< join successfull or access denied
         EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface
+//      EndpointDescriptor bootstrapEp; //< the endpoint for bootstrapping the overlay interface
 public:
 …
                 const SpoVNetID _spovnetid = SpoVNetID::UNSPECIFIED,
                 const OverlayParameterSet _param = OverlayParameterSet::DEFAULT,
                 bool _joinAllowed = false,
                 const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED()
+                bool _joinAllowed = false  /*,
+                const EndpointDescriptor _bootstrapEp = EndpointDescriptor::UNSPECIFIED()*/
         );
 …
         const OverlayParameterSet& getParam();
         bool getJoinAllowed();
         const EndpointDescriptor& getBootstrapEndpoint();
+//      const EndpointDescriptor& getBootstrapEndpoint();
 };
 …
 sznBeginDefault( ariba::overlay::JoinReply, X ) {
         uint8_t ja = joinAllowed;
+        X && &spovnetid && param && bootstrapEp && ja;
+        X && &spovnetid && param;
+//      X && bootstrapEp;
+        X && ja;
         if (X.isDeserializer()) joinAllowed = ja;
 } sznEnd();

source/ariba/overlay/messages/OverlayMsg.h

-              r10653
+              r12060
 #include "ariba/utility/types/NodeID.h"
 #include "ariba/utility/types/LinkID.h"
+#include "ariba/communication/EndpointDescriptor.h"
+// #include <ariba/utility/misc/sha1.h>
+#include "ariba/overlay/SequenceNumber.h"
 namespace ariba {
 …
 using ariba::utility::ServiceID;
 using ariba::utility::Message;
 using ariba::communication::EndpointDescriptor;
+//using ariba::communication::EndpointDescriptor;
 using_serialization;
 …
  * between nodes.
+ *
  * @author Sebastian Mies <mies@tm.uka.de>
+ * @author Sebastian Mies <mies@tm.uka.de>, Mario Hock
  */
 class OverlayMsg: public Message { VSERIALIZEABLE;
 …
                 maskTransfer    = 0x10, ///< bit mask for transfer messages
                 typeData        = 0x11, ///< message contains data for higher layers
+                typeMessageLost = 0x12, ///< message contains info about a dropped message
                 // join signaling
 …
                 typeLinkUpdate  = 0x33, ///< update message for link association
                 typeLinkDirect  = 0x34, ///< direct connection has been established
+                typeLinkAlive   = 0x35, ///< keep-alive message
+                typeKeepAlive   = 0x35, ///< keep-alive message
+                typeKeepAliveReply   = 0x36, ///< keep-alive message (replay)
+                typeLinkClose   = 0x37,
                 /// DHT routed messages
 …
                 maskDHTResponse = 0x50, ///< bit mask for dht responses
                 typeDHTData     = 0x51, ///< DHT get data
+        /// misc message types
+        typePing        = 0x44,
+        typePong        = 0x45,
                 // topology signaling
 …
                 typeSignalingEnd = 0xFF    ///< end of the signaling types
         };
+    /// message flags (uint8_t)
+    enum flags_
+    {
+        flagRelayed         = 1 << 0,
+        flagRegisterRelay   = 1 << 1,
+        flagRouteRecord     = 1 << 2,
+        flagSeqNum1         = 1 << 3,
+        flagSeqNum2         = 1 << 4,
+        flagAutoLink        = 1 << 5,
+        flagLinkMessage     = 1 << 6,
+        flagHasMoreFlags    = 1 << 7
+    };
         /// default constructor
 …
                 const LinkID& _sourceLink      = LinkID::UNSPECIFIED,
                 const LinkID& _destinationLink = LinkID::UNSPECIFIED )
         :       type(type), flags(0), hops(0), ttl(10),
+    :   type(type), flags(0), extended_flags(0), hops(0), ttl(10), priority(0),
                 service(_service),
                 sourceNode(_sourceNode), destinationNode(_destinationNode),
 …
         // copy constructor
         OverlayMsg(const OverlayMsg& rhs)
+        :       type(rhs.type), flags(rhs.flags), hops(rhs.hops), ttl(rhs.ttl),
+                service(rhs.service),
+    :   type(rhs.type), flags(rhs.flags), extended_flags(rhs.extended_flags),
+        hops(rhs.hops), ttl(rhs.ttl),
+                priority(rhs.priority), service(rhs.service),
                 sourceNode(rhs.sourceNode), destinationNode(rhs.destinationNode),
                 sourceLink(rhs.sourceLink), destinationLink(rhs.destinationLink),
 …
+        }
+        /// priority ------------------------------------------------------------------
+        uint8_t getPriority() const {
+            return priority;
+        }
+        void setPriority(uint8_t priority) {
+            this->priority = priority;
+        }
         /// flags ------------------------------------------------------------------
         bool isRelayed() const {
                 return (flags & 0x01)!=0;
+        return (flags & flagRelayed)!=0;
+        }
         void setRelayed( bool relayed = true ) {
                 if (relayed) flags |= 1; else flags &= ~1;
+        if (relayed) flags |= flagRelayed; else flags &= ~flagRelayed;
+        }
         bool isRegisterRelay() const {
                 return (flags & 0x02)!=0;
+                return (flags & flagRegisterRelay)!=0;
+        }
         void setRegisterRelay( bool relayed = true ) {
                 if (relayed) flags |= 0x02; else flags &= ~0x02;
+        if (relayed) flags |= flagRegisterRelay; else flags &= ~flagRegisterRelay;
+        }
         bool isRouteRecord() const {
                 return (flags & 0x04)!=0;
+                return (flags & flagRouteRecord)!=0;
+        }
         void setRouteRecord( bool route_record = true ) {
                 if (route_record) flags |= 0x04; else flags &= ~0x04;
+        if (route_record) flags |= flagRouteRecord; else flags &= ~flagRouteRecord;
+        }
         bool isAutoLink() const {
                 return (flags & 0x80) == 0x80;
+        return (flags & flagAutoLink) == flagAutoLink;
+        }
         void setAutoLink(bool auto_link = true ) {
                 if (auto_link) flags |= 0x80; else flags &= ~0x80;
+        if (auto_link) flags |= flagAutoLink; else flags &= ~flagAutoLink;
+        }
         bool isLinkMessage() const {
                 return (flags & 0x40)!=0;
+                return (flags & flagLinkMessage)!=0;
+        }
         void setLinkMessage(bool link_info = true ) {
+                if (link_info) flags |= 0x40; else flags &= ~0x40;
+        }
+        bool containsSourceEndpoint() const {
+                return (flags & 0x20)!=0;
+        }
+        void setContainsSourceEndpoint(bool contains_endpoint) {
+                if (contains_endpoint) flags |= 0x20; else flags &= ~0x20;
+        }
+        if (link_info) flags |= flagLinkMessage; else flags &= ~flagLinkMessage;
+        }
+        bool hasExtendedFlags() const {
+        return (flags & flagHasMoreFlags) == flagHasMoreFlags;
+    }
         /// number of hops and time to live ----------------------------------------
 …
                 this->destinationLink = link;
                 setLinkMessage();
+        }
-        void setSourceEndpoint( const EndpointDescriptor& endpoint ) {
-                sourceEndpoint = endpoint;
-                setContainsSourceEndpoint(true);
+        }
-        const EndpointDescriptor& getSourceEndpoint() const {
-                return sourceEndpoint;
+        }
 …
                 destinationLink = dummyLink;
                 hops = 0;
+                routeRecord.clear();
+        }
 …
                         routeRecord.push_back(node);
+        }
+        /// sequence numbers
+        bool hasShortSeqNum() const
+    {
+        return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum1;
+    }
+    bool hasLongSeqNum() const
+    {
+        return (flags & (flagSeqNum1 | flagSeqNum2)) == flagSeqNum2;
+    }
+    void setSeqNum(const SequenceNumber& sequence_number)
+    {
+        this->seqnum = sequence_number;
+        // short seqnum
+        if ( sequence_number.isShortSeqNum() )
+        {
+            flags |= flagSeqNum1;
+            flags &= ~flagSeqNum2;
+        }
+        // longseqnum
+        else if ( sequence_number.isShortSeqNum() )
+        {
+            flags &= ~flagSeqNum1;
+            flags |= flagSeqNum2;
+        }
+        // no seqnum
+        else
+        {
+            flags &= ~flagSeqNum1;
+            flags &= ~flagSeqNum2;
+        }
+    }
+    const SequenceNumber& getSeqNum() const
+    {
+        return seqnum;
+    }
 private:
         uint8_t type, flags, hops, ttl;
+        uint8_t type, flags, extended_flags, hops, ttl, priority;
         ServiceID service;
         NodeID sourceNode;
 …
         LinkID sourceLink;
         LinkID destinationLink;
         EndpointDescriptor sourceEndpoint;
+//      EndpointDescriptor sourceEndpoint;
         vector<NodeID> routeRecord;
+    SequenceNumber seqnum;
 };
 …
 sznBeginDefault( ariba::overlay::OverlayMsg, X ){
         // header
+        X && type && flags && hops && ttl;
+        X && type && flags;
+    if ( hasExtendedFlags() )
+        X && extended_flags;
+    X && hops && ttl;
         // addresses
         X && &service && &sourceNode && &destinationNode;
+        // priority
+        X && priority;
         // message is associated with a end-to-end link
 …
                 X && &sourceLink && &destinationLink;
+        // message is associated with a source end-point
+        if (containsSourceEndpoint())
+                X && sourceEndpoint;
+    /* seqnum */
+    // serialize
+    if ( X.isSerializer() )
+    {
+        if ( hasShortSeqNum() )
+        {
+            uint32_t short_seqnum;
+            short_seqnum = seqnum.getShortSeqNum();
+            X && short_seqnum;
+        }
+        if ( hasLongSeqNum() )
+        {
+            uint64_t long_seqnum;
+            long_seqnum = seqnum.getLongSeqNum();
+            X && long_seqnum;
+        }
+    }
+    // deserialize
+    else
+    {
+        if ( hasShortSeqNum() )
+        {
+            uint32_t short_seqnum;
+            X && short_seqnum;
+            seqnum = ariba::overlay::SequenceNumber(short_seqnum);
+        }
+        if ( hasLongSeqNum() )
+        {
+            uint64_t long_seqnum;
+            X && long_seqnum;
+            seqnum = ariba::overlay::SequenceNumber(long_seqnum);
+        }
+    }
         // message should record its route
         if (isRouteRecord()) {
 …
         // payload
         X && Payload();
+//      X && Payload();
 } sznEnd();

source/ariba/overlay/modules/OverlayFactory.cpp

-              r3718
+              r12060
 // structured overlays
 #include "chord/Chord.h"
+#include "onehop/OneHop.h"
+//#include "onehop/OneHop.h"  //DEPRECATED
 namespace ariba {
 …
                         return new Chord( baseoverlay, nodeid, routeReceiver, param );
                 case OverlayParameterSet::OverlayStructureOneHop:
                         return new OneHop( baseoverlay, nodeid, routeReceiver, param );
+//              case OverlayParameterSet::OverlayStructureOneHop:
+//                      return new OneHop( baseoverlay, nodeid, routeReceiver, param );
                 default:
+                    // NEVER return "NULL"
+                    assert(false);
+                    throw 42;
                         return NULL;
+        }
+        // NEVER return "NULL"
+        assert(false);
+        throw 42;
         return NULL;
+}

source/ariba/overlay/modules/OverlayInterface.cpp

r6854	r12060
68	68
69	69	void OverlayInterface::onLinkFail(const LinkID& lnk, const NodeID& remote) {
	70	onLinkDown(lnk, remote);
70	71	}
71	72
…	…
79	80	}
80	81
81		void OverlayInterface::onMessage(const DataMessage& msg, const NodeID& remote,
	82	void OverlayInterface::onMessage(OverlayMsg* msg,
	83	reboost::shared_buffer_t sub_msg,
	84	const NodeID& remote,
82	85	const LinkID& lnk) {
83	86	}

source/ariba/overlay/modules/OverlayInterface.h

-              r10573
+              r12060
          */
         virtual const LinkID& getNextLinkId( const NodeID& id ) const = 0;
+    /**
+     * Returns link ids of possible next hops a route message could take,
+     * sorted by "quality" (e.g. overlay-distance).
+     *
+     * The Â»numÂ« parameter can be used to specify the desired number of elements
+     * in the returned vector. This is intendet for optimizations. The
+     * implementation may choose to return a different number of elements than
+     * requested.
+     *
+     * NOTE: The returned vector may contain Â»unspecifiedÂ« links. These refer to
+     * to the own node. (e.g. If there's no closer node, the top element in the
+     * returned vector is unsoecified.)
+     *
+     * @param id The destination node id
+     * @param num The desired number of elements in the returned vector.
+     *            (0 means Â»not specified/max)Â«
+     * @return A sorted vector of link ids to possible next hops.
+     */
+    virtual std::vector<const LinkID*> getSortedLinkIdsTowardsNode(
+        const NodeID& id, int num = 0 ) const = 0;
         /**
          * Returns the NodeID of the next hop a route message would take.
 …
         /// @see CommunicationListener
+        virtual void onMessage(const DataMessage& msg, const NodeID& remote,
+        virtual void onMessage(OverlayMsg* msg,
+                reboost::shared_buffer_t sub_msg,
+                const NodeID& remote,
                         const LinkID& lnk = LinkID::UNSPECIFIED);

source/ariba/overlay/modules/OverlayStructureEvents.h

-              r5151
+              r12060
 #include "ariba/utility/types/NodeID.h"
+#include "ariba/utility/messages.h"
+#include "ariba/utility/types/LinkID.h"
+#include "ariba/utility/messages/Message.h"
 using ariba::utility::NodeID;
+using ariba::utility::LinkID;
 using ariba::utility::Message;
 …
 class OverlayInterface;
 class OneHop;
+//class OneHop;
 class OverlayStructureEvents {
         friend class ariba::overlay::OverlayInterface;
         friend class ariba::overlay::OneHop;
+//      friend class ariba::overlay::OneHop;
 public:

source/ariba/overlay/modules/chord/Chord.cpp

-              r10572
+              r12060
 #include "detail/chord_routing_table.hpp"
+#include "messages/Discovery.h"
+//#include "messages/Discovery.h"  // XXX DEPRECATED
 namespace ariba {
 …
 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,
 …
 /// 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());
+    }
+}
 …
         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);
+        }
 …
                 // 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);
+        }
 …
         // timer for stabilization management
+        Timer::setInterval(1000);
+//      Timer::setInterval(1000);  // TODO find an appropriate interval!
+        Timer::setInterval(10000);  // XXX testing...
         Timer::start();
+}
 …
         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
 …
         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;
 …
 /// @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() );
 …
 /// @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
                 );
 …
                 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;
+        }
+        }
+}

source/ariba/overlay/modules/chord/Chord.h

-              r10572
+              r12060
 #include "../OverlayInterface.h"
 #include <vector>
+#include <stdexcept>
 class chord_routing_table;
 …
                 const NodeID& node = NodeID::UNSPECIFIED );
         // helper: sends a message using the "base overlay"
         seqnum_t send( OverlayMsg* msg, const LinkID& link );
+        // helper: sends a message over a link using the "base overlay"
+        void send( OverlayMsg* msg, const LinkID& link );
+        // helper: sends a message to a node using the "base overlay"
+        void send_node( OverlayMsg* message, const NodeID& remote );
         // stabilization: sends a discovery message to the specified neighborhood
         void send_discovery_to( const NodeID& destination, int ttl = 3 );
 …
         virtual const LinkID& getNextLinkId( const NodeID& id ) const;
+    /// @see OverlayInterface.h
+    /// NOTE: This implementation excepts num == 2
+    virtual std::vector<const LinkID*> getSortedLinkIdsTowardsNode(
+        const NodeID& id, int num = 0 ) const;
         /// @see OverlayInterface.h
         virtual const NodeID& getNextNodeId( const NodeID& id ) const;
 …
         /// @see CommunicationListener.h or @see OverlayInterface.h
+        virtual void onMessage(const DataMessage& msg, const NodeID& remote,
+        virtual void onMessage(OverlayMsg* msg,
+                reboost::shared_buffer_t sub_msg,
+                const NodeID& remote,
                         const LinkID& lnk = LinkID::UNSPECIFIED);

source/ariba/overlay/modules/chord/detail/chord_routing_table.hpp

-              r8606
+              r12060
         // the own node id
         nodeid_t id;
+    // the own node id as routing item
+    item own_id_item;
         // successor and predecessor tables
 …
         /// constructs the reactive chord routing table
         explicit chord_routing_table( const nodeid_t& id, int redundancy = 4 ) :
+                id(id), succ( redundancy, succ_compare_type(this->id), *this ),
+                pred( redundancy, pred_compare_type(this->id), *this ) {
+                id(id),
+                succ( redundancy, succ_compare_type(this->id), *this ),
+                pred( redundancy, pred_compare_type(this->id), *this )
+    {
+        // init reflexive item
+        own_id_item.id = id;
+        own_id_item.ref_count = 1;
                 // create finger tables
 …
+                        }
+                }
                 if (best_item != NULL && distance(value, id)<distance(value, best_item->id))
                         return NULL;
                 return best_item;
+        }
+        std::vector<const item*> get_next_2_hops( const nodeid_t& value)
+    {
+        ring_distance distance;
+        item* best_item = &own_id_item;
+        item* second_best_item = NULL;
+        // find best and second best item
+        for (size_t i=0; i<table.size(); i++)
+        {
+            item* curr = &table[i];
+            // not not include orphans into routing!
+            if (curr->ref_count==0) continue;
+            // check if we found a better item
+            // is best item
+            if ( distance(value, curr->id) < distance(value, best_item->id) )
+            {
+                second_best_item = best_item;
+                best_item = curr;
+            }
+            // is second best item
+            else
+            {
+                if ( second_best_item == NULL )
+                {
+                    second_best_item = curr;
+                    continue;
+                }
+                if ( distance(value, curr->id) < distance(value, second_best_item->id) )
+                {
+                    second_best_item = curr;
+                }
+            }
+        }
+        // prepare return vector
+        std::vector<const item*> ret;
+        if ( best_item != NULL )
+        {
+            ret.push_back(best_item);
+        }
+        if ( second_best_item != NULL )
+        {
+            ret.push_back(second_best_item);
+        }
+        return ret;
+    }
         const nodeid_t* get_successor() {
                 if (succ.size()==NULL) return NULL;
+                if (succ.size()==0) return NULL;
                 return &succ.front();
+        }
         const nodeid_t* get_predesessor() {
                 if (pred.size()==NULL) return NULL;
+                if (pred.size()==0) return NULL;
                 return &pred.front();
+        }

source/ariba/overlay/modules/chord/messages/CMakeLists.txt

-              r10700
+              r12060
 # [License]
 add_headers(Discovery.h)
+#add_headers(Discovery.h)
 add_sources(Discovery.cpp)
+#add_sources(Discovery.cpp)

source/ariba/overlay/modules/chord/messages/Discovery.h

r5870	r12060
59	59	using_serialization;
60	60
	61	// XXX This whole message is DEPRECATED
61	62	class Discovery : public Message {
62	63	VSERIALIZEABLE;

source/ariba/overlay/modules/onehop/CMakeLists.txt

-              r10700
+              r12060
 # [License]
-add_headers(OneHop.h)
+add_sources(OneHop.cpp)
+## DEPRECATED
+#add_headers(OneHop.h)
+add_subdir_sources(messages)
+#add_sources(OneHop.cpp)
+#add_subdir_sources(messages)

source/ariba/utility/CMakeLists.txt

-              r10700
+              r12060
 add_subdir_sources(
     addressing
+    addressing2
     bootstrap
     configuration
     internal
     logging
-    measurement
     messages
     misc

source/ariba/utility/addressing

Property svn:mergeinfo changed (with no actual effect on merging)

source/ariba/utility/addressing/endpoint_set.hpp

r10653	r12060
1		#ifndef ENDPOINT_SET_HPP_
2		#define ENDPOINT_SET_HPP_
	1	#ifndef ENDPOINT_SET_HPP_DEPRECATED_
	2	#define ENDPOINT_SET_HPP_DEPRECATED_
3	3
4	4	#include "addressing.hpp"

source/ariba/utility/addressing/facades/address_v.hpp

-              r10789
+              r12060
 #include "../detail/address_convenience.hpp"
+#include <boost/shared_ptr.hpp>
 namespace ariba {
 …
 class address_v: public detail::address_convenience<address_v> {
 public:
+    typedef boost::shared_ptr<address_v> shared_ptr;
         virtual ~address_v() {}

source/ariba/utility/addressing/ip_address.hpp

r6919	r12060
1		#ifndef IP_ADDRESS_HPP_
2		#define IP_ADDRESS_HPP_
	1	#ifndef IP_ADDRESS_HPP_DEPRECATED_
	2	#define IP_ADDRESS_HPP_DEPRECATED_
3	3
4	4	#include <string>

source/ariba/utility/addressing/tcpip_endpoint.hpp

r5284	r12060
1		#ifndef TCPIP_ENDPOINT_HPP_
2		#define TCPIP_ENDPOINT_HPP_
	1	#ifndef TCPIP_ENDPOINT_HPP_DEPRECATED_
	2	#define TCPIP_ENDPOINT_HPP_DEPRECATED_
3	3
4	4	#include<string>

source/ariba/utility/bootstrap/modules/bluetoothsdp

Property svn:mergeinfo changed (with no actual effect on merging)

source/ariba/utility/bootstrap/modules/periodicbroadcast

Property svn:mergeinfo changed (with no actual effect on merging)

source/ariba/utility/bootstrap/modules/periodicbroadcast/PeriodicBroadcast.h

-              r10653
+              r12060
 #define __PERIODIC_BROADCAST_H
+// ariba
 #include "ariba/config.h"
+#include "ariba/utility/bootstrap/modules/BootstrapModule.h"
+#include "ariba/utility/logging/Logging.h"
+#include "ariba/utility/system/Timer.h"
+// ariba messages
+#include "PeriodicBroadcastMessage.h"
+// (ariba) link-local
+#include "ariba/utility/transport/StreamTransport/StreamTransport.hpp"
+// system
 #include <map>
 #include <string>
 #include <ctime>
 #include <iostream>
+// boost
 #include <boost/asio.hpp>
 #include <boost/foreach.hpp>
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/thread.hpp>
+#include "ariba/utility/bootstrap/modules/BootstrapModule.h"
+#include "ariba/utility/logging/Logging.h"
+#include "ariba/utility/system/Timer.h"
+#include "PeriodicBroadcastMessage.h"
+//link-local
+#include "ariba/utility/transport/tcpip/tcpip.hpp"
 using std::map;
 …
                                 // include all link-local interfaces
                                 vector<uint64_t> scope_ids = ariba::transport::tcpip::get_interface_scope_ids();
+                                vector<uint64_t> scope_ids = ariba::transport::get_interface_scope_ids();
                                 BOOST_FOREACH ( uint64_t id, scope_ids )

source/ariba/utility/logging/Logging.h

r10700	r12060
97	97
98	98	static int __loglevel__ = 2; //default is info
	99	// static int __loglevel__ = 1; // XXX use higher log level
99	100
100	101	#define logging_trace(x) { logging_stdout(x); }

source/ariba/utility/messages.h

-              r3690
+              r12060
 #define MESSAGES_H_
+// TODO wÃŒrde sagen das brauchen wir nicht mehr
+//   ---> ÃŒberall dieses include rausnehmen und ggf  #include "messages/Message.h"   einfÃŒgen..
 #include "messages/Message.h"
+#include "messages/MessageSender.h"
 #include "messages/MessageReceiver.h"
 #include "messages/MessageUtilities.h"
+#include "messages/MessageProvider.h"
 #include "messages/TextMessage.h"
+//#include "messages/MessageSender.h"  // TODO wird das noch genutzt..? Wenn nein, sollte es weg!
+//#include "messages/MessageReceiver.h"
+//#include "messages/MessageUtilities.h"
+////#include "messages/MessageProvider.h"  // DEPRECATED
+//#include "messages/TextMessage.h"
 #endif /* MESSAGES_H_ */

source/ariba/utility/messages/Message.cpp

-              r8620
+              r12060
 #include "ariba/utility/serialization/DataStream.hpp"
+#include "ariba/utility/logging/Logging.h"
 NAMESPACE_BEGIN
 …
+}
+reboost::message_t Message::wrap_up_for_sending()
+{
+    assert( ! wrapped_up );
+    wrapped_up = true;
+    //// Adapt to new message system ////
+    Data data = data_serialize(this, DEFAULT_V);
+    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
+    newstyle_payload.push_front(buf);
+    return newstyle_payload;
+}
+reboost::shared_buffer_t Message::serialize_into_shared_buffer()
+{
+    assert ( newstyle_payload.length() == 0 );
+    //// Adapt to new message system ////
+    Data data = data_serialize(this, DEFAULT_V);
+    reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
+    return buf;
+}
+reboost::shared_buffer_t Message::deserialize_from_shared_buffer(reboost::shared_buffer_t buff)
+{
+    // NOTE: legacy payload is not allowed when using shared buffers
+    this->legacy_payload_disabled = true;
+    assert( buff.size() > 0 );
+    // const_cast is necessary here, but without legacy payload we should be save here (more or less)
+    Data dat(const_cast<uint8_t*>(buff.data()), buff.size() * 8);
+    size_t len = this->SERIALIZATION_METHOD_NAME(DESERIALIZE, dat) / 8;
+    // return remaining sub-buffer
+    return buff(len);
+}
 NAMESPACE_END
 std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg ) {
         using_serialization;
         stream << "msg(type=" << typeid(msg).name();
+        stream << "msg(type=" << typeid(msg).name() << ",";
         stream << "len=" << (data_length(&msg)/8) << ",";
         Data data = data_serialize(&msg);
         stream << ",data=" << data;
+        stream << "data=" << data;
         data.release();
         stream << ")";
         return stream;
+}

source/ariba/utility/messages/Message.h

-              r6919
+              r12060
 #include "ariba/utility/serialization.h"
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
 std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg );
 …
         friend std::ostream& ::operator<<(std::ostream& stream, const ariba::utility::Message& msg );
-        // root binary data
-        shared_array<uint8_t> root;
         // payload
+        bool legacy_payload_disabled;
         bool releasePayload;
         Data payload; //< messages binary data
+        // XXX testing...
+        reboost::message_t newstyle_payload;
+        bool wrapped_up;
         // addresses and control info
 …
          */
         inline Message() :
+                root(), releasePayload(true), payload(), srcAddr(NULL),destAddr(NULL) {
+            legacy_payload_disabled(false), releasePayload(true), payload(),
+            newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) {
+        }
 …
          */
         explicit inline Message( const Data& data ) :
+                releasePayload(true), srcAddr(NULL),destAddr(NULL) {
+        legacy_payload_disabled(false), releasePayload(true),
+        newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) {  // FIXME newstyle_payload..?
                 this->payload = data.clone();
 //              this->root = shared_array<uint8_t>((uint8_t*)data.getBuffer());
 …
                 return decapsulate<T>();
+        }
+        // XXX testing
+        void set_payload_message(reboost::message_t msg)
+        {
+            newstyle_payload = msg;
+        }
+        void append_buffer(reboost::shared_buffer_t buff)
+        {
+            newstyle_payload.push_back(buff);
+        }
+        // XXX testing... packs this message into the payload message (do not use twice!!)
+        virtual reboost::message_t wrap_up_for_sending();
+        /**
+         * Uses the old serialization system to serialize itself into a (new style) shared buffer.
+         */
+        virtual reboost::shared_buffer_t serialize_into_shared_buffer();
+        /*
+         * XXX experimental
+         *
+         * Uses the old serialization system to deserialize itself out of a (new style) shared buffer.
+         * @return remaining sub-buffer (the "payload")
+         *
+         * Note: This is some kind of a hack! handle with care.
+         */
+        virtual reboost::shared_buffer_t deserialize_from_shared_buffer(reboost::shared_buffer_t buff);
 protected:
 …
          * @return A explicit payload serializer
          */
+        finline PayloadSerializer Payload( size_t length = ~0 ) {
+        finline PayloadSerializer Payload( size_t length = ~0 )
+        {
+//          assert( ! legacy_payload_disabled );  // FIXME aktuell
                 return PayloadSerializer( this, length );
+        }

source/ariba/utility/messages/MessageProvider.cpp

-              r6919
+              r12060
 // [License]
 #include "MessageProvider.h"
 NAMESPACE_BEGIN
 MessageProvider::MessageProvider() {
+}
 MessageProvider::~MessageProvider() {
+}
 bool MessageProvider::sendMessageToReceivers( const Message* message ) {
         bool sent =  false;
         for (size_t i=0; i<receivers.size(); i++)
                 if (receivers[i]->receiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true;
         return sent;
+}
 void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) {
         receivers.push_back(receiver);
+}
 void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) {
         for (size_t i=0; i<receivers.size(); i++)
                 if (receivers[i]==receiver) {
                         receivers.erase( receivers.begin()+i );
                         break;
+                }
+}
 NAMESPACE_END
+//#include "MessageProvider.h"
+//
+//NAMESPACE_BEGIN
+//
+//MessageProvider::MessageProvider() {
+//}
+//
+//MessageProvider::~MessageProvider() {
+//}
+//
+//bool MessageProvider::sendMessageToReceivers( const Message* message ) {
+//      bool sent =  false;
+//      for (size_t i=0; i<receivers.size(); i++)
+//              if (receivers[i]->receiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true;
+//      return sent;
+//}
+//
+//void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) {
+//      receivers.push_back(receiver);
+//}
+//
+//void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) {
+//      for (size_t i=0; i<receivers.size(); i++)
+//              if (receivers[i]==receiver) {
+//                      receivers.erase( receivers.begin()+i );
+//                      break;
+//              }
+//}
+//
+//NAMESPACE_END

source/ariba/utility/messages/MessageProvider.h

-              r3690
+              r12060
+// XXX DEPRECATED
 // [License]
 // The Ariba-Underlay Copyright
 …
 // [License]
+// XXX DEPRECATED
 #ifndef MESSAGEPROVIDER_H_
 #define MESSAGEPROVIDER_H_
 #include "_namespace.h"
 #include "MessageReceiver.h"
 #include "ariba/utility/types/LinkID.h"
 #include "ariba/utility/types/NodeID.h"
 #include <vector>
 using std::vector;
 using ariba::utility::LinkID;
 using ariba::utility::NodeID;
 NAMESPACE_BEGIN
 /**
  * This class defines an interface for message providers.
  * Implementing classes must allow receivers to register themselves.
+ *
  * @author Sebastian Mies
  */
 class MessageProvider {
 private:
         vector<MessageReceiver*> receivers;
 protected:
         bool sendMessageToReceivers( const Message* message );
 public:
         /**
          * Constructor.
          */
         MessageProvider();
         /**
          * Destructor.
          */
         ~MessageProvider();
         /**
          * Adds a message receiver.
+         *
          * @param receiver The receiver.
          */
         void addMessageReceiver( MessageReceiver* receiver );
         /**
          * Removes a message receiver.
+         *
          * @param receiver The receiver.
          */
         void removeMessageReceiver( MessageReceiver* receiver );
 };
 NAMESPACE_END
+//#include "_namespace.h"
+//#include "MessageReceiver.h"
+//#include "ariba/utility/types/LinkID.h"
+//#include "ariba/utility/types/NodeID.h"
+//#include <vector>
+//
+//using std::vector;
+//using ariba::utility::LinkID;
+//using ariba::utility::NodeID;
+//
+//NAMESPACE_BEGIN
+//
+//
+///**
+// * This class defines an interface for message providers.
+// * Implementing classes must allow receivers to register themselves.
+// *
+// * @author Sebastian Mies
+// */
+//class MessageProvider {
+//private:
+//      vector<MessageReceiver*> receivers;
+//
+//protected:
+//      bool sendMessageToReceivers( const Message* message );
+//
+//public:
+//      /**
+//       * Constructor.
+//       */
+//      MessageProvider();
+//
+//      /**
+//       * Destructor.
+//       */
+//      ~MessageProvider();
+//
+//      /**
+//       * Adds a message receiver.
+//       *
+//       * @param receiver The receiver.
+//       */
+//      void addMessageReceiver( MessageReceiver* receiver );
+//
+//      /**
+//       * Removes a message receiver.
+//       *
+//       * @param receiver The receiver.
+//       */
+//      void removeMessageReceiver( MessageReceiver* receiver );
+//};
+//
+//NAMESPACE_END
 #endif /* MESSAGEPROVIDER_H_ */

source/ariba/utility/messages/MessageReceiver.cpp

-              r3690
+              r12060
+}
 bool MessageReceiver::receiveMessage( const Message* message, const LinkID& link, const NodeID& node ) {
         //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl;
         return false;
+}
+//bool MessageReceiver::receiveMessage( reboost::shared_buffer_t message, const LinkID& link, const NodeID& node ) {
+//      //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl;
+//      return false;
+//}
 NAMESPACE_END

source/ariba/utility/messages/MessageReceiver.h

-              r3690
+              r12060
 #define MESSAGERECEIVER_H__
+#include "ariba/utility/messages/Message.h"
+//#include "ariba/utility/messages/Message.h"
+// reboost messages
+#include "ariba/utility/transport/messages/message.hpp"
 #include "ariba/utility/types/LinkID.h"
 #include "ariba/utility/types/NodeID.h"
 …
          * @return True, when the message has been accepted.
          */
+        virtual bool receiveMessage( const Message* message, const LinkID& link, const NodeID& node );
+        virtual bool receiveMessage( reboost::shared_buffer_t message,
+                const LinkID& link,
+                const NodeID& node,
+                bool bypass_overlay ) = 0;
 };

source/ariba/utility/misc/Demultiplexer.hpp

r3705	r12060
136	136
137	137	SERVICE_LISTENER_MAP_CITERATOR it = mapServiceListener.find( id );
138		if( it == mapServiceListener.end() ) return ~~NULL~~;
	138	if( it == mapServiceListener.end() ) return 0;
139	139	else return it->second;
140	140	}

source/ariba/utility/system/StartupWrapper.cpp

r10700	r12060
36	36	// Telematics.
37	37	// [License]
	38
	39	// XXX NOTE: Use this class with caution! Config support is outdated.
38	40
39	41	#include "StartupWrapper.h"

source/ariba/utility/system/StartupWrapper.h

r4483	r12060
36	36	// Telematics.
37	37	// [License]
	38
	39	// XXX NOTE: Use this class with caution! Config support is outdated.
38	40
39	41	#ifndef __STARTUP_WRAPPER_H

source/ariba/utility/system/SystemQueue.cpp

-              r7533
+              r12060
+}
+// maps to function call internally to the Event-system
+void SystemQueue::scheduleCall( const boost::function0<void>& function, uint32_t delay)
+{
+    // copy function object
+    boost::function0<void>* function_ptr = new boost::function0<void>();
+    (*function_ptr) = function;
+    // schedule special call-event
+    scheduleEvent( SystemEvent(&internal_function_caller, SystemEventType::DEFAULT, function_ptr), delay );
+}
 #ifdef UNDERLAY_OMNET
 void SystemQueue::handleMessage(cMessage* msg){

source/ariba/utility/system/SystemQueue.h

-              r7468
+              r12060
 #endif
+#include <boost/function.hpp>
 using std::vector;
 using boost::posix_time::ptime;
 …
          */
         void scheduleEvent( const SystemEvent& event, uint32_t delay = 0 );
+        /**
+         * This method schedules a function call in the SystemQueue.
+         * (Like scheduleEvent, but to be used with boost::bind.)
+         *
+         * @param function: The function to be called [void function()]
+         * @param The delay in milli-seconds
+         */
+        void scheduleCall( const boost::function0<void>& function, uint32_t delay = 0 );
         /**
 …
 #endif
 private:
 …
         volatile bool systemQueueRunning;
 #endif
+private:
+    /**
+     * This inner class handles the function-call events.
+     * @see SystemQueue::scheduleCall
+     */
+    class FunctionCaller  :  public SystemEventListener
+    {
+        void handleSystemEvent(const SystemEvent& event)
+        {
+            boost::function0<void>* function_ptr = event.getData< boost::function0<void> >();
+            (*function_ptr)();
+            delete function_ptr;
+        }
+    };
+    FunctionCaller internal_function_caller;
 }; // class SystemQueue

source/ariba/utility/transport

Property svn:mergeinfo changed (with no actual effect on merging)

source/ariba/utility/transport/CMakeLists.txt

-              r10700
+              r12060
 add_headers(
-    test_transport.hpp
-    transport_connection.hpp
-    transport.hpp
-    transport_listener.hpp
     transport_peer.cpp
     transport_peer.hpp
-    transport_protocol.hpp
+    )
+add_subdir_sources(asio messages rfcomm tcpip)
+add_subdir_sources(asio messages rfcomm StreamTransport)

source/ariba/utility/transport/messages/message.cpp

r10653	r12060
24	24	os << "message({size=" << m.size() << ",buffers=" << (int) m.length()
25	25	<< ",hash=" << m.hash() << "},";
26		m.foreach(ts);
	26	m.msg_foreach(ts);
27	27	os << ")";
28	28	return os;

source/ariba/utility/transport/messages/message.hpp

-              r10653
+              r12060
 /// message size type
+typedef signed char mlength_t;
+//typedef signed char mlength_t;  // <--- don't do this!!
+//typedef size_t mlength_t;
+typedef int mlength_t;  // signed int seems necessary
 /// maximum number of buffers per message (default is 8)
 const mlength_t message_max_buffers = (1L << 3);
+//const mlength_t message_max_buffers = (1L << 4);
 //! A Copy-on-Write Message with Shared Buffers.
 …
         /// Copy message
         inline message_t(const message_t& msg) :
+                imsg(msg.imsg) {
+                imsg->owner = NULL;
+                imsg(msg.imsg)
+        {
+            if ( imsg )
+                imsg->owner = NULL;
+        }
 …
         /// Returns the number of buffers inside this message.
         inline mlength_t length() const {
+            if ( ! imsg )
+                return 0;
                 return (imsg->length);
+        }
 …
         /// Iterates over a partial set of buffers.
         template<typename T>
         inline void foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const {
+        inline void msg_foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const {
                 T op = work;
                 if (size_ == 0) size_ = size() - index_;
 …
         inline void read(boctet_t* mem, size_t idx = 0, size_t size_ = 0) const {
                 struct read_buffer rb = { mem };
                 foreach(rb, idx, size_);
+                msg_foreach(rb, idx, size_);
+        }
 …
         inline void write(const boctet_t* mem, size_t idx = 0, size_t size_ = 0) {
                 struct write_buffer wb = { mem };
                 foreach(wb, idx, size_);
+                msg_foreach(wb, idx, size_);
+        }
 …
                 message_t m;
                 struct sub_message sm = { &m };
                 foreach(sm, index, size);
+                msg_foreach(sm, index, size);
                 return m;
+        }

source/ariba/utility/transport/messages/shared_buffer.hpp

-              r10700
+              r12060
 #include <cstring>
+#include <string>
 #include <boost/shared_ptr.hpp>
 …
 #include "buffer.hpp"
+#include <stdexcept>
 namespace reboost {
+class illegal_sub_buffer: public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    explicit illegal_sub_buffer(const std::string& __arg)  :
+        std::runtime_error(__arg)
+    {
+    }
+    virtual ~illegal_sub_buffer() throw() {}
+};
 /**
 …
                 parent(new deleteable_buffer(buffer, size))
+        {
+        }
+//    /// XXX debug... copy!
+//      /// create shared buffer from buffer
+//      inline shared_buffer_t(const char* buffer, bsize_t size) :
+//              buffer_t(), parent(new deleteable_buffer(size)) {
 //              memcpy(parent->mutable_data(), buffer, parent->size());
+//              data(parent->mutable_data());
+//              this->size(parent->size());
+        }
+//              data(parent->mutable_data()); this->size(parent->size());
+//      }
         /// clone data from a normal buffer
 …
         /// return sub-buffer.
+        inline self operator()(bsize_t index, bsize_t size = 0) const {
+        inline self operator()(bsize_t index, bsize_t size = 0) const
+        {
+            // special cases
+            if ( index + size > size_ )
+            {
+                // empty sub-buffer
+            if ( index == size_ )
+            {
+                self n;
+                return n;
+            }
+            // ERROR: index out of bounds
+            throw illegal_sub_buffer("Index or size out of bounds in shared_buffer");
+            }
+            // regular case
                 self n(*this);
                 n.data_ += index;

source/ariba/utility/transport/rfcomm/CMakeLists.txt

-              r10700
+              r12060
     bluetooth_endpoint.hpp
     bluetooth_rfcomm.hpp
-    rfcomm_transport.hpp
+    )
+add_sources(rfcomm_transport.cpp)
+#add_sources()

source/ariba/utility/transport/transport_peer.cpp

-              r10700
+              r12060
-#include "ariba/config.h"
 #include "transport_peer.hpp"
+#include "transport.hpp"
+#include <boost/asio/ip/tcp.hpp>
+// ariba
+#include "StreamTransport/StreamTransport.hpp"
+#include "ariba/utility/addressing2/tcpip_endpoint.hpp"
+// boost
 #include <boost/asio/error.hpp>
 #include <boost/foreach.hpp>
-#ifdef ECLIPSE_PARSER
-    #define foreach(a, b) for(a : b)
-#else
-    #define foreach(a, b) BOOST_FOREACH(a, b)
-#endif
 // namespace ariba::transport
 …
 namespace transport {
 using namespace ariba::addressing;
+using namespace addressing2;
 using boost::asio::ip::tcp;
 …
 use_logging_cpp(transport_peer);
+transport_peer::transport_peer( endpoint_set& local_set ) : local(local_set) {
+    // setup tcp transports
+    foreach(tcp_port_address port, local.tcp) {
+transport_peer::transport_peer()  :
+        local(new addressing2::endpoint_set())
+{
+}
+EndpointSetPtr transport_peer::add_listenOn_endpoints(EndpointSetPtr endpoints)
+{
+    // TCP Endpoints
+    BOOST_FOREACH( shared_ptr<tcpip_endpoint> endp, endpoints->get_tcpip_endpoints() )
+    {
+        // automatic port detection
+        bool port_detection = false;
+        uint16_t try_port = 41322;
+        if (local.ip.size() > 0) {
+            foreach(ip_address ip_addr, local.ip) {
+                tcp::endpoint endp(ip_addr.asio(), port.asio());
+                create_service(endp);
+            }
+        } else {
+            tcp::endpoint endp_v6(tcp::v6(), port.asio());
+            tcp::endpoint endp_v4(tcp::v4(), port.asio());
+            create_service(endp_v6);
+            create_service(endp_v4);
+        tcp::endpoint asio_endp = endp->to_asio();
+        if ( asio_endp.port() == 0 )
+        {
+            port_detection = true;
+        }
+    }
+        // create new server socket
+        do
+        {
+            try
+            {
+                // automatic port detection
+                if ( port_detection )
+                {
+                    asio_endp.port(try_port);
+                    endp = tcpip_endpoint::create_TcpIP_Endpoint(asio_endp);
+                }
+                TransportProtocolPtr tmp_ptr(new StreamTransport<tcp>(endp->to_asio()));
+                transport_streams.push_back(tmp_ptr);
+                logging_info("Listening on IP/TCP " << endp->to_string());
+                local->add_endpoint(endp);
+                port_detection = false;
+            }
+            catch (boost::system::system_error& e)
+            {
+                // address in use
+                if (e.code() == boost::asio::error::address_in_use)
+                {
+                    // BRANCH: automatic port detection
+                    if ( port_detection )
+                    {
+                        // give up ?
+                        if ( try_port > 41422 )
+                        {
+                            logging_warn("[WARN] Unable to find free port. Giving up. :-( Last try was: "
+                                << endp->to_string() << ". Endpoint will be ignored!");
+                            port_detection = false;
+                        }
+                        else
+                        {
+                            // try next port
+                            try_port++;
+                        }
+                    }
+                    // BRANCH: explicit given port --> error
+                    else
+                    {
+                        logging_warn("[WARN] Address already in use: "
+                            << endp->to_string() << ". Endpoint will be ignored!");
+                    }
+                }
+                // Rethrow
+                else
+                {
+                    throw;
+                }
+            }
+        } while ( port_detection );
+    }
+    // TODO Bluetooth Endpoints
         #ifdef HAVE_LIBBLUETOOTH
     foreach(rfcomm_channel_address channel, local.rfcomm) {
         if (local.bluetooth.size() > 0) {
                 foreach(mac_address mac, local.bluetooth) {
                         rfcomm::endpoint endp(mac.bluetooth(), channel.value());
                         create_service(endp);
+                }
         } else {
                 rfcomm::endpoint endp(channel.value());
                 create_service(endp);
+        }
+    }
+//    foreach(rfcomm_channel_address channel, local.rfcomm) {
+//      if (local.bluetooth.size() > 0) {
+//              foreach(mac_address mac, local.bluetooth) {
+//                      rfcomm::endpoint endp(mac.bluetooth(), channel.value());
+//                      create_service(endp);
+//              }
+//      } else {
+//              rfcomm::endpoint endp(channel.value());
+//              create_service(endp);
+//      }
+//    }
         #endif
+}
+void transport_peer::create_service(tcp::endpoint endp) {
+    try {
+        TcpIpPtr tmp_ptr(new tcpip(endp));
+        tcps.push_back(tmp_ptr);
+        logging_info("Listening on IP/TCP " << endp);
+    } catch (boost::system::system_error& e) {
+        if (e.code() == boost::asio::error::address_in_use) {
+            logging_warn("[WARN] Address already in use: "
+                    << endp << ". Endpoint will be ignored!");
+        } else {
+            // Rethrow
+            throw;
+        }
+    }
+}
+    return local;
+}
+//void transport_peer::create_service(tcp::endpoint endp) {
+//    try {
+//        TransportProtocolPtr tmp_ptr(new StreamTransport<tcp>(endp));
+//        tcps.push_back(tmp_ptr);
+//        logging_info("Listening on IP/TCP " << endp);
+//
+//    } catch (boost::system::system_error& e) {
+//        if (e.code() == boost::asio::error::address_in_use) {
+//            logging_warn("[WARN] Address already in use: "
+//                    << endp << ". Endpoint will be ignored!");
+//        } else {
+//            // Rethrow
+//            throw;
+//        }
+//    }
+//}
 #ifdef HAVE_LIBBLUETOOTH
 void transport_peer::create_service(rfcomm::endpoint endp) {
     try {
         rfcomm_transport::sptr tmp_ptr(new rfcomm_transport(endp));
         rfcomms.push_back(tmp_ptr);
         logging_info("Listening on bluetooth/RFCOMM " << endp);
     } catch (boost::system::system_error& e) {
         if (e.code() == boost::asio::error::address_in_use) {
             logging_warn("[WARN] Address already in use: "
                     << endp << ". Endpoint will be ignored!");
         } else {
             // Rethrow
             throw;
+        }
+    }
+}
+//void transport_peer::create_service(rfcomm::endpoint endp) {
+//    try {
+//        TransportProtocolPtr tmp_ptr(new StreamTransport<rfcomm>(endp));
+//        rfcomms.push_back(tmp_ptr);
+//        logging_info("Listening on bluetooth/RFCOMM " << endp);
+//
+//    } catch (boost::system::system_error& e) {
+//        if (e.code() == boost::asio::error::address_in_use) {
+//            logging_warn("[WARN] Address already in use: "
+//                    << endp << ". Endpoint will be ignored!");
+//        } else {
+//            // Rethrow
+//            throw;
+//        }
+//    }
+//}
 #endif
 …
+}
+void transport_peer::start() {
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->start();
+    }
+#ifdef HAVE_LIBBLUETOOTH
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+        x->start();
+    }
+#endif
+}
+void transport_peer::stop() {
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->stop();
+    }
+#ifdef HAVE_LIBBLUETOOTH
+        foreach(rfcomm_transport::sptr x, rfcomms) {
+                x->stop();
+        }
+#endif
+void transport_peer::start()
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->start();
+    }
+}
+void transport_peer::stop()
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->stop();
+    }
+}
 void transport_peer::send(
         const endpoint_set& endpoints,
+        const const_EndpointSetPtr endpoints,
         reboost::message_t message,
         uint8_t priority)
+{
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->send(endpoints, message, priority);
+    }
+#ifdef HAVE_LIBBLUETOOTH
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+                x->send(endpoints, message, priority);
+        }
+#endif
+}
+void transport_peer::terminate( const address_v* remote ) {
+        if (remote->instanceof<tcpip_endpoint>())// TODO direkt auf der richtigen verbindung
+        {
+            foreach(TcpIpPtr tcp, tcps) {
+                tcp->terminate(remote);
+            }
+        }
+#ifdef HAVE_LIBBLUETOOTH
+        if (remote->instanceof<rfcomm_endpoint>()) {
+                foreach(rfcomm_transport::sptr x, rfcomms) {
+                        x->terminate(remote);
+                }
+        }
+#endif
+}
+void transport_peer::register_listener( transport_listener* listener ) {
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->register_listener(listener);
+    }
+#ifdef HAVE_LIBBLUETOOTH
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+        x->register_listener(listener);
+    }
+#endif
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->send(endpoints, message, priority);
+    }
+}
+// XXX DEPRECATED
+//void transport_peer::terminate( const address_v* remote ) {
+//      if (remote->instanceof<tcpip_endpoint>())// TODO direkt auf der richtigen verbindung
+//      {
+//          foreach(TransportProtocolPtr tcp, tcps) {
+//              tcp->terminate(remote);
+//          }
+//      }
+//#ifdef HAVE_LIBBLUETOOTH
+//      if (remote->instanceof<rfcomm_endpoint>()) {
+//              foreach(TransportProtocolPtr x, rfcomms) {
+//                      x->terminate(remote);
+//              }
+//      }
+//#endif
+//}
+void transport_peer::register_listener( transport_listener* listener )
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->register_listener(listener);
+    }
+}

source/ariba/utility/transport/transport_peer.hpp

-              r10700
+              r12060
 #define TRANSPORT_PEER_HPP_
+// ariba
 #include "ariba/config.h"
 #include "ariba/utility/logging/Logging.h"
+#include "transport_protocol.hpp"
+#include "ariba/utility/addressing/endpoint_set.hpp"
+#include "ariba/utility/addressing2/endpoint_set.hpp"
+// ariba interfaces
+#include "interfaces/transport_protocol.hpp"
+// boost
 #include <boost/shared_ptr.hpp>
+#include "rfcomm/bluetooth_rfcomm.hpp"
+// boost-adaption
+//#include "rfcomm/bluetooth_rfcomm.hpp"
 …
 namespace transport {
-using namespace ariba::addressing;
-class tcpip;
-#ifdef HAVE_LIBBLUETOOTH
-class rfcomm_transport;
-#endif
 /**
+ * TODO: Doc
+ * This class allocates implementations of various transport
+ * protocols and can send messages to an entire set of endpoints
+ *
  * @author Sebastian Mies <mies@tm.uka.de>
+ * @author Sebastian Mies <mies@tm.uka.de>, Mario Hock
  */
+/// this transport peer allocates implementations of various transport
+/// protocols and can send messages to an entire set of endpoints
 class transport_peer : public transport_protocol {
+class transport_peer :
+    public transport_protocol
+{
         use_logging_h(transport_peer);
+        typedef boost::shared_ptr<transport_protocol> TransportProtocolPtr;
 public:
+        transport_peer( endpoint_set& local_set );
+        transport_peer();
+        /**
+         * Adds endpoints on which ariba should listen ("server"-sockets)
+         *
+         * @return An endpoint_set holding all active endpoints ariba is listening on.
+         */
+        addressing2::EndpointSetPtr add_listenOn_endpoints(addressing2::EndpointSetPtr endpoints);
         virtual ~transport_peer();
         virtual void start();
 …
         virtual void send(
                 const endpoint_set& endpoints,
+                const addressing2::const_EndpointSetPtr endpoints,
                 reboost::message_t message,
+                uint8_t priority = 0);
+        /// @deprecated: Use terminate() from transport_connection instead
+        virtual void terminate( const address_v* remote );
+                uint8_t priority = system_priority::OVERLAY);
         virtual void register_listener( transport_listener* listener );
 private:
+        void create_service(tcp::endpoint endp);
+#ifdef HAVE_LIBBLUETOOTH
+        void create_service(boost::asio::bluetooth::rfcomm::endpoint endp);
+#endif
+        endpoint_set&  local;
+        std::vector< boost::shared_ptr<tcpip> > tcps;
+#ifdef HAVE_LIBBLUETOOTH
+        std::vector< boost::shared_ptr<rfcomm_transport> > rfcomms;
+#endif
+        addressing2::EndpointSetPtr local;
+        std::vector<TransportProtocolPtr> transport_streams;
 };

source/ariba/utility/types/Identifier.h

-              r4625
+              r12060
 #include "ariba/utility/serialization.h"
+// XXX EXPERIMENTAL
+#include "ariba/utility/transport/messages/shared_buffer.hpp"
 /**< maximum length of the key */
 #define MAX_KEYLENGTH 192
 …
         use_logging_h( Identifier );
 public:
+        // XXX EXPERIMENTAL
+        reboost::shared_buffer_t serialize() const
+        {
+            Data data = data_serialize(this, DEFAULT_V);
+            reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8);
+            return buf;
+        }
+        reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff)
+        {
+            Data dat(const_cast<uint8_t*>(buff.data()), buff.size() * 8);
+            size_t len = this->SERIALIZATION_METHOD_NAME(DESERIALIZE, dat) / 8;
+            // return remaining sub-buffer
+            return buff(len);
+        }
         //-------------------------------------------------------------------------

source/ariba/utility/types/OverlayParameterSet.h

r4625	r12060
53	53
54	54	typedef enum _OverlayStructure {
55		OverlayStructureOneHop = 0,
	55	OverlayStructureOneHop = 0, // DEPRECATED, DO NOT USE
56	56	OverlayStructureChord = 1,
57	57	} OverlayStructure;

source/ariba/utility/types/ServiceID.cpp

r3690	r12060
76	76	}
77	77
78		string ServiceID::toString() const {
	78	string ServiceID::toString() const
	79	{
	80	if ( *this == ServiceID::UNSPECIFIED )
	81	return "UNSPEC";
	82
79	83	return ariba::utility::Helper::ultos( id );
80	84	}

source/services/ariba_dht/Dht.cpp

-              r10700
+              r12060
 void Dht::get(const std::string& key)
+{
         DhtMessage msg(DhtMessage::DhtGet, key);
+        DhtMessage msg(DhtMessage::DhtGet, key, node->getNodeId());
         handle_dht_message(msg, NodeID::UNSPECIFIED);
 …
             case DhtMessage::DhtGet:
+            {
                 answer_dht_request(message.getKey(), source);
+                answer_dht_request(message.getKey(), message.getSourceNode());
                 break;
 …
                                 message.getValues(),
                                 message.getTTL());
                 answer_dht_request(message.getKey(), source);
+                answer_dht_request(message.getKey(), message.getSourceNode());
                 break;

source/services/ariba_dht/Dht.h

-              r10700
+              r12060
 using ariba::utility::SystemEventType;
 using ariba::utility::SystemEventListener;
+using ariba::utility::NodeID;
+using ariba::utility::LinkID;
 // Forward declarations to avoid adding messages/*.h to the public interface

source/services/ariba_dht/messages/DhtMessage.cpp

-              r10700
+              r12060
 DhtMessage::DhtMessage() :
         ttl( 0 ),
+        replace( false )
+        replace( false ),
+        sourceNode(NodeID::UNSPECIFIED)
 {}
 DhtMessage::DhtMessage( DhtMessageType type, const std::string& key ) :
+DhtMessage::DhtMessage( DhtMessageType type, const std::string& key, const NodeID& sourceNodeID ) :
         type( static_cast<uint8_t>(type) ),
         ttl( 0 ),
         replace( false ),
+        key( key )
+        key( key ),
+        sourceNode(sourceNodeID)
 {}
 DhtMessage::DhtMessage( DhtMessageType type, const std::string& key,
                 const std::string& value, uint16_t ttl ) :
+                const std::string& value, uint16_t ttl, const NodeID& sourceNodeID ) :
         type( static_cast<uint8_t>(type) ),
         ttl( ttl ),
         replace( false ),
         key( key ),
+        values(1, value)
+        values(1, value),
+        sourceNode(sourceNodeID)
 {}
 DhtMessage::DhtMessage( DhtMessageType type, const std::string& key,
                 const vector<string>& values, uint16_t ttl ) :
+                const vector<string>& values, uint16_t ttl, const NodeID& sourceNodeID ) :
         type( static_cast<uint8_t>(type) ),
         ttl( ttl ),
         replace( false ),
+        key( key )
+        key( key ),
+        sourceNode(sourceNodeID)
+{
         // preallocate enough room so we don't need to copy a lot
 …
+}
+string DhtMessage::DhtMessageTypeToString(DhtMessageType type) {
+        string temp;
+        switch (type)
+        {
+                case DhtMessage::DhtInvalid:
+                {
+                        temp = "DhtInvalid";
+                        break;
+                }
+                case DhtMessage::DhtGet:
+                {
+                        temp = "DhtGet";
+                        break;
+                }
+                case DhtMessage::DhtPut:
+                {
+                        temp = "DhtPut";
+                        break;
+                }
+                case DhtMessage::DhtPutAndGet:
+                {
+                        temp = "DhtPutAndGet";
+                        break;
+                }
+                case DhtMessage::DhtRemove:
+                {
+                        temp = "DhtRemove";
+                        break;
+                }
+                case DhtMessage::DhtRepublish:
+                {
+                        temp = "DhtRepublish";
+                        break;
+                }
+                case DhtMessage::DhtAnswer:
+                {
+                        temp =  "DhtAnswer";
+                        break;
+                }
+                case DhtMessage::DhtReplica:
+                {
+                        temp = "DhtReplica";
+                        break;
+                }
+        }
+        return temp;
+}
 }}

source/services/ariba_dht/messages/DhtMessage.h

-              r10700
+              r12060
 #include "ariba/utility/messages.h"
+#include "ariba/utility/types/NodeID.h"
 #include "ariba/utility/serialization.h"
 #include "ariba/Name.h"
 …
 using ariba::utility::Message;
+using ariba::utility::NodeID;
 using_serialization;
 …
                 DhtRemove = 4,
                 DhtRepublish = 5,
+                DhtAnswer = 8
+                DhtAnswer = 8,
+                DhtReplica = 9
         } DhtMessageType;
         DhtMessage();
         DhtMessage( DhtMessageType type, const std::string& key );
+        DhtMessage( DhtMessageType type, const std::string& key, const NodeID& sourceNodeID = NodeID::UNSPECIFIED);
         DhtMessage( DhtMessageType type, const std::string& key,
                         const std::string& value, uint16_t ttl = 0 );
+                        const std::string& value, uint16_t ttl = 0, const NodeID& sourceNodeID = NodeID::UNSPECIFIED );
         DhtMessage( DhtMessageType type, const std::string& key,
                         const vector<std::string>& values, uint16_t ttl = 0 );
+                        const vector<std::string>& values, uint16_t ttl = 0, const NodeID& sourceNodeID = NodeID::UNSPECIFIED );
         virtual ~DhtMessage();
 …
+        }
+        NodeID getSourceNode() const {
+                return sourceNode;
+        }
         bool doReplace() const {
                 return replace;
+        }
+        static string DhtMessageTypeToString(DhtMessageType type);
 private:
+protected:
         uint8_t type;
         uint16_t ttl;
 …
         std::string key;
         vector<std::string> values;
+        NodeID sourceNode;
+private:
 };
 …
         // key serialization
         X && T(key);
+        X && &sourceNode;
         // store number of values

Context Navigation

Legend:

Download in other formats: