Index: /Makefile.am =================================================================== --- /Makefile.am (revision 10647) +++ /Makefile.am (revision 10653) @@ -1,5 +1,12 @@ ACLOCAL_AMFLAGS = -I m4 SUBDIRS = docu source sample -EXTRA_DIST = etc/pingpong/* etc/patch/* INSTALL LICENSE README bootstrap bootstrap_libs +EXTRA_DIST = etc/pingpong/* \ + etc/patch/* \ + INSTALL \ + LICENSE \ + README \ + bootstrap \ + bootstrap_libs \ + docu/doxygen/Doxyfile # hook to remove all .svn files before rolling the tarball Index: /configure.ac =================================================================== --- /configure.ac (revision 10647) +++ /configure.ac (revision 10653) @@ -1,3 +1,3 @@ -AC_INIT([ariba], [0.7.1], [http://www.ariba-underlay.org]) +AC_INIT([ariba], [0.8.1], [http://www.ariba-underlay.org]) AM_INIT_AUTOMAKE([-Wall foreign]) AC_CONFIG_SRCDIR([source/ariba/ariba.h]) @@ -58,5 +58,6 @@ sample/Makefile sample/pingpong/Makefile -sample/testdht/Makefile +source/services/Makefile +source/services/dht/Makefile docu/Makefile docu/doxygen/Makefile Index: /docu/doxygen/Doxyfile =================================================================== --- /docu/doxygen/Doxyfile (revision 10647) +++ /docu/doxygen/Doxyfile (revision 10653) @@ -6,5 +6,5 @@ DOXYFILE_ENCODING = UTF-8 PROJECT_NAME = Ariba -PROJECT_NUMBER = 0.7.0 +PROJECT_NUMBER = 0.8.1 OUTPUT_DIRECTORY = ./ CREATE_SUBDIRS = NO Index: /sample/Makefile.am =================================================================== --- /sample/Makefile.am (revision 10647) +++ /sample/Makefile.am (revision 10653) @@ -1,1 +1,2 @@ -SUBDIRS = pingpong testdht +#SUBDIRS = pingpong testdht +SUBDIRS = pingpong Index: /source/Makefile.am =================================================================== --- /source/Makefile.am (revision 10647) +++ /source/Makefile.am (revision 10653) @@ -1,1 +1,1 @@ -SUBDIRS = ariba +SUBDIRS = ariba services Index: /source/ariba/Makefile.am =================================================================== --- /source/ariba/Makefile.am (revision 10647) +++ /source/ariba/Makefile.am (revision 10653) @@ -18,5 +18,5 @@ # project version number!! -libariba_la_LDFLAGS = -version-info 0:0:0 +libariba_la_LDFLAGS = -version-info 1:0:0 # compiler flags ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -144,5 +144,4 @@ overlay/messages/JoinReply.cpp \ overlay/messages/JoinRequest.cpp \ - overlay/messages/DHTMessage.cpp \ overlay/messages/OverlayMsg.cpp @@ -150,5 +149,4 @@ overlay/messages/JoinReply.h \ overlay/messages/JoinRequest.h \ - overlay/messages/DHTMessage.h\ overlay/messages/OverlayMsg.h @@ -278,64 +276,27 @@ libariba_la_SOURCES += \ utility/transport/tcpip/tcpip.cpp \ - utility/transport/tcpip/protlib/timer_module.cpp \ - utility/transport/tcpip/protlib/setuid.cpp \ - utility/transport/tcpip/protlib/queuemanager.cpp \ - utility/transport/tcpip/protlib/messages.cpp \ - utility/transport/tcpip/protlib/fqueue.cpp \ - utility/transport/tcpip/protlib/fastqueue.c \ - utility/transport/tcpip/protlib/eclock_gettime.c \ - utility/transport/tcpip/protlib/tp_over_udp.cpp \ - utility/transport/tcpip/protlib/connectionmap_uds.cpp \ - utility/transport/tcpip/protlib/network_message.cpp \ - utility/transport/tcpip/protlib/threadsafe_db.cpp \ - utility/transport/tcpip/protlib/timer.cpp \ - utility/transport/tcpip/protlib/address.cpp \ - utility/transport/tcpip/protlib/connectionmap.cpp \ - utility/transport/tcpip/protlib/tp.cpp \ - utility/transport/tcpip/protlib/tp_over_tcp.cpp \ - utility/transport/tcpip/protlib/configuration.cpp \ - utility/transport/tcpip/protlib/ie.cpp \ - utility/transport/tcpip/protlib/threads.cpp \ - utility/transport/tcpip/protlib/logfile.cpp \ utility/transport/transport_peer.cpp \ - utility/transport/rfcomm/rfcomm.cpp \ - utility/transport/asio/asio_io_service.cpp + utility/transport/rfcomm/rfcomm_transport.cpp \ + utility/transport/asio/unique_io_service.cpp \ + utility/transport/messages/buffer.cpp \ + utility/transport/messages/message.cpp \ + utility/transport/messages/shared_buffer.cpp nobase_libariba_la_HEADERS += \ utility/transport/test_transport.hpp \ utility/transport/tcpip/tcpip.hpp \ + utility/transport/transport_connection.hpp \ + utility/transport/transport_listener.hpp \ utility/transport/transport_peer.hpp \ utility/transport/transport_protocol.hpp \ - utility/transport/rfcomm/rfcomm.hpp \ + utility/transport/rfcomm/rfcomm_transport.hpp \ + utility/transport/rfcomm/bluetooth_endpoint.hpp \ + utility/transport/rfcomm/bluetooth_rfcomm.hpp \ utility/transport/transport.hpp \ - utility/transport/asio/bluetooth_endpoint.hpp \ - utility/transport/asio/rfcomm.hpp \ - utility/transport/transport_listener.hpp \ - utility/transport/asio/asio_io_service.h \ - utility/transport/tcpip/protlib/threadsafe_db.h \ - utility/transport/tcpip/protlib/configuration.h \ - utility/transport/tcpip/protlib/ie.h \ - utility/transport/tcpip/protlib/llhashers.h \ - utility/transport/tcpip/protlib/fqueue.h \ - utility/transport/tcpip/protlib/assocdata_uds.h \ - utility/transport/tcpip/protlib/address.h \ - utility/transport/tcpip/protlib/logfile.h \ - utility/transport/tcpip/protlib/timer.h \ - utility/transport/tcpip/protlib/queuemanager.h \ - utility/transport/tcpip/protlib/messages.h \ - utility/transport/tcpip/protlib/assocdata.h \ - utility/transport/tcpip/protlib/protlib_types.h \ - utility/transport/tcpip/protlib/tp_over_tcp.h \ - utility/transport/tcpip/protlib/tp_over_udp.h \ - utility/transport/tcpip/protlib/tp.h \ - utility/transport/tcpip/protlib/threads.h \ - utility/transport/tcpip/protlib/connectionmap.h \ - utility/transport/tcpip/protlib/timer_module.h \ - utility/transport/tcpip/protlib/fastqueue.h \ - utility/transport/tcpip/protlib/tperror.h \ - utility/transport/tcpip/protlib/network_message.h \ - utility/transport/tcpip/protlib/setuid.h \ - utility/transport/tcpip/protlib/cleanuphandler.h \ - utility/transport/tcpip/protlib/connectionmap_uds.h + utility/transport/asio/unique_io_service.h \ + utility/transport/messages/buffer.hpp \ + utility/transport/messages/buffers.hpp \ + utility/transport/messages/message.hpp \ + utility/transport/messages/shared_buffer.hpp #------------> utility :: messages Index: /source/ariba/Node.cpp =================================================================== --- /source/ariba/Node.cpp (revision 10647) +++ /source/ariba/Node.cpp (revision 10653) @@ -177,4 +177,11 @@ } +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); @@ -209,14 +216,4 @@ } -// service directory - -void Node::put( const Data& key, const Data& value, uint16_t ttl, bool replace ) { - base_overlay->dhtPut(key,value,ttl,replace); -} - -void Node::get( const Data& key, const ServiceID& sid ) { - base_overlay->dhtGet(key,sid); -} - // @see Module.h string Node::getName() const { Index: /source/ariba/Node.h =================================================================== --- /source/ariba/Node.h (revision 10647) +++ /source/ariba/Node.h (revision 10653) @@ -242,4 +242,11 @@ /** + * like the above function, but sends the message to the closest directly known node + * to the specified address + */ + 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 @@ -281,23 +288,4 @@ */ bool unbind(CommunicationListener* listener, const ServiceID& sid); - - /** - * Adds a key value pair to the DHT - * - * @param key The key data - * @param value The value data - * @param ttl The time to live in seconds - */ - void put( const Data& key, const Data& value, uint16_t ttl, bool replace = false); - - /** - * Queries for values stored in the DHT. Fires an communication event when - * values arrive. - * - * @param key The key data - * @param sid The service that is requesting the values - */ - void get( const Data& key, const ServiceID& sid ); - //------------------------------------------------------------------------- Index: /source/ariba/SideportListener.h =================================================================== --- /source/ariba/SideportListener.h (revision 10647) +++ /source/ariba/SideportListener.h (revision 10653) @@ -47,5 +47,4 @@ #include "CommunicationListener.h" -using std::cout; using std::map; using std::vector; Index: /source/ariba/communication/BaseCommunication.cpp =================================================================== --- /source/ariba/communication/BaseCommunication.cpp (revision 10647) +++ /source/ariba/communication/BaseCommunication.cpp (revision 10653) @@ -41,4 +41,5 @@ #include "networkinfo/AddressDiscovery.h" #include "ariba/utility/types/PeerID.h" +#include #ifdef UNDERLAY_OMNET @@ -283,12 +284,4 @@ SystemEventType MessageDispatchEvent("MessageDispatchEvent", TransportEvent ); -class DispatchMsg { -public: - DispatchMsg() : local(NULL), remote(NULL), message(NULL) {} - address_v* local; - address_v* remote; - Message* message; -}; - /// called when a system event is emitted by system queue void BaseCommunication::handleSystemEvent(const SystemEvent& event) { @@ -297,42 +290,55 @@ if ( event.getType() == MessageDispatchEvent ){ logging_debug( "Forwarding message receiver" ); - DispatchMsg* dmsg = event.getData(); - Message* msg = dmsg->message; - receiveMessage(msg, dmsg->local, dmsg->remote); - msg->dropPayload(); - delete dmsg->local; - delete dmsg->remote; - delete msg; - delete dmsg; - } -} - -/// called when a message is received from transport_peer -void BaseCommunication::receive_message(transport_protocol* transport, - const address_vf local, const address_vf remote, const uint8_t* data, - size_t size) { - -// logging_debug( "Dispatching message" ); - - // convert data - Data data_( const_cast(data), size * 8 ); - DispatchMsg* dmsg = new DispatchMsg(); - - Message* msg = new Message(data_); - dmsg->local = local->clone(); - dmsg->remote = remote->clone(); - dmsg->message = msg; - - SystemQueue::instance().scheduleEvent( - SystemEvent( this, MessageDispatchEvent, dmsg ) - ); -} - -/// handles a message from the underlay transport -void BaseCommunication::receiveMessage(const Message* message, - const address_v* local, const address_v* remote ){ - + boost::function0* handler = event.getData< boost::function0 >(); + (*handler)(); + delete handler; + } +} + +/** + * called within the ASIO thread + * when a message is received from underlay transport + */ +void BaseCommunication::receive_message(transport_connection::sptr connection, + reboost::message_t msg) { + + logging_debug( "Dispatching message" ); + + boost::function0* handler = new boost::function0( + boost::bind( + &BaseCommunication::receiveMessage, + this, + connection, + msg) + ); + + SystemQueue::instance().scheduleEvent( + SystemEvent(this, MessageDispatchEvent, handler) + ); +} + +/** + * called within the ARIBA thread (System Queue) + * when a message is received from underlay transport + */ +void BaseCommunication::receiveMessage(transport_connection::sptr connection, + reboost::message_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 = ((Message*)message)->decapsulate(); + AribaBaseMsg* msg = legacy_message.decapsulate(); logging_debug( "Receiving message of type " << msg->getTypeString() ); @@ -379,6 +385,7 @@ LinkID localLink = LinkID::create(); LinkID remoteLink = msg->getLocalLink(); - logging_debug( "local=" << local->to_string() - << " remote=" << remote->to_string() + logging_debug( + "local=" << connection->getLocalEndpoint()->to_string() + << " remote=" << connection->getRemoteEndpoint()->to_string() ); @@ -386,5 +393,7 @@ bool allowlink = true; BOOST_FOREACH( CommunicationEvents* i, eventListener ){ - allowlink &= i->onLinkRequest( localLink, local, remote ); + allowlink &= i->onLinkRequest( localLink, + connection->getLocalEndpoint(), + connection->getRemoteEndpoint()); } @@ -400,6 +409,7 @@ ld->localLink = localLink; ld->remoteLink = remoteLink; - ld->localLocator = local->clone(); - ld->remoteLocator = remote->clone(); + ld->localLocator = connection->getLocalEndpoint()->clone(); + ld->remoteLocator = connection->getRemoteEndpoint()->clone(); + ld->connection = connection; ld->remoteEndpoint = msg->getLocalDescriptor(); add_endpoint(ld->remoteLocator); @@ -409,5 +419,6 @@ ld->remoteLocator, endpoint_set::Layer1_3 | endpoint_set::NoLoopback); localDescriptor.getEndpoints().add( - local, endpoint_set::Layer1_3 | endpoint_set::NoLoopback); + connection->getLocalEndpoint(), + endpoint_set::Layer1_3 | endpoint_set::NoLoopback); // link is now up-> add it @@ -459,7 +470,10 @@ } + // store the connection + ld.connection = connection; + // set remote locator and link id ld.remoteLink = msg->getLocalLink(); - ld.remoteLocator = remote->clone(); + ld.remoteLocator = connection->getRemoteEndpoint()->clone(); ld.remoteEndpoint.getEndpoints().add( msg->getLocalDescriptor().getEndpoints(), @@ -536,5 +550,5 @@ // update the remote locator const address_v* oldremote = linkDesc.remoteLocator; - linkDesc.remoteLocator = remote->clone(); + linkDesc.remoteLocator = connection->getRemoteEndpoint()->clone(); // inform the listeners (local link has _not_ changed!) @@ -747,16 +761,31 @@ /// sends a message to all end-points in the end-point descriptor -void BaseCommunication::send(Message* message, const EndpointDescriptor& endpoint) { - Data data = data_serialize( message, DEFAULT_V ); - transport->send( endpoint.getEndpoints(), data.getBuffer(), data.getLength() / 8); - data.release(); +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* message, const LinkDescriptor& desc) { +void BaseCommunication::send(Message* legacy_message, const LinkDescriptor& desc) { if (desc.remoteLocator==NULL) return; - Data data = data_serialize( message, DEFAULT_V ); - transport->send( desc.remoteLocator, data.getBuffer(), data.getLength() / 8); - data.release(); + + 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); } Index: /source/ariba/communication/BaseCommunication.h =================================================================== --- /source/ariba/communication/BaseCommunication.h (revision 10647) +++ /source/ariba/communication/BaseCommunication.h (revision 10653) @@ -60,4 +60,5 @@ #include "ariba/utility/addressing/addressing.hpp" #include "ariba/utility/transport/transport.hpp" +#include "ariba/utility/transport/transport_connection.hpp" // communication @@ -190,14 +191,19 @@ virtual void handleSystemEvent(const SystemEvent& event); - /// called when a message is received form transport_peer - virtual void receive_message(transport_protocol* transport, - const address_vf local, const address_vf remote, const uint8_t* data, - size_t size); + /** + * called within the ASIO thread + * when a message is received from underlay transport + */ + virtual void receive_message(transport_connection::sptr connection, + reboost::message_t msg); protected: - /// handle received message from a transport module - void receiveMessage(const Message* message, - const address_v* local, const address_v* remote ); + /** + * called within the ARIBA thread (System Queue) + * when a message is received from underlay transport + */ + void receiveMessage(transport_connection::sptr connection, + reboost::message_t msg); /// called when a network interface change happens @@ -250,4 +256,7 @@ /// flag, whether this link is up bool up; + + /// connection if link is up + transport_connection::sptr connection; }; Index: /source/ariba/communication/networkinfo/AddressDiscovery.cpp =================================================================== --- /source/ariba/communication/networkinfo/AddressDiscovery.cpp (revision 10647) +++ /source/ariba/communication/networkinfo/AddressDiscovery.cpp (revision 10653) @@ -129,5 +129,5 @@ ip_address ip = straddr; if (ip.is_loopback()) continue; - if (ip.is_link_local()) continue; +// if (ip.is_link_local()) continue; address_vf vf = ip; endpoints.add( vf ); Index: /source/ariba/communication/networkinfo/NetworkInformation.h =================================================================== --- /source/ariba/communication/networkinfo/NetworkInformation.h (revision 10647) +++ /source/ariba/communication/networkinfo/NetworkInformation.h (revision 10653) @@ -40,4 +40,5 @@ #define __NETWORK_INFORMATION_H +#include #include #include Index: /source/ariba/overlay/BaseOverlay.cpp =================================================================== --- /source/ariba/overlay/BaseOverlay.cpp (revision 10647) +++ /source/ariba/overlay/BaseOverlay.cpp (revision 10653) @@ -51,5 +51,4 @@ #include "ariba/overlay/messages/OverlayMsg.h" -#include "ariba/overlay/messages/DHTMessage.h" #include "ariba/overlay/messages/JoinRequest.h" #include "ariba/overlay/messages/JoinReply.h" @@ -66,193 +65,4 @@ #define visualIdBase ariba::utility::ServerVis::NETWORK_ID_BASE_COMMUNICATION -class ValueEntry { -public: - ValueEntry( const Data& value ) : ttl(0), last_update(time(NULL)), - last_change(time(NULL)), value(value.clone()) { - } - - ValueEntry( const ValueEntry& value ) : - ttl(value.ttl), last_update(value.last_update), - last_change(value.last_change), value(value.value.clone()) { - } - - ~ValueEntry() { - value.release(); - } - - void refresh() { - last_update = time(NULL); - } - - void set_value( const Data& value ) { - this->value.release(); - this->value = value.clone(); - this->last_change = time(NULL); - this->last_update = time(NULL); - } - - Data get_value() const { - return value; - } - - uint16_t get_ttl() const { - return ttl; - } - - void set_ttl( uint16_t ttl ) { - this->ttl = ttl; - } - - bool is_ttl_elapsed() const { - // is persistent? yes-> always return false - if (ttl==0) return false; - // return true, if ttl is elapsed - return ( difftime( time(NULL), this->last_update ) > ttl ); - } - -private: - uint16_t ttl; - time_t last_update; - time_t last_change; - Data value; -}; - -class DHTEntry { -public: - Data key; - vector values; - - vector get_values() { - vector vect; - BOOST_FOREACH( ValueEntry& e, values ) - vect.push_back( e.get_value() ); - return vect; - } - - void erase_expired_entries() { - for (vector::iterator i = values.begin(); - i != values.end(); i++ ) - if (i->is_ttl_elapsed()) - i = values.erase(i)-1; - } -}; - -class DHT { -public: - typedef vector Entries; - typedef vector Values; - Entries entries; - static const bool verbose = false; - - static bool equals( const Data& lhs, const Data& rhs ) { - if (rhs.getLength()!=lhs.getLength()) return false; - for (size_t i=0; i refresh ttl - if ( equals(entry.values[j].get_value(), value) ) { - entry.values[j].refresh(); - if (verbose) - std::cout << "DHT: Republished value. Refreshing value timestamp." - << std::endl; - return; - } - } - - // new value-> add to entry - if (verbose) - std::cout << "DHT: Added value to " - << " key=" << key << " with value=" << value << std::endl; - entry.values.push_back( ValueEntry( value ) ); - entry.values.back().set_ttl(ttl); - return; - } - } - - // key is unknown-> add key value pair - if (verbose) - std::cout << "DHT: New key value pair " - << " key=" << key << " with value=" << value << std::endl; - - // add new entry - entries.push_back( DHTEntry() ); - DHTEntry& entry = entries.back(); - entry.key = key.clone(); - entry.values.push_back( ValueEntry(value) ); - entry.values.back().set_ttl(ttl); - } - - vector get( const Data& key ) { - cleanup(); - // find entry - for (size_t i=0; i(); - } - - bool remove( const Data& key ) { - cleanup(); - - // find entry - for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { - DHTEntry& entry = *i; - - // found? yes-> delete entry - if ( equals(entry.key, key) ) { - entries.erase(i); - return true; - } - } - return false; - } - - bool remove( const Data& key, const Data& value ) { - cleanup(); - // find entry - for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { - DHTEntry& entry = *i; - - // found? yes-> try to find value - if ( equals(entry.key, key) ) { - for (Values::iterator j = entry.values.begin(); - j != entry.values.end(); j++) { - - // value found? yes-> delete - if (equals(j->get_value(), value)) { - j = entry.values.erase(j)-1; - return true; - } - } - } - } - return false; - } - - void cleanup() { - for (Entries::iterator i = entries.begin(); i != entries.end(); i++) { - DHTEntry& entry = *i; - entry.erase_expired_entries(); - if (entry.values.size()==0) - i = entries.erase(i)-1; - } - } -}; // ---------------------------------------------------------------------------- @@ -758,9 +568,7 @@ sideport(&SideportListener::DEFAULT), overlayInterface(NULL), counter(0) { - initDHT(); } BaseOverlay::~BaseOverlay() { - destroyDHT(); } @@ -1078,4 +886,5 @@ } + seqnum_t BaseOverlay::sendMessage(const Message* message, const NodeID& node, const ServiceID& service) { @@ -1114,4 +923,22 @@ } + +NodeID BaseOverlay::sendMessageCloserToNodeID(const Message* message, + const NodeID& address, const ServiceID& service) { + + if ( overlayInterface->isClosestNodeTo(address) ) + { + return NodeID::UNSPECIFIED; + } + + const NodeID& closest_node = overlayInterface->getNextNodeId(address); + + if ( closest_node != NodeID::UNSPECIFIED ) + { + seqnum_t seqnum = sendMessage(message, closest_node, service); + } + + return closest_node; // XXX return seqnum ?? tuple? closest_node via (non const) reference? +} // ---------------------------------------------------------------------------- @@ -1831,11 +1658,4 @@ overlayMsg->addRouteRecord(nodeId); - // handle dht messages (do not route) - if (overlayMsg->isDHTMessage()) { - bool ret = handleDHTMessage(overlayMsg); - delete overlayMsg; - return ret; - } - // handle signaling messages (do not route!) if (overlayMsg->getType()>=OverlayMsg::typeSignalingStart && @@ -1856,11 +1676,4 @@ delete overlayMsg; return true; - } - - // handle DHT response messages - if (overlayMsg->hasTypeMask( OverlayMsg::maskDHTResponse )) { - bool ret = handleDHTMessage(overlayMsg); - delete overlayMsg; - return ret; } @@ -1964,5 +1777,4 @@ stabilizeRelays(); stabilizeLinks(); - stabilizeDHT(); updateVisual(); } @@ -2110,221 +1922,4 @@ // ---------------------------------------------------------------------------- -void BaseOverlay::initDHT() { - dht = new DHT(); - localDHT = new DHT(); - republishCounter = 0; -} - -void BaseOverlay::destroyDHT() { - delete dht; - delete localDHT; -} - -/// stabilize DHT state -void BaseOverlay::stabilizeDHT() { - - // do refresh every 2 seconds - if (republishCounter < 2) { - republishCounter++; - return; - } - republishCounter = 0; - - // remove old values from DHT - BOOST_FOREACH( DHTEntry& entry, dht->entries ) { - // erase old entries - entry.erase_expired_entries(); - } - - // re-publish values-> do not refresh locally stored values - BOOST_FOREACH( DHTEntry& entry, localDHT->entries ) { - BOOST_FOREACH( ValueEntry& value, entry.values ) - dhtPut(entry.key, value.get_value(), value.get_ttl(), false, true ); - } -} - -// handle DHT messages -bool BaseOverlay::handleDHTMessage( OverlayMsg* msg ) { - - // de-capsulate message - logging_debug("Received DHT message"); - DHTMessage* dhtMsg = msg->decapsulate(); - - // handle DHT data message - if (msg->getType()==OverlayMsg::typeDHTData) { - const ServiceID& service = msg->getService(); - logging_info( "Received DHT data for service " << service.toString() ); - - // delegate data message - CommunicationListener* lst = getListener(service); - if(lst != NULL) lst->onKeyValue(dhtMsg->getKey(), dhtMsg->getValues() ); - delete dhtMsg; - return true; - } - - // route message to closest node - if (!overlayInterface->isClosestNodeTo(dhtMsg->getHashedKey())) { - logging_debug("Routing DHT message to closest node " - << " from " << msg->getSourceNode() - << " to " << dhtMsg->getHashedKey() - ); - dhtSend(msg, dhtMsg->getHashedKey()); - delete dhtMsg; - return true; - } - - // now, we are the closest node... - switch (msg->getType()) { - - // ----------------------------------------------------------------- put --- - case OverlayMsg::typeDHTPut: { - logging_debug("DHT-Put: Attempt to store values for key " - << dhtMsg->getKey()); - if (dhtMsg->doReplace()) { - logging_debug("DHT-Put: Attempt to replace key: remove old values first!"); - dht->remove(dhtMsg->getKey()); - } - BOOST_FOREACH( Data value, dhtMsg->getValues() ) { - logging_debug("DHT-Put: Stored value: " << value ); - dht->put(dhtMsg->getKey(), value, dhtMsg->getTTL() ); - } - break; - } - - // ----------------------------------------------------------------- get --- - case OverlayMsg::typeDHTGet: { - logging_info("DHT-Get: key=" << dhtMsg->getKey() ); - vector vect = dht->get(dhtMsg->getKey()); - BOOST_FOREACH(const Data& d, vect) - logging_info("DHT-Get: value=" << d); - OverlayMsg omsg(*msg); - omsg.swapRoles(); - omsg.setType(OverlayMsg::typeDHTData); - DHTMessage dhtmsg(dhtMsg->getKey(), vect); - omsg.encapsulate(&dhtmsg); - - logging_info("Sending DHT response to " << omsg.getDestinationNode()); - sendMessage(&omsg, omsg.getDestinationNode()); - break; - } - - // -------------------------------------------------------------- remove --- - case OverlayMsg::typeDHTRemove: { - if (dhtMsg->hasValues()) { - BOOST_FOREACH( Data value, dhtMsg->getValues() ) - dht->remove(dhtMsg->getKey(), value ); - } else - dht->remove( dhtMsg->getKey() ); - break; - } - - // -------------------------------------------------------------- default--- - default: - logging_error("DHT Message type unknown."); - return false; - } - delete dhtMsg; - return true; -} - -/// put a value to the DHT with a ttl given in seconds -void BaseOverlay::dhtPut( const Data& key, const Data& value, int ttl, bool replace, bool no_local_refresh ) { - - // log - logging_info("DHT-Put:" - << " key=" << key << " value=" << value - << " ttl=" << ttl << " replace=" << replace - ); - - if (!no_local_refresh) { - - // put into local data store (for refreshes) - if (replace) localDHT->remove(key); - localDHT->put(key, value, ttl); - } - - DHTMessage dhtmsg( key, value ); - dhtmsg.setReplace( replace ); - dhtmsg.setTTL(ttl); - - OverlayMsg msg( OverlayMsg::typeDHTPut ); - msg.encapsulate( &dhtmsg ); - msg.setSourceNode(this->nodeId); - dhtSend(&msg, dhtmsg.getHashedKey()); -} - -/// removes a key value pair from the DHT -void BaseOverlay::dhtRemove( const Data& key, const Data& value ) { - // remove from local data store - localDHT->remove(key,value); - - DHTMessage dhtmsg(key,value); - - // send message - OverlayMsg msg(OverlayMsg::typeDHTRemove); - msg.encapsulate( &dhtmsg ); - msg.setSourceNode(this->nodeId); - dhtSend(&msg, dhtmsg.getHashedKey()); -} - -/// removes all data stored at the given key -void BaseOverlay::dhtRemove( const Data& key ) { - // log: remove key - logging_info("DHT-Remove: Removing key=" << key ); - - DHTMessage dhtmsg(key); - - // send message - OverlayMsg msg(OverlayMsg::typeDHTRemove); - msg.encapsulate( &dhtmsg ); - msg.setSourceNode(this->nodeId); - dhtSend(&msg, dhtmsg.getHashedKey()); -} - -/// requests data stored using key -void BaseOverlay::dhtGet( const Data& key, const ServiceID& service ) { - // log: get - logging_info("DHT-Get: Trying to resolve key=" << - key << " for service=" << service.toString() ); - - DHTMessage dhtmsg(key); - - // send message - OverlayMsg msg(OverlayMsg::typeDHTGet); - msg.setService(service); - msg.encapsulate( &dhtmsg ); - msg.setSourceNode(this->nodeId); - dhtSend(&msg, dhtmsg.getHashedKey()); -} - -void BaseOverlay::dhtSend( OverlayMsg* msg, const NodeID& dest ) { - // log: dht send - logging_info("DHT-Send: Sending message with key=" << dest.toString() ); - - // local storage? yes-> put into DHT directly - if (overlayInterface->isClosestNodeTo(dest)) { - // be compatible with old code so set destination to hashed key - msg->setDestinationNode(dest); - - Data d = data_serialize(msg); - Message m2(d); - OverlayMsg* m3 = m2.decapsulate(); - - handleDHTMessage(m3); - - delete m3; - return; - } else { - // need to route - NodeID next_hop = overlayInterface->getNextNodeId(dest); - msg->setDestinationNode(next_hop); - logging_info("DHT-Send: sending via node " << next_hop.toString()); - - send(msg, next_hop); - - return; - } -} - std::string BaseOverlay::debugInformation() { std::stringstream s; Index: /source/ariba/overlay/BaseOverlay.h =================================================================== --- /source/ariba/overlay/BaseOverlay.h (revision 10647) +++ /source/ariba/overlay/BaseOverlay.h (revision 10653) @@ -74,5 +74,4 @@ using std::vector; using std::list; -using std::cout; using std::map; using std::make_pair; @@ -187,4 +186,12 @@ const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + /** + * send a message to the closest directly known node to an address + * + * @return NodeID of the (closest) destination node; + */ + NodeID sendMessageCloserToNodeID(const Message* message, const NodeID& address, + const ServiceID& service = OverlayInterface::OVERLAY_SERVICE_ID); + /** * Send out a message to all nodes that are known in the overlay structure. @@ -287,16 +294,4 @@ */ void leaveSpoVNet(); - - /// put a value to the DHT with a ttl given in seconds - void dhtPut( const Data& key, const Data& value, int ttl = 0, bool replace = false, bool no_local_refresh = false); - - /// removes a key value pair from the DHT - void dhtRemove( const Data& key, const Data& value ); - - /// removes all data stored at the given key - void dhtRemove( const Data& key ); - - /// requests data stored using key - void dhtGet( const Data& key, const ServiceID& service ); protected: @@ -411,7 +406,4 @@ bool handleJoinRequest( OverlayMsg* msg, const LinkID& bcLink ); bool handleJoinReply( OverlayMsg* msg, const LinkID& bcLink ); - - // handle DHT messages - bool handleDHTMessage( OverlayMsg* msg ); // handle link messages @@ -506,15 +498,4 @@ bool ignore_down = false ); - // distributed hashtable handling ------------------------------------------ - - DHT* dht; - DHT* localDHT; - int republishCounter; - - void initDHT(); - void destroyDHT(); - void stabilizeDHT(); - void dhtSend( OverlayMsg* msg, const NodeID& dest ); - // misc -------------------------------------------------------------------- Index: urce/ariba/overlay/messages/DHTMessage.cpp =================================================================== --- /source/ariba/overlay/messages/DHTMessage.cpp (revision 10647) +++ (revision ) @@ -1,44 +1,0 @@ -#include "DHTMessage.h" - -#include - -namespace ariba { -namespace overlay { - -vsznDefault(DHTMessage); - -DHTMessage::DHTMessage() : - ttl( 0 ), - replace( false ) -{} - -DHTMessage::DHTMessage( const Data& key ) : - ttl( 0 ), - replace( false ), - key( key.clone() ) -{} - -DHTMessage::DHTMessage( const Data& key, const Data& value ) : - ttl( 0 ), - replace( false ), - key( key.clone() ), - values(1, value.clone()) -{} - -DHTMessage::DHTMessage( const Data& key, const vector& values ) : - ttl( 0 ), - replace( false ), - key( key.clone() ) -{ - // preallocate enough room so we don't need to copy a lot - this->values.reserve(values.size()); - BOOST_FOREACH(const Data value, values ) - this->values.push_back( value.clone() ); -} - -DHTMessage::~DHTMessage() { - this->key.release(); - BOOST_FOREACH( Data& value, values ) value.release(); -} - -}} Index: urce/ariba/overlay/messages/DHTMessage.h =================================================================== --- /source/ariba/overlay/messages/DHTMessage.h (revision 10647) +++ (revision ) @@ -1,98 +1,0 @@ -#ifndef DHTMESSAGE_H_ -#define DHTMESSAGE_H_ - -#include "ariba/utility/messages.h" -#include "ariba/utility/serialization.h" - -namespace ariba { -namespace overlay { - -using ariba::utility::Message; -using_serialization; - -class DHTMessage : public Message { VSERIALIZEABLE -public: - DHTMessage(); - DHTMessage( const Data& key ); - DHTMessage( const Data& key, const Data& value ); - DHTMessage( const Data& key, const vector& values ); - virtual ~DHTMessage(); - - NodeID getHashedKey() const { - return NodeID::sha1( key.getBuffer(), key.getLength() / 8 ); - } - - const Data& getKey() const { - return key; - } - - /// returns the first element of the key vector - const Data& getValue() const { - return values.at(0); - } - - bool hasValues() const { - return values.size() != 0; - } - - uint16_t getTTL() const { - return ttl; - } - - void setTTL( uint16_t ttl ) { - this->ttl = ttl; - } - - void setReplace( bool replace ) { - this->replace = replace; - } - - bool doReplace() const { - return replace; - } - - /// return alles values for the key - const vector& getValues() const { - return values; - } - -private: - uint16_t ttl; - bool replace; - Data key; - vector values; -}; - -}} // namespace ariba::overlay - -sznBeginDefault( ariba::overlay::DHTMessage, X ) { - - // serialize flags - X && replace && cI(0,7); - - // serialize tll - X && ttl; - - // key serialization - uint16_t key_length = key.isUnspecified() ? 0 : key.getLength(); - X && key_length; - if (X.isDeserializer()) key.setLength( key_length ); - X && this->key; - - // store number of values - uint16_t num_values = values.size(); - X && num_values; - - // value serialization - for (size_t i=0; i scope_ids = ariba::transport::tcpip::get_interface_scope_ids(); + + BOOST_FOREACH ( uint64_t id, scope_ids ) + { + all_nodes.scope_id(id); + endp.address( all_nodes ); + + socket_v6.send_to( boost::asio::buffer(pnt, len), endp, 0, err ); + if(err) logging_warn("failed sending message through ipv6 socket"); + } } } Index: /source/ariba/utility/misc/Helper.h =================================================================== --- /source/ariba/utility/misc/Helper.h (revision 10647) +++ /source/ariba/utility/misc/Helper.h (revision 10653) @@ -67,5 +67,4 @@ using std::setfill; using std::setw; -using std::cout; using std::string; using std::ostream; Index: /source/ariba/utility/serialization/Data.hpp =================================================================== --- /source/ariba/utility/serialization/Data.hpp (revision 10647) +++ /source/ariba/utility/serialization/Data.hpp (revision 10653) @@ -41,4 +41,5 @@ //== library includes == +#include #include #include Index: /source/ariba/utility/system/Timer.cpp =================================================================== --- /source/ariba/utility/system/Timer.cpp (revision 10647) +++ /source/ariba/utility/system/Timer.cpp (revision 10653) @@ -104,5 +104,5 @@ void Timer::eventFunction() { - //std::cout << "unimplemented eventFunction Timer(" << millis << ")" << std::endl; + logging_warn("unimplemented eventFunction Timer(" << millis << ")"); } Index: urce/ariba/utility/transport/asio/asio_io_service.cpp =================================================================== --- /source/ariba/utility/transport/asio/asio_io_service.cpp (revision 10647) +++ (revision ) @@ -1,103 +1,0 @@ -// Internal version: Please do not publish! -// (... until released under FreeBSD-like license *g*) -// Code: Sebastian Mies - -#include "asio_io_service.h" - -#include -#include - -namespace ariba { -namespace transport { -namespace detail { - -using namespace boost::asio; -using namespace std; - -asio_io_service* asio_io_service::singleton = NULL; - -//#define DBG(x) cout << x << endl; -#define DBG(x) - -void asio_io_service::operator ()() { - running = true; - DBG("io_service started"); - boost::asio::io_service::work work(*service); - service->run(); - DBG("io_service stopped"); - if (destroy) { - delete singleton; - singleton = NULL; - DBG(cout << "asio io_service singleton destroyed" << endl); - } - running = false; -} - -asio_io_service::asio_io_service() : - references(1), running(false), destroy(false), thread(NULL), service(NULL) { - service = new io_service(); -} - -asio_io_service::~asio_io_service() { - if (running) { - service->stop(); - thread->join(); - } - if (thread != NULL) delete thread; - if (service != NULL) delete service; - thread = NULL; - service = NULL; -} - -void asio_io_service::internal_start() { - if (!running) { - if (thread != NULL) delete thread; - thread = new boost::thread(boost::ref(*this)); - } -} - -void asio_io_service::internal_stop() { - service->stop(); - singleton->running = false; -} - -io_service& asio_io_service::alloc() { - if (singleton != NULL) { - DBG("new asio io_service reference"); - singleton->references++; - return *singleton->service; - } else { - DBG("creating new asio io_service singleton"); - singleton = new asio_io_service(); - return *singleton->service; - } -} - -void asio_io_service::free() { - if (singleton != NULL) { - DBG("decreasing asio io_service reference"); - singleton->references--; - if (singleton->references == 0) { - DBG("request asio io_service destruction"); - if (singleton->running == false) { - delete singleton; - singleton = NULL; - DBG("asio io_service singleton destroyed"); - } else { - singleton->destroy = true; - singleton->service->stop(); - } - } - } -} - -void asio_io_service::start() { - singleton->internal_start(); -} - -void asio_io_service::stop() { - singleton->internal_stop(); -} - - -}}} // namespace ariba::transport::detail Index: urce/ariba/utility/transport/asio/asio_io_service.h =================================================================== --- /source/ariba/utility/transport/asio/asio_io_service.h (revision 10647) +++ (revision ) @@ -1,48 +1,0 @@ -// Internal version: Please do not publish! -// (... until released under FreeBSD-like license *g*) -// Code: Sebastian Mies - -#ifndef ASIO_IO_SERVICE_H_ -#define ASIO_IO_SERVICE_H_ - -#include -#include - -namespace ariba { -namespace transport { -namespace detail { - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -class asio_io_service { -private: - int references; - volatile bool running; - volatile bool destroy; - boost::thread* thread; - boost::asio::io_service* service; - - static asio_io_service* singleton; - - friend class boost::thread; - friend class boost::detail::thread_data >; - -protected: - void operator ()(); - asio_io_service(); - ~asio_io_service(); - void internal_start(); - void internal_stop(); -public: - static boost::asio::io_service& alloc(); - static void free(); - static void start(); - static void stop(); -}; - -}}} // namespace ariba::transport::detail - -#endif /* ASIO_IO_SERVICE_H_ */ Index: urce/ariba/utility/transport/asio/bluetooth_endpoint.hpp =================================================================== --- /source/ariba/utility/transport/asio/bluetooth_endpoint.hpp (revision 10647) +++ (revision ) @@ -1,186 +1,0 @@ -#include "ariba/config.h" - -#ifdef HAVE_LIBBLUETOOTH - -#ifndef BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ -#define BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ - -#include -#include - -#include - -namespace boost { -namespace asio { -namespace bluetooth { - -/** - * Describes an endpoint for a RFCOMM Bluetooth socket. - * - * @author Martin Florian - */ -template -class bluetooth_endpoint { -private: - static bdaddr_t addr_any; - -public: - /// The protocol type associated with the endpoint. - typedef BluetoothProtocol protocol_type; - - /// The type of the endpoint structure. This type is dependent on the - /// underlying implementation of the socket layer. - typedef boost::asio::detail::socket_addr_type data_type; // <-- Do I need this? - //typedef sockaddr_rc data_type; - - /// Default constructor. - bluetooth_endpoint() : - data_() { - data_.rc_family = AF_BLUETOOTH; - data_.rc_bdaddr = addr_any; - data_.rc_channel = (uint8_t) 0; - } - - bluetooth_endpoint(const BluetoothProtocol& protocol, - unsigned short channel) : - data_() { - data_.rc_family = AF_BLUETOOTH; - data_.rc_bdaddr = addr_any; - data_.rc_channel = channel; - } - - /// Construct an endpoint using a port number, specified in the host's byte - /// order. The IP address will be the any address (i.e. INADDR_ANY or - /// in6addr_any). This constructor would typically be used for accepting new - /// connections. - bluetooth_endpoint(unsigned short channel) : - data_() { - data_.rc_family = AF_BLUETOOTH; - data_.rc_bdaddr = *BDADDR_ANY; - data_.rc_channel = channel; - } - - /// Construct an endpoint using a port number and an BT address. - /// The address is in human readable form as a string. - bluetooth_endpoint(const char *addr, unsigned short channel) : - data_() { - data_.rc_family = AF_BLUETOOTH; - data_.rc_channel = channel; - str2ba(addr, &data_.rc_bdaddr); - } - - /// Construct an endpoint using a port number and an BT address. - /// The address is given in the bluetooth-internal format. - bluetooth_endpoint(bdaddr_t addr, unsigned short channel) : - data_() { - data_.rc_family = AF_BLUETOOTH; - data_.rc_channel = channel; - data_.rc_bdaddr = addr; - } - - /// Copy constructor. - bluetooth_endpoint(const bluetooth_endpoint& other) : - data_(other.data_) { - } - - /// Assign from another endpoint. - bluetooth_endpoint& operator=(const bluetooth_endpoint& other) { - data_ = other.data_; - return *this; - } - - /// The protocol associated with the endpoint. - protocol_type protocol() const { - return protocol_type::get(); - } - - /// Get the underlying endpoint in the native type. - /// TODO: make this nice and generic -> union like in tcp - data_type* data() { - return (boost::asio::detail::socket_addr_type*) &data_; - } - - /// Get the underlying endpoint in the native type. - const data_type* data() const { - return (boost::asio::detail::socket_addr_type*) &data_; - } - - /// Get the underlying size of the endpoint in the native type. - std::size_t size() const { - return sizeof(data_type); - } - - /// Set the underlying size of the endpoint in the native type. - void resize(std::size_t size) { - if (size > sizeof(data_type)) { - boost::system::system_error e(boost::asio::error::invalid_argument); - boost::throw_exception(e); - } - } - - /// Get the capacity of the endpoint in the native type. - std::size_t capacity() const { - return sizeof(data_type); - } - - /// Get the channel associated with the endpoint. The port number is always in - /// the host's byte order. - unsigned short channel() const { - return data_.rc_channel; - } - - /// Set the channel associated with the endpoint. The port number is always in - /// the host's byte order. - void channel(unsigned short channel_num) { - data_.rc_channel = channel_num; - } - - /// Get the Bluetooth address associated with the endpoint. - bdaddr_t address() const { - return data_.rc_bdaddr; - } - - /// Set the Bluetooth address associated with the endpoint. - void address(const boost::asio::ip::address& addr) { - bluetooth_endpoint tmp_endpoint(addr, channel()); - data_ = tmp_endpoint.data_; - } - - /// Get the Bluetooth address in human readable form and write it to buf. - void address_hr(char &buf) { - ba2str(&data_.rc_bdaddr, buf); - } - - /// Compare two endpoints for equality. - friend bool operator==(const bluetooth_endpoint& e1, - const bluetooth_endpoint& e2) { - return e1.address() == e2.address() && e1.channel() == e2.channel(); - } - - /// Compare two endpoints for inequality. - friend bool operator!=(const bluetooth_endpoint& e1, - const bluetooth_endpoint& e2) { - return e1.address() != e2.address() || e1.channel() != e2.channel(); - } - - /// Compare endpoints for ordering. - friend bool operator<(const bluetooth_endpoint& e1, - const bluetooth_endpoint& e2) { - if (e1.address() < e2.address()) return true; - if (e1.address() != e2.address()) return false; - return e1.channel() < e2.channel(); - } - -private: - // The underlying rfcomm socket address structure thingy. - //struct data_type data_; - struct sockaddr_rc data_; -}; - -template -bdaddr_t bluetooth_endpoint::addr_any = { {0u, 0u, 0u, 0u, 0u, 0u} }; - -}}} // namespace boost::asio::bluetooth - -#endif -#endif Index: urce/ariba/utility/transport/asio/rfcomm.hpp =================================================================== --- /source/ariba/utility/transport/asio/rfcomm.hpp (revision 10647) +++ (revision ) @@ -1,54 +1,0 @@ -#ifndef BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ -#define BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ - -#include "bluetooth_endpoint.hpp" - -#include -#include - -namespace boost { -namespace asio { -namespace bluetooth { - -/** - * The rfcomm class contains flags necessary for RFCOMM sockets. - * - * @author Martin Florian - */ -class rfcomm { -public: - /// The type of endpoint. - typedef bluetooth_endpoint endpoint; - - /// Get an Instance. - /// We need this to fulfill the asio Endpoint requirements, I think. - static rfcomm get() { - return rfcomm(); - } - - /// Obtain an identifier for the type of the protocol. - int type() const { - return SOCK_STREAM; - } - - /// Obtain an identifier for the protocol. - int protocol() const { - return BTPROTO_RFCOMM; - } - - /// Obtain an identifier for the protocol family. - int family() const { - return AF_BLUETOOTH; - } - - /// The RFCOMM socket type, lets pray that this will work. - typedef basic_stream_socket socket; - - /// The RFCOMM acceptor type. - typedef basic_socket_acceptor acceptor; - -}; - -}}} // namespace boost::asio::bluetooth - -#endif /* BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ */ Index: /source/ariba/utility/transport/asio/unique_io_service.cpp =================================================================== --- /source/ariba/utility/transport/asio/unique_io_service.cpp (revision 10653) +++ /source/ariba/utility/transport/asio/unique_io_service.cpp (revision 10653) @@ -0,0 +1,152 @@ +// [License] +// The Ariba-Underlay Copyright +// +// Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +// +// Institute of Telematics +// UniversitÀt Karlsruhe (TH) +// Zirkel 2, 76128 Karlsruhe +// Germany +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// 1. Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// 2. Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR +// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +// The views and conclusions contained in the software and documentation +// are those of the authors and should not be interpreted as representing +// official policies, either expressed or implied, of the Institute of +// Telematics. +// [License] + +/** + * @author Mario Hock , Michael TÀnzer + */ + +#include "unique_io_service.h" + +#include +#include + +namespace ariba { +namespace transport { +namespace detail { + +using namespace std; + + +/* init static members */ +boost::asio::io_service* unique_io_service::asio_io_service = NULL; +int unique_io_service::asio_io_service_ref_count = 0; +boost::thread* unique_io_service::thread = NULL; +int unique_io_service::run_count = 0; +boost::mutex unique_io_service::thread_stopped_mutex; +boost::condition_variable unique_io_service::thread_stopped_cond; +bool unique_io_service::thread_stopped = true; + + +unique_io_service::unique_io_service(): + running(false) +{ + if ( asio_io_service_ref_count == 0 ) + { + asio_io_service = new boost::asio::io_service(); + } + + asio_io_service_ref_count++; +} + + + +unique_io_service::~unique_io_service() +{ + assert( !running ); + + asio_io_service_ref_count--; + + if ( asio_io_service_ref_count <= 0 ) + { + delete asio_io_service; + asio_io_service = NULL; + } +} + + + +boost::asio::io_service & unique_io_service::get_asio_io_service() +{ + assert(asio_io_service != NULL ); + + return *asio_io_service; +} + + + +void unique_io_service::start() +{ + assert(!running); + + boost::mutex::scoped_lock lock(thread_stopped_mutex); + if (thread_stopped) + { + thread_stopped = false; + thread = new boost::thread(&unique_io_service::thread_function); + } + + run_count++; + running = true; +} + + + +void unique_io_service::stop() +{ + assert( running ); + + running = false; + run_count--; + + if ( run_count == 0 ) + { + asio_io_service->stop(); + + boost::mutex::scoped_lock lock(thread_stopped_mutex); + while (!thread_stopped) + { + thread_stopped_cond.wait(lock); + } + delete thread; + } +} + + +void unique_io_service::thread_function() +{ + asio_io_service->run(); + + { + boost::mutex::scoped_lock lock(thread_stopped_mutex); + thread_stopped = true; + } + thread_stopped_cond.notify_one(); +} + + +}}} // namespace ariba::transport::detail Index: /source/ariba/utility/transport/asio/unique_io_service.h =================================================================== --- /source/ariba/utility/transport/asio/unique_io_service.h (revision 10653) +++ /source/ariba/utility/transport/asio/unique_io_service.h (revision 10653) @@ -0,0 +1,101 @@ +// [License] +// The Ariba-Underlay Copyright +// +// Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +// +// Institute of Telematics +// UniversitÀt Karlsruhe (TH) +// Zirkel 2, 76128 Karlsruhe +// Germany +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are +// met: +// +// 1. Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// 2. Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR +// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +// +// The views and conclusions contained in the software and documentation +// are those of the authors and should not be interpreted as representing +// official policies, either expressed or implied, of the Institute of +// Telematics. +// [License] + + +#ifndef UNIQUE_IO_SERVICE_H_ +#define UNIQUE_IO_SERVICE_H_ + +#include +#include +#include + +namespace ariba { +namespace transport { +namespace detail { + +/** + * Multiplexes an asio_io_service so it can be used by multiple components. + * This class is _not_ threadsafe (use only from ariba thread/system queue) + * + * @author Mario Hock , Michael TÀnzer + */ +class unique_io_service : public boost::noncopyable +{ +public: + unique_io_service(); + ~unique_io_service(); + + /** + * Returned io_service is only valid during the lifetime of the + * unique_io_service object. + */ + boost::asio::io_service& get_asio_io_service(); + + /** + * Executes the io_service::run() method in a parallel thread. + * Multiple calls don't lead to multiple threads. + */ + void start(); + + /** + * Make the io_service::run() method return and stop thread if this is the + * last stop() call (number of start() calls == number of stop() calls) + */ + void stop(); + +private: + static void thread_function(); + + +private: + static boost::asio::io_service* asio_io_service; + static int asio_io_service_ref_count; + + static boost::thread* thread; + static int run_count; + + static boost::mutex thread_stopped_mutex; + static boost::condition_variable thread_stopped_cond; + static bool thread_stopped; + + bool running; +}; + +}}} // namespace ariba::transport::detail + +#endif /* UNIQUE_IO_SERVICE_H_ */ Index: /source/ariba/utility/transport/messages/CHANGELOG =================================================================== --- /source/ariba/utility/transport/messages/CHANGELOG (revision 10653) +++ /source/ariba/utility/transport/messages/CHANGELOG (revision 10653) @@ -0,0 +1,16 @@ +Changelog of the reboost buffers + +2012-05-09; author: Sebastian Mies ++ added compression (miniz) to tlv + +2012-05-08; author: Sebastian Mies ++ fixed shared_buffer_t::make_unique() ++ added shared_buffer_t::resize() +o made message_t::operator() const ++ added first type-length-value (tlv) implementations based on the buffers + +2012-02-07; author: Sebastian Mies ++ Message compaction when no. of buffers are depleted ++ Shared buffer concatenation using operator+ ++ FreeBSD license header + Index: /source/ariba/utility/transport/messages/LICENSE =================================================================== --- /source/ariba/utility/transport/messages/LICENSE (revision 10653) +++ /source/ariba/utility/transport/messages/LICENSE (revision 10653) @@ -0,0 +1,21 @@ +Copyright (c) 2012, Sebastian Mies . All rights reserved. + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +Redistributions of source code must retain the above copyright notice, this +list of conditions and the following disclaimer. Redistributions in binary form +must reproduce the above copyright notice, this list of conditions and the +following disclaimer in the documentation and/or other materials provided +with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE +USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. Index: /source/ariba/utility/transport/messages/buffer.cpp =================================================================== --- /source/ariba/utility/transport/messages/buffer.cpp (revision 10653) +++ /source/ariba/utility/transport/messages/buffer.cpp (revision 10653) @@ -0,0 +1,42 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#include +#include + +#include "buffer.hpp" + +namespace reboost { + +std::ostream& operator<<( std::ostream& os, const buffer_t& buf ) { + os << "{"; + for (unsigned i=0; i<(buf.size()/8)+1; i++) { + if (i!=0) os << ","; + os << "["; + for (unsigned j=0; j<8; j++) { + if ((i*8+j) < buf.size()) { + if (j!=0) os << " "; + char str[8]; + int v = buf.data()[i*8+j]; + sprintf(str,"%02x", v); + os << str; + } + } + os << "|"; + for (unsigned j=0; j<8; j++) { + if (i*8+j < buf.size()) { + char c = buf.data()[i*8+j]; + if (c<32) c='.'; + os << c; + } + } + os <<"]"; + } + os << "}"; + return os; +} + +} /* namespace reboost */ Index: /source/ariba/utility/transport/messages/buffer.hpp =================================================================== --- /source/ariba/utility/transport/messages/buffer.hpp (revision 10653) +++ /source/ariba/utility/transport/messages/buffer.hpp (revision 10653) @@ -0,0 +1,198 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#ifndef BUFFER_HPP_ +#define BUFFER_HPP_ + +#include +#include + +namespace reboost { + +/// buffer size and octet +typedef size_t bsize_t; +typedef unsigned char boctet_t; + +/** + * A simple buffer. + * + * @author Sebastian Mies + */ +class buffer_t { + typedef buffer_t self; +protected: + bsize_t size_; + boctet_t * data_; + +public: + /// a buffer of zero size, pointing to NULL + inline buffer_t() : + size_(0), data_((boctet_t*) 0) { + } + + /// copy constructor + inline buffer_t(const self& rhs) : + size_(rhs.size_), data_(rhs.data_){ + } + + /// allocate a buffer of a certain size + inline buffer_t(bsize_t size) : + size_(size), data_(new boctet_t[size]) { + } + + /// convenience constructor + inline buffer_t(bsize_t size, boctet_t* data) : + size_(size), data_(data) { + } + + /// interpret string as buffer (maximal size 64K) + inline explicit buffer_t(const char* string) : + size_(0), data_((boctet_t*) string) { + while (string[size_] != 0 && size_ < 64000) + size_++; + } + + /// clone a buffer + inline self clone() const { + buffer_t n(this->size_); + for (bsize_t i = 0; i < size_; i++) + n.data_[i] = this->data_[i]; + return n; + } + + /// copy a buffer pointer and size. + inline self& operator=(const self& rhs) { + this->size_ = rhs.size_; + this->data_ = rhs.data_; + return *this; + } + + /// return sub-buffer. + inline self operator()(bsize_t index, bsize_t size = 0) const { + self n(*this); + n.data_ += index; + if (size == 0) n.size_ -= index; + else n.size_ = size; + return n; + } + + /// returns a data element at given index + inline boctet_t operator[](bsize_t index) const { + return data_[index]; + } + + /// returns a data element at given index + inline boctet_t& operator[](bsize_t index) { + return data_[index]; + } + + /// assigns contents to another buffer + inline const self& copy_to( self& buf, bsize_t index = 0 ) const { + assert(index + size_ <= buf.size_); + const boctet_t *src=data_; + boctet_t* dst=buf.data_+index; + for (bsize_t i=0; i inline const T* cast_to() { + return (T*)data_; + } + + /// set buffer size + inline void size( bsize_t size ) { + size_ = size; + } + + + /// returns the size in octets + inline bsize_t size() const { + return size_; + } + + /// sets the data pointer + inline void data( boctet_t* data ) { + data_ = data; + } + + /// returns a pointer to immutable data + inline const boctet_t * data() const { + return data_; + } + + /// returns a pointer to mutable data + inline boctet_t * mutable_data() { + return data_; + } + + /// returns a hash value of this buffer using the ELF hash + inline size_t hash() const { + unsigned h = 0, g; + for (size_t i = 0; i < size_; i++) { + h = (h << 4) + data_[i]; + g = h & 0xf0000000L; + if (g != 0) h ^= g >> 24; + h &= ~g; + } + return (size_t) h; + } + + /// returns true if data_ is null + inline bool is_null() const { + return data_ == NULL; + } + + /// returns true if size is zero + inline bool is_empty() const { + return size_ == 0; + } + + /// compare two buffers + inline int compare_to(const self& rhs) const { + if (rhs.data_ == data_ && rhs.size_ == size_) return 0; + if (size_ < rhs.size_) return -1; + if (size_ > rhs.size_) return 1; + for (bsize_t i = 0; i < size_; i++) { + if (data_[i] > rhs.data_[i]) return 1; + if (data_[i] < rhs.data_[i]) return -1; + } + return 0; + } + + /// convenience + inline bool operator==(const self& rhs) const { + return (compare_to(rhs) == 0); + } + inline bool operator!=(const self& rhs) const { + return (compare_to(rhs) != 0); + } + inline bool operator<(const self& rhs) const { + return (compare_to(rhs) < 0); + } + inline bool operator<=(const self& rhs) const { + return (compare_to(rhs) <= 0); + } + inline bool operator>(const self& rhs) const { + return (compare_to(rhs) > 0); + } + inline bool operator>=(const self& rhs) const { + return (compare_to(rhs) >= 0); + } +}; + +/// stream operator +std::ostream& operator<<( std::ostream& os, const buffer_t& buf ); + +} + +/// boost hash +namespace boost { +inline size_t hash_value(const reboost::buffer_t& buf) { + return buf.hash(); +}} + +#endif /* BUFFER_HPP_ */ Index: /source/ariba/utility/transport/messages/buffers.hpp =================================================================== --- /source/ariba/utility/transport/messages/buffers.hpp (revision 10653) +++ /source/ariba/utility/transport/messages/buffers.hpp (revision 10653) @@ -0,0 +1,13 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- +#ifndef REBOOST_BUFFERS_HPP_ +#define REBOOST_BUFFERS_HPP_ + +#include "buffer.hpp" +#include "shared_buffer.hpp" +#include "message.hpp" + +#endif /* REBOOST_BUFFERS_HPP_ */ Index: /source/ariba/utility/transport/messages/message.cpp =================================================================== --- /source/ariba/utility/transport/messages/message.cpp (revision 10653) +++ /source/ariba/utility/transport/messages/message.cpp (revision 10653) @@ -0,0 +1,31 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#include "message.hpp" +#include + +namespace reboost { + +struct to_stream { + std::ostream& os; + int i; + inline void operator()(buffer_t buf) { + if (i!=0) os <<","; + os << buf; + i++; + } +}; + +std::ostream& operator<<(std::ostream& os, const message_t m) { + struct to_stream ts = { os, 0 }; + os << "message({size=" << m.size() << ",buffers=" << (int) m.length() + << ",hash=" << m.hash() << "},"; + m.foreach(ts); + os << ")"; + return os; +} + +} Index: /source/ariba/utility/transport/messages/message.hpp =================================================================== --- /source/ariba/utility/transport/messages/message.hpp (revision 10653) +++ /source/ariba/utility/transport/messages/message.hpp (revision 10653) @@ -0,0 +1,362 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#ifndef REBOOST_MESSAGE_HPP_ +#define REBOOST_MESSAGE_HPP_ + +#include +#include +#include + +#include "shared_buffer.hpp" + +namespace reboost { + +/// message size type +typedef signed char mlength_t; + +/// maximum number of buffers per message (default is 8) +const mlength_t message_max_buffers = (1L << 3); + +//! A Copy-on-Write Message with Shared Buffers. +/** + * A Copy-on-Write Message with Shared Buffers. + * + * A message holds a limited (defined by message_max_buffers) + * number of shared buffers. One can add new buffers and messages in front and + * at the end of a message. If the no. of buffers exceed + * message_max_buffers, then the two smallest successive buffers + * are compacted to one buffer. + * + * @author Sebastian Mies + */ +class message_t { +private: + // read sub-message + struct sub_message { + message_t* msg; + inline void operator()(shared_buffer_t buf) { + msg->push_back(buf); + } + }; + + // read from buffer + struct read_buffer { + boctet_t* buffer; + inline void operator()(buffer_t buf) { + memcpy((void*) buffer, (void*) buf.data(), buf.size()); + buffer += buf.size(); + } + }; + + // write to buffer + struct write_buffer { + const boctet_t* buffer; + inline void operator()(buffer_t buf) { + memcpy((void*) buf.data(), (void*) buffer, buf.size()); + buffer += buf.size(); + } + }; + +public: + /// Create a new message + inline message_t() : + imsg() { + } + + /// Copy message + inline message_t(const message_t& msg) : + imsg(msg.imsg) { + imsg->owner = NULL; + } + + /// Linearize message + inline operator shared_buffer_t() const { + return linearize(); + } + + /// Assign another message + inline message_t& operator=(const message_t& msg) { + msg.imsg->owner = NULL; + imsg = msg.imsg; + return *this; + } + + /// Adds a shared buffer of given site at the end + inline shared_buffer_t& push_back( bsize_t size ) { + shared_buffer_t b(size); push_back(b); + return imsg->at(-1); + } + + /// Adds a buffer at the end of the message + inline void push_back(const shared_buffer_t& buf) { + own().push_back(buf); + } + + /// Adds a message at the end of the message + inline void push_back(const message_t& msg) { + own(); + for (mlength_t i = 0; i < msg.length(); i++) + push_back(msg[i]); + } + + /// Adds a shared buffer of given size at the front + inline shared_buffer_t& push_front( bsize_t size ) { + shared_buffer_t b(size); push_front(b); + return imsg->at(0); + } + + /// Adds a buffer at the front of the messsage + inline void push_front(const shared_buffer_t& buf) { + own().push_front(buf); + } + + /// Adds a message at the end of the message + inline void push_front(const message_t& msg) { + own(); + for (mlength_t i = msg.length() - 1; i != 0; i--) + push_front(msg[i]); + } + + /// Removes a buffer from the end of the message + inline shared_buffer_t pop_back() { + return own().pop_back(); + } + + /// Removes a buffer from the beginning of this message. + inline shared_buffer_t pop_front() { + return own().pop_front(); + } + + /// Returns the size of the message in bytes (or octets). + inline size_t size() const { + size_t s = 0; + for (mlength_t i = 0; i < length(); i++) + s += operator[](i).size(); + return (s); + } + + /// Returns the number of buffers inside this message. + inline mlength_t length() const { + return (imsg->length); + } + + /// Returns the buffer at the given index. + inline shared_buffer_t& operator[](mlength_t idx) { + return at(idx); + } + + /// Returns the buffer at the given index. + inline shared_buffer_t& at(mlength_t idx) { + return imsg->at(idx); + } + + /// Returns the constant buffer at the given index. + inline const shared_buffer_t& operator[](mlength_t idx) const { + return at(idx); + } + + /// Returns the buffer at the given index + inline const shared_buffer_t& at(mlength_t idx) const { + return imsg->at(idx); + } + + /// Iterates over a partial set of buffers. + template + inline void foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const { + T op = work; + if (size_ == 0) size_ = size() - index_; + + // get first buffer + mlength_t f = 0, pf = 0; + for (; (pf + at(f).size()) <= index_ && f < imsg->length; + pf += at(f).size(), f++); + // get last buffer + mlength_t l = f, pl = pf; + for (; (pl + at(l).size()) < (index_ + size_) && l < imsg->length; + pl += at(l).size(), l++); + + // same buffer? yes-> get sub-buffer + if (l == f) op(at(l)(index_ - pf, size_)); + else { // no-> get sub-buffers :) + op(at(f)(index_ - pf)); + for (mlength_t i = f + 1; i < l; i++) op(at(i)); + op(at(l)(0, index_ + size_ - pl)); + } + } + + /// Read bytes (gather). + inline void read(boctet_t* mem, size_t idx = 0, size_t size_ = 0) const { + struct read_buffer rb = { mem }; + foreach(rb, idx, size_); + } + + /// write bytes + inline void write(const boctet_t* mem, size_t idx = 0, size_t size_ = 0) { + struct write_buffer wb = { mem }; + foreach(wb, idx, size_); + } + + /// Read an arbitrary, binary object. + template + inline T read(size_t index) { + T obj; + read((boctet_t*) &obj, index, sizeof(T)); + return obj; + } + + /// Write an arbitrary, binary object. + template + inline void write(const T& value, size_t index) { + write((boctet_t*) &value, index, sizeof(T)); + } + + /// Calculate a (ELF-like) hash. + inline size_t hash() const { + size_t h = 0; + for (mlength_t i = 0; i < length(); i++) + h += at(i).hash() * (i + 1); + return h; + } + + /// Returns a sub-message. + message_t operator()(size_t index, size_t size = 0) const { + message_t m; + struct sub_message sm = { &m }; + foreach(sm, index, size); + return m; + } + + /// Linearizes the complete/partial message into one shared buffer. + inline shared_buffer_t linearize(size_t index = 0, size_t size_ = 0) const { + shared_buffer_t b(size_ == 0 ? size() : size_); + read(b.mutable_data(), index, size_); + return b; + } + +private: + class imsg_t { + public: + volatile message_t* owner; + shared_buffer_t buffers[message_max_buffers]; + mlength_t index, length; + public: + inline imsg_t() : + index(0), length(0) { + } + inline imsg_t(const imsg_t& imsg) : + index(imsg.index), length(imsg.length) { + for (mlength_t i = 0; i < length; i++) + at(index + i) = imsg.at(index + i); + } + inline shared_buffer_t& at(mlength_t idx) { + if (idx < 0) idx += length; + return buffers[(idx + index) & (message_max_buffers - 1)]; + } + inline const shared_buffer_t& at(mlength_t idx) const { + if (idx < 0) idx += length; + return buffers[(idx + index) & (message_max_buffers - 1)]; + } + + inline void push_back(const shared_buffer_t& buf) { + if (buf.size() == 0) return; + if (length == message_max_buffers) compact(); + at(length) = buf; + length++; + } + + inline void push_front(const shared_buffer_t& buf) { + if (buf.size() == 0) return; + if (length == message_max_buffers) compact(); + index--; + length++; + at(0) = buf; + } + + inline shared_buffer_t pop_back() { + shared_buffer_t& buf = at(-1); + shared_buffer_t ret = buf; + buf.reset(); + length--; + return ret; + } + + inline shared_buffer_t pop_front() { + shared_buffer_t& buf = at(0); + shared_buffer_t ret = buf; + buf.reset(); + length--; + index++; + return ret; + } + + /// compacts the buffers, so one more buffer is available + inline void compact() { + + // find compacting candidate + bsize_t min_size=~0, min_pos=0; + for (mlength_t i=0; iowner == this) return *imsg; + if (imsg.get() == NULL) imsg = boost::shared_ptr(new imsg_t()); + else imsg = boost::shared_ptr(new imsg_t(*imsg)); + imsg->owner = this; + return *imsg; + } + boost::shared_ptr imsg; +}; + +inline message_t operator+(const message_t& lhs, const message_t& rhs) { + message_t m = lhs; + m.push_back(rhs); + return m; +} + +inline message_t operator+(const message_t& lhs, const buffer_t& rhs) { + message_t m = lhs; + m.push_back(rhs); + return m; +} + +inline message_t operator+(const shared_buffer_t& lhs, const message_t& rhs) { + message_t m = rhs; + m.push_front(lhs); + return m; +} + +inline message_t operator+(const shared_buffer_t& lhs, + const shared_buffer_t& rhs) { + message_t m; + m.push_back(lhs); + m.push_back(rhs); + return m; +} + +std::ostream& operator<<(std::ostream&, const message_t); + +} /* namespace reboost */ + +#endif /* REBOOST_MESSAGE_HPP_ */ Index: /source/ariba/utility/transport/messages/shared_buffer.cpp =================================================================== --- /source/ariba/utility/transport/messages/shared_buffer.cpp (revision 10653) +++ /source/ariba/utility/transport/messages/shared_buffer.cpp (revision 10653) @@ -0,0 +1,31 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#include "shared_buffer.hpp" +#include + +namespace reboost { + +using namespace std; + +#ifdef DEBUG + +void shared_buffer_t::onexit() { + if (allocated_buffers != 0) + cerr << "shared_buffer_t: " << allocated_buffers << " leaked buffers." << endl; +} + +size_t shared_buffer_t::init() { + atexit(&shared_buffer_t::onexit); + return 0; +} + +size_t shared_buffer_t::allocated_buffers = init(); +boost::mutex allocated_buffers_mutex; + +#endif //DEBUG + +} Index: /source/ariba/utility/transport/messages/shared_buffer.hpp =================================================================== --- /source/ariba/utility/transport/messages/shared_buffer.hpp (revision 10653) +++ /source/ariba/utility/transport/messages/shared_buffer.hpp (revision 10653) @@ -0,0 +1,196 @@ +//----------------------------------------------------------------------------- +// Part of reboost (http://reboost.org). Released under the +// BSD 2-clause license (http://www.opensource.org/licenses/bsd-license.php). +// Copyright 2012, Sebastian Mies --- All rights reserved. +//----------------------------------------------------------------------------- + +#ifndef REBOOST_SHARED_BUFFER_HPP_ +#define REBOOST_SHARED_BUFFER_HPP_ + +#include +#include + +#ifdef DEBUG +#include +#endif + +#include "buffer.hpp" + +namespace reboost { + +/** + * A simple shared buffer. + * + * Important: if not shared, a buffer is writable. After the buffer is shared, + * the buffer is immutable. It uses shared_ptr/default allocators and prints + * error messages to cerr if buffers leaked at the end of the + * program. + * + * @author Sebastian Mies + */ +class shared_buffer_t: public buffer_t { + typedef shared_buffer_t self; +private: + static void onexit(); + static size_t init(); + +#ifdef DEBUG + static size_t allocated_buffers; + static boost::mutex allocated_buffers_mutex; +#endif + + class deleteable_buffer: public buffer_t { + public: + deleteable_buffer(bsize_t size) : + buffer_t(size) + { + +#ifdef DEBUG + boost::mutex::scoped_lock lock(allocated_buffers_mutex); + allocated_buffers++; +#endif + + } + + deleteable_buffer(boctet_t* data, bsize_t size) : + buffer_t(size, data) + { + +#ifdef DEBUG + boost::mutex::scoped_lock lock(allocated_buffers_mutex); + allocated_buffers++; +#endif + + } + + ~deleteable_buffer() { + if (!is_null()) delete[] data_; + +#ifdef DEBUG + boost::mutex::scoped_lock lock(allocated_buffers_mutex); + allocated_buffers--; +#endif + + } + }; + boost::shared_ptr parent; + +public: + /// create null-buffer + inline shared_buffer_t() : + buffer_t(), parent() { + } + + /// create a shared buffer of a specific size + inline shared_buffer_t(bsize_t size) : + buffer_t(), parent(new deleteable_buffer(size)) { + buffer_t::operator=(*parent); + } + + /// create shared buffer from string + inline shared_buffer_t(const char* string) : + buffer_t(), parent() { + parent.reset(new deleteable_buffer(strlen(string))); + memcpy(parent->mutable_data(),string,parent->size()); + data(parent->mutable_data()); size(parent->size()); + } + + /// create shared buffer from buffer + /// ----> ALTERED BEHAVIOUR: + /// ----> This constructor transfers ownership rater than copying the data. + inline shared_buffer_t(boctet_t* buffer, bsize_t size) : + buffer_t(size, buffer), + parent(new deleteable_buffer(buffer, size)) + { +// memcpy(parent->mutable_data(), buffer, parent->size()); +// data(parent->mutable_data()); +// this->size(parent->size()); + } + + /// clone data from a normal buffer + inline shared_buffer_t(const buffer_t& rhs) : + buffer_t(), parent(new deleteable_buffer(rhs.size())) { + buffer_t::operator=(*parent); + rhs.copy_to(*parent, 0); + } + + /// copy constructor + inline shared_buffer_t(const self& rhs) : + buffer_t(rhs), parent(rhs.parent) { + } + + /// copy a buffer pointer and size. + inline self& operator=(const self& rhs) { + parent = rhs.parent; + buffer_t::operator=(rhs); + return (*this); + } + + /// return sub-buffer. + inline self operator()(bsize_t index, bsize_t size = 0) const { + self n(*this); + n.data_ += index; + if (size == 0) n.size_ -= index; + else n.size_ = size; + return (n); + } + + /// returns the number of references + inline size_t use_count() const { + return (parent.use_count()); + } + + /// returns true, if this is root buffer + inline bool is_root() const { + return ((parent.get() != NULL) && (this->data() == parent->data()) && ( + this->size() == parent->size())); + } + + /// returns true, if the buffer is not shared + inline bool is_unique() const { + return parent.unique(); + } + + /// makes the buffer unique / not shared by copying the contents + inline void make_unique() { + if (is_unique()) return; + boost::shared_ptr old = parent; + parent.reset(new deleteable_buffer(old->size())); + buffer_t::data(parent->mutable_data()); + old->copy_to(*parent,0); + } + + /// resize the shared buffer (and make unique) + inline void resize( bsize_t new_size ) { + boost::shared_ptr old = parent; + parent.reset(new deleteable_buffer(new_size)); + buffer_t::data(parent->mutable_data()); + (*old)(0,new_size).copy_to(*parent,0); + size_ = new_size; + } + + /// returns a pointer to mutable data, if shared_buffer is not shared yet. + inline boctet_t * mutable_data() { + assert(parent.unique()); + return buffer_t::mutable_data(); + } + + /// drop reference to buffer + inline void reset() { + parent.reset(); + data(NULL); + size(0); + } + + /// concatenate buffers + self operator+( const buffer_t& rhs ) const { + self nb(size()+rhs.size()); + this->copy_to(nb,0); + rhs.copy_to(nb,size()); + return nb; + } +}; + +} /* namespace reboost */ + +#endif /* REBOOST_SHARED_BUFFER_HPP_ */ Index: /source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp =================================================================== --- /source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp (revision 10653) +++ /source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp (revision 10653) @@ -0,0 +1,195 @@ +#include "ariba/config.h" + +#ifdef HAVE_LIBBLUETOOTH + +#ifndef BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ +#define BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ + +#include +#include +#include + +#include + +namespace boost { +namespace asio { +namespace bluetooth { + +/** + * Describes an endpoint for a RFCOMM Bluetooth socket. + * + * @author Martin Florian + */ +template +class bluetooth_endpoint { +private: + static const bdaddr_t addr_any; + +public: + /// The protocol type associated with the endpoint. + typedef BluetoothProtocol protocol_type; + + /// The type of the endpoint structure. This type is dependent on the + /// underlying implementation of the socket layer. + typedef boost::asio::detail::socket_addr_type data_type; // <-- Do I need this? + //typedef sockaddr_rc data_type; + + /// Default constructor. + bluetooth_endpoint() : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = 0; + } + + bluetooth_endpoint(const BluetoothProtocol& protocol, + uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = channel; + } + + /// Construct an endpoint using a port number, specified in the host's byte + /// order. The IP address will be the any address (i.e. INADDR_ANY or + /// in6addr_any). This constructor would typically be used for accepting new + /// connections. + bluetooth_endpoint(uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = channel; + } + + /// Construct an endpoint using a port number and an BT address. + /// The address is in human readable form as a string. + bluetooth_endpoint(const char *addr, uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_channel = channel; + str2ba(addr, &data_.rc_bdaddr); + } + + /// Construct an endpoint using a port number and an BT address. + /// The address is given in the bluetooth-internal format. + bluetooth_endpoint(const bdaddr_t& addr, uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_channel = channel; + data_.rc_bdaddr = addr; + } + + /// Copy constructor. + bluetooth_endpoint(const bluetooth_endpoint& other) : + data_(other.data_) { + } + + /// Assign from another endpoint. + bluetooth_endpoint& operator=(const bluetooth_endpoint& other) { + data_ = other.data_; + return *this; + } + + /// The protocol associated with the endpoint. + protocol_type protocol() const { + return protocol_type::get(); + } + + /// Get the underlying endpoint in the native type. + /// TODO: make this nice and generic -> union like in tcp + data_type* data() { + return (boost::asio::detail::socket_addr_type*) &data_; + } + + /// Get the underlying endpoint in the native type. + const data_type* data() const { + return (boost::asio::detail::socket_addr_type*) &data_; + } + + /// Get the underlying size of the endpoint in the native type. + std::size_t size() const { + return sizeof(data_type); + } + + /// Set the underlying size of the endpoint in the native type. + void resize(std::size_t size) { + if (size > sizeof(data_type)) { + boost::system::system_error e(boost::asio::error::invalid_argument); + boost::throw_exception(e); + } + } + + /// Get the capacity of the endpoint in the native type. + std::size_t capacity() const { + return sizeof(data_type); + } + + /// Get the channel associated with the endpoint. The port number is always in + /// the host's byte order. + uint8_t channel() const { + return data_.rc_channel; + } + + /// Set the channel associated with the endpoint. The port number is always in + /// the host's byte order. + void channel(uint8_t channel_num) { + data_.rc_channel = channel_num; + } + + /// Get the Bluetooth address associated with the endpoint. + const bdaddr_t& address() const { + return data_.rc_bdaddr; + } + + /// Set the Bluetooth address associated with the endpoint. + void address(const boost::asio::ip::address& addr) { + bluetooth_endpoint tmp_endpoint(addr, channel()); + data_ = tmp_endpoint.data_; + } + + /// Compare two endpoints for equality. + friend bool operator==(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + return e1.address() == e2.address() && e1.channel() == e2.channel(); + } + + /// Compare two endpoints for inequality. + friend bool operator!=(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + return e1.address() != e2.address() || e1.channel() != e2.channel(); + } + + /// Compare endpoints for ordering. + friend bool operator<(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + int cmp = bacmp(&e1.address(), &e2.address()); + + if (cmp == 0) { + return e1.channel() < e2.channel(); + } else { + return cmp < 0; + } + } + + friend ostream& operator<<( + ostream& out, + const bluetooth_endpoint& endp) { + char* mac_str = batostr(&endp.data_.rc_bdaddr); + out << "[" << mac_str << "]:" << (int)endp.data_.rc_channel; + free(mac_str); + return out; + } + +private: + // The underlying rfcomm socket address structure thingy. + //struct data_type data_; + struct sockaddr_rc data_; +}; + +template +const bdaddr_t bluetooth_endpoint::addr_any = { {0u, 0u, 0u, 0u, 0u, 0u} }; + +}}} // namespace boost::asio::bluetooth + +#endif /* HAVE_LIBBLUETOOTH */ +#endif /* BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ */ Index: /source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp =================================================================== --- /source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp (revision 10653) +++ /source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp (revision 10653) @@ -0,0 +1,54 @@ +#ifndef BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ +#define BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ + +#include "bluetooth_endpoint.hpp" + +#include +#include + +namespace boost { +namespace asio { +namespace bluetooth { + +/** + * The rfcomm class contains flags necessary for RFCOMM sockets. + * + * @author Martin Florian + */ +class rfcomm { +public: + /// The type of endpoint. + typedef bluetooth_endpoint endpoint; + + /// Get an Instance. + /// We need this to fulfill the asio Endpoint requirements, I think. + static rfcomm get() { + return rfcomm(); + } + + /// Obtain an identifier for the type of the protocol. + int type() const { + return SOCK_STREAM; + } + + /// Obtain an identifier for the protocol. + int protocol() const { + return BTPROTO_RFCOMM; + } + + /// Obtain an identifier for the protocol family. + int family() const { + return AF_BLUETOOTH; + } + + /// The RFCOMM socket type, lets pray that this will work. + typedef basic_stream_socket socket; + + /// The RFCOMM acceptor type. + typedef basic_socket_acceptor acceptor; + +}; + +}}} // namespace boost::asio::bluetooth + +#endif /* BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ */ Index: urce/ariba/utility/transport/rfcomm/rfcomm.cpp =================================================================== --- /source/ariba/utility/transport/rfcomm/rfcomm.cpp (revision 10647) +++ (revision ) @@ -1,439 +1,0 @@ -#include "ariba/config.h" - -#ifdef HAVE_LIBBLUETOOTH - -#include "rfcomm.hpp" - -#include "ariba/utility/transport/asio/asio_io_service.h" -#include "ariba/utility/transport/asio/rfcomm.hpp" -#include "ariba/utility/transport/asio/bluetooth_endpoint.hpp" - -#include -#include -#include -#include -#include -#include - -namespace ariba { -namespace transport { - -use_logging_cpp(rfcomm) - -using namespace boost::asio; -using namespace detail; -using namespace std; - -using boost::system::error_code; - -class link_data { -public: - uint8_t size_[4]; - size_t size; - uint8_t* buffer; -}; - -class link_info { -public: - link_info(io_service& io ) : - io(io), up(false), connecting(false), local(), remote(), socket(new bluetooth::rfcomm::socket(io)), connect_retries(0), - size(0), buffer(NULL), sending(false) { - } - - ~link_info() { - delete socket; - } - - void reinit() { - delete socket; - socket = new bluetooth::rfcomm::socket(io); - up = false; - } - - io_service& io; - - // state - bool up; - bool connecting; - rfcomm_endpoint local, remote; - bluetooth::rfcomm::socket* socket; - int connect_retries; - - // read buffer - uint8_t size_[4]; - size_t size; - uint8_t* buffer; - - // send buffer - bool sending; - boost::mutex mutex; - std::deque send_buffer; -}; - -void rfcomm::shutdown(link_info* info) { - if (info != NULL && info->up) { - info->up = false; - info->socket->shutdown( bluetooth::rfcomm::socket::shutdown_both ); - } -} - - -inline bluetooth::rfcomm::endpoint convert( const rfcomm_endpoint& endpoint ) { - return bluetooth::rfcomm::endpoint( - endpoint.mac().bluetooth(), endpoint.channel().value() - ); -} - -inline rfcomm_endpoint convert( const bluetooth::rfcomm::endpoint& endpoint ) { - mac_address mac; - mac.bluetooth( endpoint.address() ); - rfcomm_channel_address channel; - channel.value( endpoint.channel() ); - return rfcomm_endpoint( mac, channel ); -} - - -rfcomm::rfcomm(uint16_t channel) : - channel(channel), io(asio_io_service::alloc()) { - accept_retries = 0; -} - -rfcomm::~rfcomm() { - asio_io_service::free(); -} - -void rfcomm::start() { - - // start io service - asio_io_service::start(); - - // create acceptor - logging_info( "Binding to channel " << channel ); - acceptor = new bluetooth::rfcomm::acceptor(io, - bluetooth::rfcomm::endpoint(bluetooth::rfcomm::get(), channel ) - ); - - send_data = new link_data(); - - // start accepting - start_accept(); -} - -void rfcomm::stop() { - logging_info( "Stopping asio rfcomm" ); -} - -void rfcomm::send(const address_v* remote, const uint8_t* data, size_t size) { - - // get end-point - rfcomm_endpoint endpoint = *remote; - endpoint = convert(convert(endpoint)); - - // try to find established connector - logging_debug("Trying to find a already existing link."); - link_info* info = NULL; - for (size_t i = 0; i < links.size(); i++) - if (links[i]->remote.mac() == endpoint.mac()) { - logging_debug("Using already established link"); - info = links[i]; - break; - } - - // not found, or not up? ->try to (re-)connect - if (info==NULL || ((!info->up || !info->socket->is_open()) && !info->connecting) ) { - logging_debug( "Connecting to " << endpoint.to_string() ); - if (info != NULL && (!info->up || !info->socket->is_open())) { - logging_error("Old link is down. Trying to re-establish link."); - info->reinit(); - } else { - info = new link_info(io); - } - info->connect_retries = 0; - info->remote = endpoint; - info->connecting = true; - info->socket->async_connect( convert(endpoint), boost::bind( - &rfcomm::handle_connect, this, - boost::asio::placeholders::error, info - )); - links.push_back(info); - } - - // copy message - link_data ldata; - ldata.size = size; - ldata.size_[0] = (size >> 24) & 0xFF; - ldata.size_[1] = (size >> 16) & 0xFF; - ldata.size_[2] = (size >> 8) & 0xFF; - ldata.size_[3] = (size >> 0) & 0xFF; - ldata.buffer = new uint8_t[size]; - memcpy(ldata.buffer, data, size); - - // enqueue message - info->mutex.lock(); - info->send_buffer.push_back(ldata); - info->mutex.unlock(); - - // start writing - io.post( boost::bind( &rfcomm::start_write, this, info) ); -} - -void rfcomm::send(const endpoint_set& endpoints, const uint8_t* data, size_t size) { - // send a message to each combination of mac-address and channel - BOOST_FOREACH( const mac_address mac, endpoints.bluetooth ) { - BOOST_FOREACH( const rfcomm_channel_address channel, endpoints.rfcomm ) { - rfcomm_endpoint endpoint(mac, channel); - address_vf vf = endpoint; - send(vf,data,size); - } - } -} - -void rfcomm::terminate( const address_v* remote) { - // get end-point - rfcomm_endpoint endpoint = *remote; - - for (size_t i = 0; i < links.size(); i++) - if (links[i]->remote.mac() == endpoint.mac()) { - // close socket - shutdown(links[i]); - break; - } -} - -void rfcomm::register_listener(transport_listener* listener) { - this->listener = listener; -} - -void rfcomm::start_accept() { - - logging_info( "Waiting for connections ..." ); - - // start accepting a connection - link_info* info = new link_info(io); - acceptor->async_accept(*info->socket, boost::bind( - // bind parameters - &rfcomm::handle_accept, this, - - // handler parameters - boost::asio::placeholders::error, info - )); - asio_io_service::start(); -} - -void rfcomm::handle_accept(const error_code& error, link_info* info) { - if (error) { - logging_error( "Error waiting for new connections. Error code: "<< error.message() - << ", trying to recover (attempt " << accept_retries << ")"); - - // restart accepting - if (accept_retries<3) { - accept_retries++; - start_accept(); - } else - delete info; - - return; - } - - links_mutex.lock(); - - // convert endpoints - info->up = true; - info->local = convert( info->socket->local_endpoint() ); - info->remote = convert( info->socket->remote_endpoint() ); - - logging_debug("Accepted incoming connection from " - << info->remote.to_string() ); - - // add to list - links.push_back(info); - links_mutex.unlock(); - - // start i/o - start_read(info); - start_write(info); - - // restart accept - start_accept(); -} - -void rfcomm::handle_connect( const error_code& error, link_info* info ) { - if (error) { - logging_error( "Can not connect. Error code: " - << error.message() << " Retrying ... " - "(attempt " << info->connect_retries << ")" ); - - // do we retry this connection? yes-> - if (info->connect_retries<3) { - // increase counter - info->connecting = false; - info->connect_retries++; - info->reinit(); - - // retry connection attempt - info->socket->async_connect( convert(info->remote), boost::bind( - &rfcomm::handle_connect, this, - boost::asio::placeholders::error, info - )); - - } else { // no-> delete link and stop - return; - } - } - - // convert endpoints - info->up = true; - info->connecting = false; - info->local = convert( info->socket->local_endpoint() ); - info->remote = convert( info->socket->remote_endpoint() ); - - logging_debug( "Connected to " << info->remote.to_string() ); - - // add to list - links_mutex.lock(); - links.push_back(info); - links_mutex.unlock(); - - // start i/o - start_read(info); - start_write(info); -} - -void rfcomm::start_read(link_info* info) { - logging_debug("Start reading ..."); - - // start read - boost::asio::async_read(*info->socket, - - // read size of packet - boost::asio::buffer(info->size_, 4), - - // bind handler - boost::bind( - - // bind parameters - &rfcomm::handle_read_header, this, - - // handler parameters - placeholders::error, placeholders::bytes_transferred, info - ) - ); -} - -void rfcomm::handle_read_header(const error_code& error, size_t bytes, - link_info* info) { - - // handle error - if (error) { - logging_error("Failed to receive message payload. Error code: " - << error.message() ); - shutdown(info); - return; - } - - // ignore errors and wait for all data to be received - if (bytes != 4) return; - - // get size - info->size = (info->size_[0]<<24) + (info->size_[1] << 16) + - (info->size_[2]<< 8) + (info->size_[3] << 0); - - // allocate buffer - info->buffer = new uint8_t[info->size]; - - // start read - boost::asio::async_read(*info->socket, - // read size of packet - boost::asio::buffer(info->buffer, info->size), - // bind handler - boost::bind( - // bind parameters - &rfcomm::handle_read_data, this, - // handler parameters - placeholders::error, placeholders::bytes_transferred, info - ) - ); -} - -void rfcomm::handle_read_data(const error_code& error, size_t bytes, - link_info* info) { - - // check error - if (error) { - logging_error("Failed to receive message payload. Error: " << error.message() ); - shutdown(info); - return; - } - - // wait for all data to be received - if (bytes != info->size) - return; - - // deliver data - listener->receive_message(this, info->local, info->remote, info->buffer, info->size ); - - // free buffers and reset size buffer - delete [] info->buffer; - for (size_t i=0; i<4; i++) info->size_[i] = 0; - - start_read(info); -} - -void rfcomm::start_write( link_info* info ) { - // do not start writing if sending is in progress - if (info->sending || !info->up || info->send_buffer.size()==0) return; - - // set sending flag - info->sending = true; - - // safely remove data from deque - *send_data = info->send_buffer.front(); - info->send_buffer.pop_front(); - - boost::array buffer; - buffer[0] = boost::asio::buffer(send_data->size_,4); - buffer[1] = boost::asio::buffer(send_data->buffer,send_data->size); - - // start writing - boost::asio::async_write(*info->socket, - // read size of packet - buffer, - // bind handler - boost::bind( - // bind parameters - &rfcomm::handle_write_data, this, - // handler parameters - placeholders::error, placeholders::bytes_transferred, - info, send_data->size, send_data->buffer - ) - ); -} - -void rfcomm::handle_write_data(const error_code& error, size_t bytes, - link_info* info, size_t size, uint8_t* buffer ) { - - // handle error - if (error) { - logging_error( "Message sent error. Error: " << error.message() ); - info->sending = false; - shutdown(info); - return; - } - - // wait for all data to be sent - if (bytes != (size+4) ) - return; - - logging_debug( "Message sent" ); - - // free buffer - delete [] buffer; - info->sending = false; - - // restart-write - start_write(info); -} - -}} // namespace ariba::transport - -#endif Index: urce/ariba/utility/transport/rfcomm/rfcomm.hpp =================================================================== --- /source/ariba/utility/transport/rfcomm/rfcomm.hpp (revision 10647) +++ (revision ) @@ -1,78 +1,0 @@ -#include "ariba/config.h" - -#ifdef HAVE_LIBBLUETOOTH - -#ifndef RFCOMM_HPP_ -#define RFCOMM_HPP_ - -#include "ariba/utility/transport/transport.hpp" -#include "ariba/utility/transport/asio/asio_io_service.h" -#include "ariba/utility/transport/asio/bluetooth_endpoint.hpp" -#include "ariba/utility/transport/asio/rfcomm.hpp" - -#include -#include -#include - -#include "ariba/utility/logging/Logging.h" - -namespace ariba { -namespace transport { - -using boost::system::error_code; -using namespace boost::asio; - -class link_info; -class link_data; - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -class rfcomm : public transport_protocol { - use_logging_h(rfcomm) -public: - rfcomm( uint16_t channel ); - virtual ~rfcomm(); - virtual void start(); - virtual void stop(); - virtual void send( const address_v* remote, const uint8_t* data, size_t size ); - virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ); - virtual void terminate( const address_v* remote ); - virtual void register_listener( transport_listener* listener ); - -private: - uint16_t channel; - boost::mutex links_mutex; - vector links; - io_service& io; - transport_listener* listener; - bluetooth::rfcomm::acceptor* acceptor; - int accept_retries; - link_data* send_data; - - void start_accept(); - - void handle_accept( const error_code& error, link_info* info ); - - void start_read( link_info* info ); - - void handle_read_header( const error_code& error, size_t bytes, link_info* info ); - - void handle_read_data( const error_code& error, size_t bytes, link_info* info ); - - void handle_connect( const error_code& error, link_info* info ); - - void start_write( link_info* info ); - - void handle_write_data(const error_code& error, size_t bytes, - link_info* info, size_t size, uint8_t* buffer ); - - void shutdown(link_info* info); -}; - -}} // namespace ariba::transport - -#endif /* RFCOMM_HPP_ */ -#endif Index: /source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp =================================================================== --- /source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp (revision 10653) +++ /source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp (revision 10653) @@ -0,0 +1,464 @@ +#include "rfcomm_transport.hpp" + +#ifdef HAVE_LIBBLUETOOTH +#include + +namespace ariba { +namespace transport { + +use_logging_cpp(rfcomm_transport) + +using namespace ariba::addressing; + +typedef boost::mutex::scoped_lock unique_lock; + +/* constructor */ +rfcomm_transport::rfcomm_transport( const rfcomm_transport::rfcomm::endpoint& endp ) : + listener(NULL), + acceptor(u_io_service.get_asio_io_service(), endp) +{ +} + +rfcomm_transport::~rfcomm_transport(){} + +void rfcomm_transport::start() +{ + // open server socket + accept(); + + u_io_service.start(); +} + + +void rfcomm_transport::stop() +{ + acceptor.close(); + + u_io_service.stop(); +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const rfcomm::endpoint& dest_addr, + reboost::message_t message, + uint8_t priority) +{ + ConnPtr conn; + bool need_to_connect = false; + + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(dest_addr); + if (it == connections.end()) + { + ConnPtr tmp_ptr( + new rfcomm_connection( + u_io_service.get_asio_io_service(), + shared_from_this() ) + ); + conn = tmp_ptr; + + conn->partner = dest_addr; + conn->remote = convert_address(dest_addr); + + // Note: starting the send is the obligation of the connect_handler + // (avoids trying to send while not connected yet) + conn->sending = true; + need_to_connect = true; + + ConnectionMap::value_type item(dest_addr, conn); + connections.insert(item); + + } else { + conn = it->second; + } + } + + + // * the actual send * + conn->enqueue_for_sending(message, priority); + + // if new connection connect to the other party + if ( need_to_connect ) + { + conn->sock.async_connect( + dest_addr, + boost::bind( + &rfcomm_connection::async_connect_handler, + conn, + boost::asio::placeholders::error)); + } +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const address_v* remote, + reboost::message_t message, + uint8_t priority) +{ + send(convert_address(remote), message, priority); +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority ) +{ + // send a message to each combination of address-address and port + BOOST_FOREACH( const mac_address mac, endpoints.bluetooth ) { + BOOST_FOREACH( const rfcomm_channel_address channel, endpoints.rfcomm ) { + rfcomm::endpoint endp(mac.bluetooth(), channel.value()); + + // * send * + send(endp, message, priority); + } + } +} + + +void rfcomm_transport::register_listener( transport_listener* listener ) +{ + this->listener = listener; +} + + +void rfcomm_transport::terminate( const address_v* remote ) +{ + terminate(convert_address(remote)); +} + +void rfcomm_transport::terminate( const rfcomm::endpoint& remote ) +{ + ConnPtr conn; + + // find and forget connection + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(remote); + if (it == connections.end()) + { + return; + } + + conn = it->second; + + connections.erase(it); + } + + // close connection + boost::system::error_code ec; + conn->sock.shutdown(tcp::socket::shutdown_both, ec); + conn->sock.close(ec); +} + + +/* private */ +void rfcomm_transport::accept() +{ + // create new connection object + ConnPtr conn( + new rfcomm_connection( + u_io_service.get_asio_io_service(), + shared_from_this() + ) + ); + + // wait for incoming connection + acceptor.async_accept( + conn->sock, + boost::bind(&self::async_accept_handler, + this->shared_from_this(), + conn, + boost::asio::placeholders::error) + ); +} + +void rfcomm_transport::async_accept_handler(ConnPtr conn, const error_code& error) +{ + if ( ! error ) + { + conn->partner = conn->sock.remote_endpoint(); + conn->remote = convert_address(conn->partner); + conn->local = convert_address(conn->sock.local_endpoint()); + + { + unique_lock lock(connections_lock); + + ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn); + connections.insert(item); + } + + // read + conn->listen(); + } + + // accept further connections + accept(); +} + +inline rfcomm_transport::rfcomm::endpoint +rfcomm_transport::convert_address( const address_v* address ) +{ + rfcomm_endpoint endpoint = *address; + + return rfcomm::endpoint( + endpoint.mac().bluetooth(), endpoint.channel().value() + ); +} + + +inline rfcomm_endpoint rfcomm_transport::convert_address(const rfcomm::endpoint& endpoint) +{ + mac_address mac; + mac.bluetooth(endpoint.address()); + rfcomm_channel_address channel; + channel.value(endpoint.channel()); + return rfcomm_endpoint(mac, channel); +} + + +/***************** + ** inner class ** + *****************/ + +rfcomm_transport::rfcomm_connection::rfcomm_connection( + boost::asio::io_service & io_service, + rfcomm_transport::sptr parent) : + sock(io_service), + valid(true), + parent(parent), + out_queues(8), //TODO How much priorities shall we have? + sending(false) +{ + header.length = 0; + header.prot = 0; +} + +/*------------------------------------------- + | implement transport_connection interface | + -------------------------------------------*/ +void rfcomm_transport::rfcomm_connection::send( + reboost::message_t message, + uint8_t priority) +{ + enqueue_for_sending(message, priority); +} + + +address_vf rfcomm_transport::rfcomm_connection::getLocalEndpoint() +{ + return local; +} + + +address_vf rfcomm_transport::rfcomm_connection::getRemoteEndpoint() +{ + return remote; +} + + +void rfcomm_transport::rfcomm_connection::terminate() +{ + parent->terminate(partner); +} + + +/*------------------------------ + | things we defined ourselves | + ------------------------------*/ +void rfcomm_transport::rfcomm_connection::async_connect_handler(const error_code& error) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // save address in ariba format + local = parent->convert_address(sock.local_endpoint()); + + // Note: sending has to be true at this point + send_next_package(); + + listen(); +} + + +void rfcomm_transport::rfcomm_connection::listen() +{ + boost::asio::async_read( + this->sock, + boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)), + boost::bind( + &rfcomm_transport::rfcomm_connection::async_read_header_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + + +void rfcomm_transport::rfcomm_connection::async_read_header_handler(const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // convert byte order + header.length = ntohl(header.length); + header.length -= 2; // XXX protlib + + assert(header.length > 0); + + // new buffer for the new packet + buffy = shared_buffer_t(header.length); + + // * read data * + boost::asio::async_read( + this->sock, + boost::asio::buffer(buffy.mutable_data(), buffy.size()), + boost::bind( + &rfcomm_transport::rfcomm_connection::async_read_data_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + +void rfcomm_transport::rfcomm_connection::async_read_data_handler( + const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + message_t msg; + msg.push_back(buffy); + buffy = shared_buffer_t(); + + if ( parent->listener ) + parent->listener->receive_message(shared_from_this(), msg); + + listen(); +} + +/* see header file for comments */ +void rfcomm_transport::rfcomm_connection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred) +{ + if ( error ) + { + // remove this connection + parent->terminate(partner); + + return; + } + + send_next_package(); +} + + + +void rfcomm_transport::rfcomm_connection::enqueue_for_sending(Packet packet, uint8_t priority) +{ + bool restart_sending = false; + + // enqueue packet [locked] + { + unique_lock(out_queues_lock); + + assert( priority < out_queues.size() ); + out_queues[priority].push(packet); + + if ( ! sending ) + { + restart_sending = true; + sending = true; + } + } + + // if sending was stopped, we have to restart it here + if ( restart_sending ) + { + send_next_package(); + } +} + +/* see header file for comments */ +void rfcomm_transport::rfcomm_connection::send_next_package() +{ + Packet packet; + bool found = false; + + // find packet with highest priority [locked] + { + unique_lock(out_queues_lock); + + for ( vector::iterator it = out_queues.begin(); + it != out_queues.end(); it++ ) + { + if ( !it->empty() ) + { + packet = it->front(); + it->pop(); + found = true; + + break; + } + } + + // no packets waiting --> stop sending + if ( ! found ) + { + sending = false; + } + } + + // * send * + if ( found ) + { + reboost::shared_buffer_t header_buf(sizeof(header_t)); + header_t* header = (header_t*)(header_buf.mutable_data()); + header->length = htonl(packet.size()+2); // XXX protlib + + packet.push_front(header_buf); + + // "convert" message to asio buffer sequence + vector send_sequence(packet.length()); + for ( int i=0; i < packet.length(); i++ ) + { + shared_buffer_t b = packet.at(i); + send_sequence.push_back(boost::asio::buffer(b.data(), b.size())); + } + + // * async write * + boost::asio::async_write( + this->sock, + send_sequence, + boost::bind( + &rfcomm_transport::rfcomm_connection::async_write_handler, + this->shared_from_this(), + packet, // makes sure our shared pointer lives long enough ;-) + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred) + ); + } +} + +}} // namespace ariba::transport + +#endif /* HAVE_LIBBLUETOOTH */ Index: /source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp =================================================================== --- /source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp (revision 10653) +++ /source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp (revision 10653) @@ -0,0 +1,170 @@ +#include "ariba/config.h" + +#ifdef HAVE_LIBBLUETOOTH + +#ifndef RFCOMM_TRANSPORT_HPP_ +#define RFCOMM_TRANSPORT_HPP_ + +#include "ariba/utility/transport/transport.hpp" +#include "ariba/utility/transport/asio/unique_io_service.h" +#include "ariba/utility/transport/transport_connection.hpp" +#include "ariba/utility/addressing/rfcomm_endpoint.hpp" +#include +#include +#include +#include +#include "ariba/utility/transport/messages/buffers.hpp" +#include "ariba/utility/logging/Logging.h" +#include "bluetooth_rfcomm.hpp" + +namespace ariba { +namespace transport { + +using namespace std; +using ariba::transport::detail::unique_io_service; +using ariba::addressing::rfcomm_endpoint; +using boost::system::error_code; +using reboost::shared_buffer_t; +using reboost::message_t; + + +class rfcomm_transport : + public transport_protocol, + public boost::enable_shared_from_this +{ +public: + typedef boost::shared_ptr sptr; + +private: + typedef rfcomm_transport self; + typedef boost::asio::bluetooth::rfcomm rfcomm; + use_logging_h(rfcomm_transport) + + class rfcomm_connection : + public transport_connection, + public boost::enable_shared_from_this + { + public: + typedef reboost::message_t Packet; + typedef std::queue OutQueue; + + struct header_t + { + uint32_t length; + uint16_t prot; // XXX protlib + } __attribute__((packed)); + + rfcomm_connection(boost::asio::io_service& io_service, + rfcomm_transport::sptr parent); + + /// Inherited from transport_connection + virtual void send(reboost::message_t message, uint8_t priority = 0); + virtual address_vf getLocalEndpoint(); + virtual address_vf getRemoteEndpoint(); + virtual void terminate(); + + void listen(); + + void async_connect_handler(const error_code& error); + + void async_read_header_handler(const error_code& error, size_t bytes_transferred); + void async_read_data_handler(const error_code& error, size_t bytes_transferred); + + /* + * is called from asio when write operation "returns", + * calls private function `send_next_package()` + */ + void async_write_handler( + reboost::shared_buffer_t packet, + const error_code& error, + size_t bytes_transferred); + + + void enqueue_for_sending(Packet packet, uint8_t priority); + + private: + /* + * is called from `send` or `async_write_handler` to begin/keep sending + * sends the next message with the highest priority in this connection + */ + void send_next_package(); + + + public: + rfcomm::socket sock; + bool valid; + rfcomm_transport::sptr parent; + + rfcomm::endpoint partner; + rfcomm_endpoint remote; + rfcomm_endpoint local; + + vector out_queues; // to be locked with out_queues_lock + boost::mutex out_queues_lock; + + bool sending; // to be locked with out_queues_lock + + header_t header; + shared_buffer_t buffy; + }; + typedef boost::shared_ptr ConnPtr; + typedef std::map ConnectionMap; + +public: + /* constructor */ + rfcomm_transport( const rfcomm::endpoint& endp ); + virtual ~rfcomm_transport(); + + virtual void start(); + virtual void stop(); + + /** + * enqueues message for sending + * create new connection if necessary + * starts sending mechanism (if not already running) + */ + void send( + const rfcomm::endpoint&, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * Converts address_v to rfcomm::endpoint and calls the real send() function + */ + virtual void send( + const address_v* remote, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * calls send for each destination endpoint in `endpoint_set& endpoints` + */ + virtual void send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority = 0 ); + + virtual void terminate( const address_v* remote ); + virtual void terminate( const rfcomm::endpoint& remote ); + virtual void register_listener( transport_listener* listener ); + + +private: + void accept(); + void async_accept_handler(ConnPtr conn, const error_code& error); + rfcomm::endpoint convert_address(const address_v* endpoint); + rfcomm_endpoint convert_address(const rfcomm::endpoint& endpoint); + +private: + transport_listener* listener; + unique_io_service u_io_service; + rfcomm::acceptor acceptor; + + ConnectionMap connections; + boost::mutex connections_lock; +}; + +}} // namespace ariba::transport + +#endif /* RFCOMM_TRANSPORT_HPP_ */ +#endif /* HAVE_LIBBLUETOOTH */ Index: /source/ariba/utility/transport/tcpip/tcpip.cpp =================================================================== --- /source/ariba/utility/transport/tcpip/tcpip.cpp (revision 10647) +++ /source/ariba/utility/transport/tcpip/tcpip.cpp (revision 10653) @@ -1,212 +1,539 @@ #include "tcpip.hpp" -#define _NO_LOGGING - -// std includes -#include -#include -#include -#include -#include - -// protlib includes -#include "protlib/network_message.h" -#include "protlib/tp_over_tcp.h" -#include "protlib/tperror.h" -#include "protlib/logfile.h" -#include "protlib/queuemanager.h" -#include "protlib/threadsafe_db.h" -#include "protlib/setuid.h" - -// protlib namespaces -using namespace protlib; -using namespace protlib::log; - -logfile commonlog; -protlib::log::logfile& protlib::log::DefaultLog(commonlog); +#include + +// interface discovery for link-local destinations +#include namespace ariba { namespace transport { +use_logging_cpp(tcpip) + using namespace ariba::addressing; - -tcpip_endpoint convert( const appladdress* addr ) { - const char* ip_str = addr->get_ip_str(); - tcpip_endpoint endpoint( std::string(ip_str), addr->get_port() ); - return endpoint; -} - -appladdress convert( const tcpip_endpoint& endpoint ) { - tcpip_endpoint* e = const_cast(&endpoint); - appladdress - peer(e->address().to_string().c_str(), "tcp", e->port().asio() ); -// cout << endpoint.to_string() << " to " << peer.get_ip_str() << ":" << peer.get_port() << endl; - return peer; -} - -tcpip::tcpip( uint16_t port ) : - done ( false ), - running ( false ), - port( port ), - tpreceivethread ( NULL ), - tpthread ( NULL ), - listener ( NULL ) { -} - -tcpip::~tcpip() { - if (running) stop(); -} - -bool get_message_length( NetMsg& m, uint32& clen_bytes ) { - clen_bytes = m.decode32(); - m.set_pos_r(-4); - return true; -} - -void tcpip::start() { - done = false; - running = false; - - // initalize netdb and setuid - protlib::tsdb::init(); - protlib::setuid::init(); - - // set tcp parameters - port_t port = this->port; // port - TPoverTCPParam tppar(4, get_message_length, port); - - // create receiver thread - FastQueue* tpchecker_fq = new FastQueue("TCPTransport", true); - QueueManager::instance()->register_queue(tpchecker_fq, - message::qaddr_signaling); - - // start thread - pthread_create( &tpreceivethread, NULL, tcpip::receiverThread, this ); - tpthread = new ThreadStarter ( 1, tppar ); - tpthread->start_processing(); -} - -void tcpip::stop() { - // stop receiver thread - done = true; - - // stop TPoverTCP - tpthread->stop_processing(); - tpthread->abort_processing(true); - tpthread->wait_until_stopped(); - - // unregister TPoverTCP - delete QueueManager::instance()->get_queue( message::qaddr_signaling ); - QueueManager::instance()->unregister_queue( message::qaddr_signaling ); - - // destroy QueueManager - QueueManager::clear(); - - // de-initalize netdb and setuid - protlib::setuid::end(); - protlib::tsdb::end(); - - // wait for thread to finish and delete - pthread_join(tpreceivethread, NULL); -} - -void tcpip::send( const address_v* remote, const uint8_t* data, size_t size ) { - - // prepare netmsg with length and and port - NetMsg* datamsg = new NetMsg(size + 6); - datamsg->encode32( size + 2, true ); - datamsg->encode16( this->port,true ); - - for (size_t i=0; iencode8( data[i],true ); - - // send message - tcpip_endpoint endpoint = *remote; - appladdress peer = convert(endpoint); - - // add to output queue - tpthread->get_thread_object()->send( datamsg, peer, false ); -} - -void tcpip::send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) { - // send a message to each combination of ip-address and port - BOOST_FOREACH( const ip_address ip, endpoints.ip ) { - BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) { - tcpip_endpoint endpoint(ip,port); - address_vf vf = endpoint; - send(vf,data,size); - } - } -} - -void tcpip::terminate( const address_v* remote) { - tcpip_endpoint endpoint = *remote; - appladdress peer = convert(endpoint); - peer.convert_to_ipv6(); - tpthread->get_thread_object()->terminate( peer ); -} - -void tcpip::register_listener( transport_listener* listener ) { - this->listener = listener; -} - -void* tcpip::receiverThread( void* ptp ) { - // get reference to transport object - tcpip& tp = *((tcpip*)ptp); - - // get queue - FastQueue* fq = - QueueManager::instance()->get_queue(message::qaddr_signaling); - - // main processing loop - tp.running = true; - while (!tp.done) { - - // wait for new message to approach - message* msg = fq->dequeue_timedwait(300); - - // message has arrived? no-> continue - if (!msg) continue; - - // handle transport message - TPMsg* tpmsg = dynamic_cast (msg); - if (!tpmsg) { - delete msg; - continue; - } - - // get address & message - const appladdress* remote_peer = static_cast( tpmsg->get_peeraddress() ); - const appladdress* local_peer = static_cast( tpmsg->get_ownaddress() ); - NetMsg* datamsg = tpmsg->get_message(); - - // not a data message? -> continue! - if (!datamsg) { - delete tpmsg; - continue; - } - - // get length and remote endpoint port - datamsg->set_pos(0); - uint32_t message_size = datamsg->decode32(true)-2; - //uint16_t remote_port = datamsg->decode16(true); - - // inform listener - if (tp.listener != NULL) { - tcpip_endpoint remote = convert(remote_peer); - tcpip_endpoint local = convert(local_peer); - tp.listener->receive_message( - &tp, local, remote, datamsg->get_buffer()+6, message_size ); - } - - tpmsg->set_message(NULL); - delete datamsg; - delete tpmsg; - } - // clean queue & stop - fq->cleanup(); - tp.running = false; - return NULL; +typedef boost::mutex::scoped_lock unique_lock; + +tcpip::tcpip( const tcp::endpoint& endp ) : + listener(NULL), + acceptor(u_io_service.get_asio_io_service(), endp) +{ +} + +tcpip::~tcpip(){} + +void tcpip::start() +{ + // open server socket + accept(); + + u_io_service.start(); +} + + +void tcpip::stop() +{ + acceptor.close(); + + u_io_service.stop(); +} + + +/* see header file for comments */ +void tcpip::send( + const tcp::endpoint& dest_addr, + reboost::message_t message, + uint8_t priority) +{ + ConnPtr conn; + bool need_to_connect = false; + + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(dest_addr); + if (it == connections.end()) + { + ConnPtr tmp_ptr( + new tcpip_connection( + u_io_service.get_asio_io_service(), + shared_from_this() ) + ); + conn = tmp_ptr; + + conn->partner = dest_addr; + conn->remote = convert_address(dest_addr); + + // Note: starting the send is the obligation of the connect_handler + // (avoids trying to send while not connected yet) + conn->sending = true; + need_to_connect = true; + + ConnectionMap::value_type item(dest_addr, conn); + connections.insert(item); + + } else { + conn = it->second; + } + } + + + // * the actual send * + conn->enqueue_for_sending(message, priority); + + // if new connection connect to the other party + if ( need_to_connect ) + { + conn->sock.async_connect( + dest_addr, + boost::bind( + &tcpip_connection::async_connect_handler, + conn, + boost::asio::placeholders::error)); + } +} + + +/* see header file for comments */ +void tcpip::send( + const address_v* remote, + reboost::message_t message, + uint8_t priority) +{ + send(convert_address(remote), message, priority); +} + + +/* see header file for comments */ +void tcpip::send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority ) +{ + // network interfaces scope_ids, for link-local connections (lazy initialization) + vector scope_ids; + + // send a message to each combination of address-address and port + BOOST_FOREACH( const ip_address address, endpoints.ip ) { + BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) { + tcp::endpoint endp(address.asio(), port.asio()); + + // special treatment for link local addresses + // ---> send over all (suitable) interfaces + if ( endp.address().is_v6() ) + { + boost::asio::ip::address_v6 v6_addr = endp.address().to_v6(); + + if ( v6_addr.is_link_local() ) + { + // initialize scope_ids + if ( scope_ids.size() == 0 ) + scope_ids = get_interface_scope_ids(); + + BOOST_FOREACH ( uint64_t id, scope_ids ) + { + v6_addr.scope_id(id); + endp.address(v6_addr); + + logging_debug("------> SEND TO (link-local): " << endp); + // * send * + send(endp, message, priority); + } + } + + continue; + } + + // * send * + send(endp, message, priority); + } + } +} + + +void tcpip::register_listener( transport_listener* listener ) +{ + this->listener = listener; +} + + +void tcpip::terminate( const address_v* remote ) +{ + terminate(convert_address(remote)); +} + +void tcpip::terminate( const tcp::endpoint& remote ) +{ + ConnPtr conn; + + // find and forget connection + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(remote); + if (it == connections.end()) + { + return; + } + + conn = it->second; + + connections.erase(it); + } + + // close connection + boost::system::error_code ec; + conn->sock.shutdown(tcp::socket::shutdown_both, ec); + conn->sock.close(ec); +} + + +/* private */ +void tcpip::accept() +{ + // create new connection object + ConnPtr conn( + new tcpip_connection( + u_io_service.get_asio_io_service(), + shared_from_this() + ) + ); + + // wait for incoming connection + acceptor.async_accept( + conn->sock, + boost::bind(&self::async_accept_handler, + this->shared_from_this(), + conn, + boost::asio::placeholders::error) + ); +} + +void tcpip::async_accept_handler(ConnPtr conn, const error_code& error) +{ + if ( ! error ) + { + conn->partner = conn->sock.remote_endpoint(); + conn->remote = convert_address(conn->partner); + conn->local = convert_address(conn->sock.local_endpoint()); + + { + unique_lock lock(connections_lock); + + ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn); + connections.insert(item); + } + + // read + conn->listen(); + } + + // accept further connections + accept(); +} + +inline tcp::endpoint tcpip::convert_address( const address_v* address ) +{ + tcpip_endpoint endpoint = *address; + + return tcp::endpoint( + endpoint.address().asio(), endpoint.port().value() + ); +} + + +inline tcpip_endpoint tcpip::convert_address(const tcp::endpoint& endpoint) +{ + ip_address address; + address.asio(endpoint.address()); + tcp_port_address port; + port.value(endpoint.port()); + return tcpip_endpoint(address, port); +} + + +vector tcpip::get_interface_scope_ids() +{ + vector ret; + + struct ifaddrs* ifaceBuffer = NULL; + void* tmpAddrPtr = NULL; + + int ok = getifaddrs( &ifaceBuffer ); + if( ok != 0 ) return ret; + + 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; + + // only use ethX and wlanX devices + string device = string(i->ifa_name); + if ( (device.find("eth") == string::npos) && + (device.find("wlan") == string::npos) /* && + (device.find("lo") == string::npos) XXX */ ) + { + continue; + } + + // only use interfaces with ipv6 link-local addresses + if (addr->sa_family == AF_INET6) + { + // convert address + // TODO should be possible without detour over strings + char straddr[INET6_ADDRSTRLEN]; + tmpAddrPtr= &((struct sockaddr_in6*)addr)->sin6_addr; + inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) ); + + address_v6 v6addr = address_v6::from_string(straddr); + if ( v6addr.is_link_local() ) + { + // * append the scope_id to the return vector * + ret.push_back(if_nametoindex(i->ifa_name)); + } + + } + } + + freeifaddrs(ifaceBuffer); + + return ret; +} + + +/***************** + ** inner class ** + *****************/ + +tcpip::tcpip_connection::tcpip_connection(boost::asio::io_service & io_service, TcpIpPtr parent) : + sock(io_service), + valid(true), + parent(parent), + out_queues(8), //TODO How much priorities shall we have? + sending(false) +{ + header.length = 0; + header.prot = 0; +} + +/*------------------------------------------- + | implement transport_connection interface | + -------------------------------------------*/ +void tcpip::tcpip_connection::send( + reboost::message_t message, + uint8_t priority) +{ + enqueue_for_sending(message, priority); +} + + +address_vf tcpip::tcpip_connection::getLocalEndpoint() +{ + return local; +} + + +address_vf tcpip::tcpip_connection::getRemoteEndpoint() +{ + return remote; +} + + +void tcpip::tcpip_connection::terminate() +{ + parent->terminate(partner); +} + + +/*------------------------------ + | things we defined ourselves | + ------------------------------*/ +void tcpip::tcpip_connection::async_connect_handler(const error_code& error) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // save address in ariba format + local = parent->convert_address(sock.local_endpoint()); + + // Note: sending has to be true at this point + send_next_package(); + + listen(); +} + + +void tcpip::tcpip_connection::listen() +{ + boost::asio::async_read( + this->sock, + boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)), + boost::bind( + &tcpip::tcpip_connection::async_read_header_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + + +void tcpip::tcpip_connection::async_read_header_handler(const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // convert byte order + header.length = ntohl(header.length); + header.length -= 2; // XXX protlib + + assert(header.length > 0); + + // new buffer for the new packet + buffy = shared_buffer_t(header.length); + + // * read data * + boost::asio::async_read( + this->sock, + boost::asio::buffer(buffy.mutable_data(), buffy.size()), + boost::bind( + &tcpip::tcpip_connection::async_read_data_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + +void tcpip::tcpip_connection::async_read_data_handler( + const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + message_t msg; + msg.push_back(buffy); + buffy = shared_buffer_t(); + + if ( parent->listener ) + parent->listener->receive_message(shared_from_this(), msg); + + listen(); +} + +/* see header file for comments */ +void tcpip::tcpip_connection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred) +{ + if ( error ) + { + // remove this connection + parent->terminate(partner); + + return; + } + + send_next_package(); +} + + + +void tcpip::tcpip_connection::enqueue_for_sending(Packet packet, uint8_t priority) +{ + bool restart_sending = false; + + // enqueue packet [locked] + { + unique_lock(out_queues_lock); + + assert( priority < out_queues.size() ); + out_queues[priority].push(packet); + + if ( ! sending ) + { + restart_sending = true; + sending = true; + } + } + + // if sending was stopped, we have to restart it here + if ( restart_sending ) + { + send_next_package(); + } +} + +/* see header file for comments */ +void tcpip::tcpip_connection::send_next_package() +{ + Packet packet; + bool found = false; + + // find packet with highest priority [locked] + { + unique_lock(out_queues_lock); + + for ( vector::iterator it = out_queues.begin(); + it != out_queues.end(); it++ ) + { + if ( !it->empty() ) + { + packet = it->front(); + it->pop(); + found = true; + + break; + } + } + + // no packets waiting --> stop sending + if ( ! found ) + { + sending = false; + } + } + + // * send * + if ( found ) + { + reboost::shared_buffer_t header_buf(sizeof(header_t)); + header_t* header = (header_t*)(header_buf.mutable_data()); + header->length = htonl(packet.size()+2); // XXX protlib + + packet.push_front(header_buf); + + // "convert" message to asio buffer sequence + vector send_sequence(packet.length()); + for ( int i=0; i < packet.length(); i++ ) + { + shared_buffer_t b = packet.at(i); + send_sequence.push_back(boost::asio::buffer(b.data(), b.size())); + } + + // * async write * + boost::asio::async_write( + this->sock, + send_sequence, + boost::bind( + &tcpip::tcpip_connection::async_write_handler, + this->shared_from_this(), + packet, // makes sure our shared pointer lives long enough ;-) + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred) + ); + } } Index: /source/ariba/utility/transport/tcpip/tcpip.hpp =================================================================== --- /source/ariba/utility/transport/tcpip/tcpip.hpp (revision 10647) +++ /source/ariba/utility/transport/tcpip/tcpip.hpp (revision 10653) @@ -3,42 +3,168 @@ #include "ariba/utility/transport/transport.hpp" -#include - -// forward declaration -namespace protlib { -template -class ThreadStarter; -class TPoverTCP; -class TPoverTCPParam; -} +#include "ariba/utility/transport/asio/unique_io_service.h" +#include "ariba/utility/transport/transport_connection.hpp" +#include "ariba/utility/addressing/tcpip_endpoint.hpp" +#include +#include +#include +#include +#include "ariba/utility/transport/messages/buffers.hpp" +#include "ariba/utility/logging/Logging.h" namespace ariba { namespace transport { -using namespace protlib; +using namespace std; +using ariba::transport::detail::unique_io_service; +using ariba::addressing::tcpip_endpoint; +using boost::asio::ip::tcp; +using boost::asio::ip::address_v6; +using boost::system::error_code; +using reboost::shared_buffer_t; +using reboost::message_t; -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -class tcpip : public transport_protocol { +class tcpip; +typedef boost::shared_ptr TcpIpPtr; + +class tcpip : + public transport_protocol, + public boost::enable_shared_from_this +{ + typedef tcpip self; +use_logging_h(tcpip) + +private: + class tcpip_connection : + public transport_connection, + public boost::enable_shared_from_this + { + public: + typedef reboost::message_t Packet; + typedef std::queue OutQueue; + + struct header_t + { + uint32_t length; + uint16_t prot; // XXX protlib + } __attribute__((packed)); + + tcpip_connection(boost::asio::io_service& io_service, TcpIpPtr parent); + + /// Inherited from transport_connection + virtual void send(reboost::message_t message, uint8_t priority = 0); + virtual address_vf getLocalEndpoint(); + virtual address_vf getRemoteEndpoint(); + virtual void terminate(); + + void listen(); + + void async_connect_handler(const error_code& error); + + void async_read_header_handler(const error_code& error, size_t bytes_transferred); + void async_read_data_handler(const error_code& error, size_t bytes_transferred); + + /* + * is called from asio when write operation "returns", + * calls private function `send_next_package()` + */ + void async_write_handler( + reboost::shared_buffer_t packet, + const error_code& error, + size_t bytes_transferred); + + + void enqueue_for_sending(Packet packet, uint8_t priority); + + private: + /* + * is called from `send` or `async_write_handler` to begin/keep sending + * sends the next message with the highest priority in this connection + */ + void send_next_package(); + + + public: + tcp::socket sock; + bool valid; + TcpIpPtr parent; + + tcp::endpoint partner; + tcpip_endpoint remote; + tcpip_endpoint local; + + vector out_queues; // to be locked with out_queues_lock + boost::mutex out_queues_lock; + + bool sending; // to be locked with out_queues_lock + + header_t header; + shared_buffer_t buffy; + }; + typedef boost::shared_ptr ConnPtr; + typedef std::map ConnectionMap; + public: - tcpip( uint16_t port ); + tcpip( const tcp::endpoint& endp ); virtual ~tcpip(); virtual void start(); virtual void stop(); - virtual void send( const address_v* remote, const uint8_t* data, size_t size ); - virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ); + + /** + * enqueues message for sending + * create new connection if necessary + * starts sending mechanism (if not already running) + */ + void send( + const tcp::endpoint&, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * Converts address_v to tcp::endpoint and calls the real send() function + */ + virtual void send( + const address_v* remote, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * calls send for each destination endpoint in `endpoint_set& endpoints` + */ + virtual void send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority = 0 ); + virtual void terminate( const address_v* remote ); + virtual void terminate( const tcp::endpoint& remote ); virtual void register_listener( transport_listener* listener ); + + /** + * returns a vector of (interesting) network interfaces + * + * [NOTE: The current implementation returns the scope_ids of + * all ethX and wlanX network interfaces, to be used for + * connections to link-local ipv6 addresses.] + * + * TODO move to ariba/communication/networkinfo/AddressDiscovery ?? + * + */ + static vector get_interface_scope_ids(); + private: - volatile bool done, running; - uint16_t port; - pthread_t tpreceivethread; - ThreadStarter* tpthread; - static void* receiverThread( void* ptp ); + void accept(); + void async_accept_handler(ConnPtr conn, const error_code& error); + tcp::endpoint convert_address(const address_v* endpoint); + tcpip_endpoint convert_address(const tcp::endpoint& endpoint); + +private: transport_listener* listener; + unique_io_service u_io_service; + tcp::acceptor acceptor; + + ConnectionMap connections; + boost::mutex connections_lock; }; Index: /source/ariba/utility/transport/transport.hpp =================================================================== --- /source/ariba/utility/transport/transport.hpp (revision 10647) +++ /source/ariba/utility/transport/transport.hpp (revision 10653) @@ -8,5 +8,5 @@ // transport protocol implementations #include "tcpip/tcpip.hpp" -#include "rfcomm/rfcomm.hpp" +#include "rfcomm/rfcomm_transport.hpp" // common transport peer using all known protocols Index: /source/ariba/utility/transport/transport_connection.hpp =================================================================== --- /source/ariba/utility/transport/transport_connection.hpp (revision 10653) +++ /source/ariba/utility/transport/transport_connection.hpp (revision 10653) @@ -0,0 +1,32 @@ + +#ifndef TRANSPORT_CONNECTION_HPP_ +#define TRANSPORT_CONNECTION_HPP_ + +#include "ariba/utility/addressing/addressing.hpp" +#include "ariba/utility/transport/messages/message.hpp" +#include + +using ariba::addressing::address_vf; + +namespace ariba { +namespace transport { + +class transport_connection +{ +public: + typedef boost::shared_ptr sptr; + + /// Allow deleting implementing classes by pointer + virtual ~transport_connection() {} + + virtual void send(reboost::message_t message, uint8_t priority = 0) = 0; + + virtual address_vf getLocalEndpoint() = 0; + virtual address_vf getRemoteEndpoint() = 0; + + virtual void terminate() = 0; +}; + +} /* namespace transport */ +} /* namespace ariba */ +#endif /* TRANSPORT_CONNECTION_HPP_ */ Index: /source/ariba/utility/transport/transport_listener.hpp =================================================================== --- /source/ariba/utility/transport/transport_listener.hpp (revision 10647) +++ /source/ariba/utility/transport/transport_listener.hpp (revision 10653) @@ -5,4 +5,6 @@ #include "ariba/utility/addressing/addressing.hpp" +#include "ariba/utility/transport/messages/buffers.hpp" +#include "ariba/utility/transport/transport_connection.hpp" // namespace ariba::transport @@ -11,6 +13,4 @@ using namespace ariba::addressing; - -class transport_protocol; /** @@ -21,9 +21,11 @@ class transport_listener { public: + /// Allow deleting implementing classes by pointer + virtual ~transport_listener() {} + /// called when a message is received virtual void receive_message( - transport_protocol* transport, - const address_vf local, const address_vf remote, - const uint8_t* data, size_t size + transport_connection::sptr connection, + reboost::message_t msg ) { std::cout << "transport_listener: not implemented" << std::endl; Index: /source/ariba/utility/transport/transport_peer.cpp =================================================================== --- /source/ariba/utility/transport/transport_peer.cpp (revision 10647) +++ /source/ariba/utility/transport/transport_peer.cpp (revision 10653) @@ -3,4 +3,14 @@ #include "transport_peer.hpp" #include "transport.hpp" +#include "ariba/utility/logging/Logging.h" +#include +#include +#include + +#ifdef ECLIPSE_PARSER + #define foreach(a, b) for(a : b) +#else + #define foreach(a, b) BOOST_FOREACH(a, b) +#endif // namespace ariba::transport @@ -9,78 +19,153 @@ using namespace ariba::addressing; +using boost::asio::ip::tcp; + +#ifdef HAVE_LIBBLUETOOTH +using boost::asio::bluetooth::rfcomm; +#endif + +use_logging_cpp(transport_peer); transport_peer::transport_peer( endpoint_set& local_set ) : local(local_set) { - // setup tcp transports - tcp = NULL; - //cout << "#tcpip_transports = " << local.tcp.size() << endl; - if (local.tcp.size()==1) { - tcp = new tcpip(local.tcp.begin()->value()); - //cout << "Started tcpip_transport on port " << local.tcp.begin()->value() << endl; - } - + + // setup tcp transports + foreach(tcp_port_address port, local.tcp) { + + 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); + } + + } + #ifdef HAVE_LIBBLUETOOTH - // setup rfcomm transports - rfc = NULL; - //cout << "#rfcomm_transports = " << local.rfcomm.size() << endl; - if ( local.rfcomm.size() == 1 ) { - rfc = new rfcomm( local.rfcomm.begin()->value() ); - //cout << "Started rfcomm_transport on port " << local.rfcomm.begin()->value() << endl; - } + 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; + } + } +} + +#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; + } + } +} +#endif + transport_peer::~transport_peer() { - if (tcp !=NULL ) delete tcp; -#ifdef HAVE_LIBBLUETOOTH - if (rfc !=NULL ) delete rfc; -#endif } void transport_peer::start() { - if (tcp!=NULL) tcp->start(); + foreach(TcpIpPtr tcp, tcps) { + tcp->start(); + } + #ifdef HAVE_LIBBLUETOOTH - if (rfc!=NULL) rfc->start(); + foreach(rfcomm_transport::sptr x, rfcomms) { + x->start(); + } #endif } void transport_peer::stop() { - if (tcp!=NULL) tcp->stop(); + foreach(TcpIpPtr tcp, tcps) { + tcp->stop(); + } + #ifdef HAVE_LIBBLUETOOTH - if (rfc!=NULL) rfc->stop(); + foreach(rfcomm_transport::sptr x, rfcomms) { + x->stop(); + } #endif } -void transport_peer::send( const address_v* remote, const uint8_t* data, size_t size ) { - if (remote->instanceof() && tcp!=NULL) { - tcp->send(remote,data,size); - } else + +void transport_peer::send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority) +{ + foreach(TcpIpPtr tcp, tcps) { + tcp->send(endpoints, message, priority); + } + #ifdef HAVE_LIBBLUETOOTH - if (remote->instanceof() && rfc!=NULL) { - rfc->send(remote,data,size); - } else -#endif - cerr << "Could not send message to " << remote->to_string() << endl; -} - -void transport_peer::send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) { - if (tcp!=NULL) tcp->send(endpoints,data,size); -#ifdef HAVE_LIBBLUETOOTH - if (rfc!=NULL) rfc->send(endpoints,data,size); + foreach(rfcomm_transport::sptr x, rfcomms) { + x->send(endpoints, message, priority); + } #endif } void transport_peer::terminate( const address_v* remote ) { - if (remote->instanceof() && tcp!=NULL) - tcp->terminate(remote); + if (remote->instanceof())// TODO direkt auf der richtigen verbindung + { + foreach(TcpIpPtr tcp, tcps) { + tcp->terminate(remote); + } + } #ifdef HAVE_LIBBLUETOOTH - if (remote->instanceof() && rfc!=NULL) - rfc->terminate(remote); + if (remote->instanceof()) { + foreach(rfcomm_transport::sptr x, rfcomms) { + x->terminate(remote); + } + } #endif } void transport_peer::register_listener( transport_listener* listener ) { - if (tcp!=NULL) tcp->register_listener(listener); + foreach(TcpIpPtr tcp, tcps) { + tcp->register_listener(listener); + } + #ifdef HAVE_LIBBLUETOOTH - if (rfc!=NULL) rfc->register_listener(listener); + foreach(rfcomm_transport::sptr x, rfcomms) { + x->register_listener(listener); + } #endif } Index: /source/ariba/utility/transport/transport_peer.hpp =================================================================== --- /source/ariba/utility/transport/transport_peer.hpp (revision 10647) +++ /source/ariba/utility/transport/transport_peer.hpp (revision 10653) @@ -5,4 +5,7 @@ #include "transport_protocol.hpp" #include "ariba/utility/addressing/endpoint_set.hpp" +#include +#include "rfcomm/bluetooth_rfcomm.hpp" + // namespace ariba::transport @@ -13,6 +16,7 @@ class tcpip; + #ifdef HAVE_LIBBLUETOOTH -class rfcomm; +class rfcomm_transport; #endif @@ -30,14 +34,25 @@ virtual void start(); virtual void stop(); - virtual void send( const address_v* remote, const uint8_t* data, size_t size ); - virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ); + + virtual void send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority = 0); + + /// @deprecated: Use terminate() from transport_connection instead virtual void terminate( const address_v* remote ); + 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; - tcpip* tcp; + std::vector< boost::shared_ptr > tcps; #ifdef HAVE_LIBBLUETOOTH - rfcomm* rfc; + std::vector< boost::shared_ptr > rfcomms; #endif }; Index: /source/ariba/utility/transport/transport_protocol.hpp =================================================================== --- /source/ariba/utility/transport/transport_protocol.hpp (revision 10647) +++ /source/ariba/utility/transport/transport_protocol.hpp (revision 10653) @@ -3,5 +3,6 @@ #include "ariba/utility/addressing/addressing.hpp" -#include "transport_listener.hpp" +#include "ariba/utility/transport/transport_listener.hpp" +#include "ariba/utility/transport/messages/message.hpp" // namespace ariba::transport @@ -18,9 +19,18 @@ class transport_protocol { public: + /// Allow deleting implementing classes by pointer + virtual ~transport_protocol() {} + virtual void start() = 0; virtual void stop() = 0; - virtual void send( const address_v* remote, const uint8_t* data, size_t size ) = 0; - virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) = 0; + + virtual void send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority = 0) = 0; + + /// @deprecated: Use terminate() from transport_connection instead virtual void terminate( const address_v* remote ) = 0; + virtual void register_listener( transport_listener* listener ) = 0; }; Index: /source/ariba/utility/visual/DddVis.h =================================================================== --- /source/ariba/utility/visual/DddVis.h (revision 10647) +++ /source/ariba/utility/visual/DddVis.h (revision 10653) @@ -59,5 +59,4 @@ using std::pair; using std::make_pair; -using std::cout; using std::ostringstream; using ariba::utility::NodeID; Index: /source/ariba/utility/visual/OvlVis.h =================================================================== --- /source/ariba/utility/visual/OvlVis.h (revision 10647) +++ /source/ariba/utility/visual/OvlVis.h (revision 10653) @@ -54,5 +54,4 @@ using std::pair; using std::make_pair; -using std::cout; using std::ostringstream; using ariba::utility::KeyMapping; Index: /source/services/Makefile.am =================================================================== --- /source/services/Makefile.am (revision 10653) +++ /source/services/Makefile.am (revision 10653) @@ -0,0 +1,1 @@ +SUBDIRS = dht Index: /source/services/dht/Dht.cpp =================================================================== --- /source/services/dht/Dht.cpp (revision 10653) +++ /source/services/dht/Dht.cpp (revision 10653) @@ -0,0 +1,552 @@ +/* + * Dht.cpp + * + * Created on: 20.06.2012 + * Author: mario + */ + +#include "Dht.h" +#include + +namespace ariba_service { +namespace dht { + +use_logging_cpp(Dht) + +using namespace std; +using boost::date_time::second_clock; +using boost::posix_time::ptime; + +SystemEventType DhtRepublishEvent("DhtRepublishEvent"); +SystemEventType DhtCleanupEvent("DhtCleanupEvent"); + + +Dht::Dht(ariba::ServiceID serviceID, ariba::Node* node) : + serviceID(serviceID), + node(node), + cleanup_running(false), + listener(NULL) +{ + this->node->bind(this, serviceID); +} + +Dht::~Dht() +{ + this->node->unbind(this, serviceID); +} + + + +void Dht::put(const std::string& key, const std::string& value, uint16_t ttl) +{ + DhtMessage msg(DhtMessage::DhtPut, key, value, ttl); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::get(const std::string& key) +{ + DhtMessage msg(DhtMessage::DhtGet, key); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::atomic_put_and_get(const std::string& key, const std::string& value, uint16_t ttl) +{ + DhtMessage msg(DhtMessage::DhtPutAndGet, key, value, ttl); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + +void Dht::meet(const std::string& key, const std::string& value, uint16_t ttl_in_sec) +{ + // insert into meet_store + insert_into_table(meet_store, + key, + std::vector(1, value), + ttl_in_sec); + + // send message (and program republishing) + send_meet_message(key, value); +} + +void Dht::stop_meet(const std::string& key, const std::string& value) +{ + remove_from_table(meet_store, + key, + std::vector(1, value)); +} + +void Dht::remove(const std::string& key, const std::string& value) +{ + // send delete message + DhtMessage msg(DhtMessage::DhtRemove, key, value); + + handle_dht_message(msg, NodeID::UNSPECIFIED); +} + + + + +bool Dht::add_listener(DhtAnswerInterface* new_listener) +{ + if ( listener == NULL ) + { + listener = new_listener; + + return true; + } + else + { + return false; + } +} + +bool Dht::remove_listener(DhtAnswerInterface* new_listener) +{ + if (listener == new_listener) { + listener = NULL; + return true; + + } else { + return false; + } +} + + + + +//** PRIVATE FUNCTIONS **// + +void Dht::handle_dht_message(const DhtMessage& message, const NodeID& source) +{ + // send message closer to hashed key + NodeID addr = message.getHashedKey(); + + logging_debug("Processing DHT message..."); + + logging_debug("Dest Addr: " << addr.toString()); + + // * send closer, if possible * + const ariba::NodeID dest = node->sendMessageCloserToNodeID(message, addr, this->serviceID); + + logging_debug("Closer Node: " << dest.toString()); + + // couldn't send closer, so we are the closest node + // ---> * handle dht request * (store value, etc.) + if ( dest == NodeID::UNSPECIFIED ) + { + logging_debug("DHT: We are the closest node!"); + + switch (message.getType()) + { + case DhtMessage::DhtPut: + { + insert_into_table( + table, + message.getKey(), + message.getValues(), + message.getTTL()); + + break; + } + + case DhtMessage::DhtGet: + { + answer_dht_request(message.getKey(), source); + + break; + } + + case DhtMessage::DhtPutAndGet: + { + insert_into_table( + table, + message.getKey(), + message.getValues(), + message.getTTL()); + answer_dht_request(message.getKey(), source); + + break; + } + + case DhtMessage::DhtRemove: + { + remove_from_table(table, message.getKey(), message.getValues()); + + break; + } + } + } +} + + +void Dht::insert_into_table(DhtTableType& table, + const std::string& key, + const vector& values, + uint16_t ttl) +{ + DhtTableType::mapped_type& value_entries = table[key]; + + BOOST_FOREACH(const std::string& value, values) { + + // Debug output + logging_info("DHT: Inserting (" << key << ", " << value << ")"); + + // push the value for the given key (into the vector) + bool entry_updated = false; + for ( + DhtTableType::mapped_type::iterator position = value_entries.begin(); + position != value_entries.end(); + ++position) + { + if (position->get_value() == value) { + position->set_ttl(ttl); + entry_updated = true; + break; + } + } + + if (!entry_updated) { + value_entries.push_back(ValueEntry(value, ttl)); + } + } + + schedule_cleanup_event(); +} + + +void Dht::remove_from_table(DhtTableType& table, + const std::string& key, + const vector& values) +{ + logging_debug("DHT: trying to delete some values for key " << key); + // find key + DhtTableType::iterator key_position = table.find(key); + if (key_position == table.end()) { + return; + } + + // delete values from set of values + DhtTableType::mapped_type& entries = key_position->second; + BOOST_FOREACH(const std::string& value, values) { + for ( + DhtTableType::mapped_type::iterator entry = entries.begin(); + entry != entries.end(); + ++entry) + { + if (entry->get_value() == value) { + logging_info("DHT: Deleting " + "(" <get_value() << ")"); + entries.erase(entry); + break; + } + } + } + + // the key could empty now + // ---> remove it + if ( entries.size() == 0 ) + { + table.erase(key_position); + } +} + + +void Dht::cleanup_table(DhtTableType& table) +{ + logging_debug("DHT: cleaning up table"); + + vector to_be_deleted; + + for ( + DhtTableType::iterator position = table.begin(); + position != table.end(); + ++position) + { + cleanup_entries(position->second); + + // mark entry container for removal if empty + if (position->second.size() == 0) { + to_be_deleted.push_back(position->first); + } + } + + BOOST_FOREACH(const std::string& key, to_be_deleted) { + table.erase(key); + } +} + +void Dht::cleanup_entries(DhtTableType::mapped_type& entries) +{ + DhtTableType::mapped_type::iterator position = entries.begin(); + while (position != entries.end()) { + + if (position->is_ttl_elapsed()) { + // remove stale entry + position = entries.erase(position); + + } else { + // move on otherwise + ++position; + } + } +} + + +void Dht::answer_dht_request(const std::string& key, const NodeID& source) +{ + // get entries from table + const DhtTableType::mapped_type& entries = table[key]; + + // need to convert value entries to strings + vector values; + values.reserve(entries.size()); + BOOST_FOREACH(const ValueEntry& entry, entries) { + + if (!entry.is_ttl_elapsed()) { + values.push_back(entry.get_value()); + } + + } + + // BRANCH: request comes from another node + // ---> send answer message + if ( source != NodeID::UNSPECIFIED ) + { + // create answer message + DhtMessage msg(DhtMessage::DhtAnswer, key, values); + + // * send answer * + node->sendMessage(msg, source, serviceID); + } + + // BRANCH: local request + // ---> inform listeners directly (TODO code duplicates...) + else + { + logging_debug("DHT: Answering request for key '" << key << "' locally"); + + // * inform listeners * + if ( listener ) + { + listener->handle_dht_answer(key, values); + } + } + + + // an empty key could have been created + // ---> remove it + if ( entries.size() == 0 ) + { + table.erase(key); + } +} + + +void Dht::send_meet_message(const std::string& key, const std::string& value) +{ + // send put&get message + DhtMessage msg(DhtMessage::DhtPutAndGet, key, value, MEET_DHT_TTL); + + handle_dht_message(msg, NodeID::UNSPECIFIED); + + // program timer for republish (or deletion) + Key_Value* kv = new Key_Value; + kv->key = key; + kv->value = value; + + SystemQueue::instance().scheduleEvent( + SystemEvent( this, DhtRepublishEvent, kv), + MEET_REPUBLISH_INTERVAL * 1000 ); +} + + +void Dht::meet_update_event(const std::string& key, const std::string& value) +{ + // get entries from table + DhtTableType::mapped_type& entries = meet_store[key]; + + cleanup_entries(entries); + + // find right entry + BOOST_FOREACH(const ValueEntry& entry, entries) { + if (entry.get_value() == value) { + + // republish value + logging_debug("DHT: Republishing " + "(" << key << ", " << entry.get_value() << ")"); + send_meet_message(key, entry.get_value()); + } + } + + // an empty key could have been created + // ---> remove it + if ( entries.size() == 0 ) + { + meet_store.erase(key); + } +} + +void Dht::schedule_cleanup_event(bool reschedule) +{ + if (reschedule || !cleanup_running) { + SystemQueue::instance().scheduleEvent( + SystemEvent(this, DhtCleanupEvent), + CLEANUP_INTERVAL * 1000); + cleanup_running = true; + } +} + + +void Dht::print_dht() +{ + logging_debug("======== DHT ========"); + for ( DhtTableType::iterator dht_it = table.begin(); dht_it != table.end(); dht_it++) + { + logging_debug("Key: " << dht_it->first); + + for ( DhtTableType::mapped_type::iterator value_it = dht_it->second.begin(); + value_it != dht_it->second.end(); + value_it++ ) + { + logging_debug("--> " << value_it->get_value()); + } + + logging_debug("- - - - -"); + } + + logging_debug("======== [DHT] ========"); +} + + + +void Dht::onMessage(const ariba::DataMessage& msg, const ariba::NodeID& source, + const ariba::LinkID& lnk) +{ + logging_debug("DHT: Incoming message..."); + + DhtMessage* mess = msg.getMessage()->convert (); + + // handle message + switch (mess->getType()) + { + // BRANCH: Message is an Answer for our request + case DhtMessage::DhtAnswer: + { + logging_debug("DHT: Got answer for key '" << mess->getKey() << "'"); + + BOOST_FOREACH(string str, mess->getValues()) + { + logging_debug("--> Value: '" << str << "'"); + } + + // * inform listeners * + if ( listener ) + { + listener->handle_dht_answer(mess->getKey(), mess->getValues()); + } + + break; + } + + // BRANCH: Message is a Request + // ---> route or handle + default: + { + handle_dht_message(*mess, source); + + break; + } + } + + delete mess; +} + + +void Dht::handleSystemEvent( const SystemEvent& event ) +{ + + if (event.getType() == DhtRepublishEvent) { + logging_debug("DHT: Meet republish event!"); + + // republish meet entry + Key_Value* kv = event.getData(); + meet_update_event(kv->key, kv->value); + delete kv; + + } else if (event.getType() == DhtCleanupEvent) { + logging_debug("DHT: Cleanup event!"); + + cleanup_table(table); + schedule_cleanup_event(true); + } +} + + +/************** + * ValueEntry * + **************/ + +Dht::ValueEntry::ValueEntry( + const std::string& value, + uint16_t ttl) : + ttl(ttl), + last_update(second_clock::universal_time()), + value(value) +{ +} + + +void Dht::ValueEntry::refresh() { + last_update = second_clock::universal_time(); +} + + +const std::string& Dht::ValueEntry::get_value() const { + return value; +} + +uint16_t Dht::ValueEntry::get_age() const +{ + boost::posix_time::time_duration diff = + second_clock::universal_time() - last_update; + + return diff.total_seconds(); +} + +uint16_t Dht::ValueEntry::get_ttl() const { + return ttl; +} + +void Dht::ValueEntry::set_ttl(uint16_t ttl) { + this->refresh(); + this->ttl = ttl; +} + +bool Dht::ValueEntry::is_ttl_elapsed() const { + // ttl == 0 signals infinite lifetime + if (ttl == 0) { + return false; + } + + return second_clock::universal_time() >= + (last_update + boost::posix_time::seconds(ttl)); +} + +uint16_t Dht::ValueEntry::get_remaining_ttl() const +{ + if ( ttl == 0 ) + return -1; + + if ( is_ttl_elapsed() ) + return 0; + + boost::posix_time::time_duration diff = + (last_update + boost::posix_time::seconds(ttl)) - + second_clock::universal_time(); + + return ttl - get_age(); +} + +}} /* namespace ariba_service::dht */ Index: /source/services/dht/Dht.h =================================================================== --- /source/services/dht/Dht.h (revision 10653) +++ /source/services/dht/Dht.h (revision 10653) @@ -0,0 +1,211 @@ +/* + * Dht.h + * + * Created on: 20.06.2012 + * Author: mario + */ + +#ifndef DHT_H_ +#define DHT_H_ + +#include "ariba/ariba.h" +#include "ariba/utility/system/SystemQueue.h" +#include "ariba/utility/logging/Logging.h" +#include "messages/DhtMessage.h" +#include "DhtAnswerInterface.h" +#include +#include +#include + +namespace ariba_service { +namespace dht { + +using ariba::utility::SystemQueue; +using ariba::utility::SystemEvent; +using ariba::utility::SystemEventType; +using ariba::utility::SystemEventListener; + +#define MEET_REPUBLISH_INTERVAL 10 +#define MEET_DHT_TTL 30 +#define CLEANUP_INTERVAL (5 * 60) + +class Dht : + public ariba::CommunicationListener, + public ariba::utility::SystemEventListener, + public boost::noncopyable +{ +use_logging_h(Dht) +public: + Dht(ariba::ServiceID serviceID, ariba::Node* node); + virtual ~Dht(); + + /** + * Put the value into the DHT under the specified key + * + * @param key + * Key to put the value under + * @param value + * The value which is put + * @param ttl + * The lifetime of the entry in seconds. The value will be removed + * automatically when it expires + */ + void put( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Get the values specified by the key + * + * @param key + * Key of the values which should be fetched + */ + void get(const std::string& key); + + /** + * Put and get in one single operation + * + * @param key + * The key the value will be put under and retrieved from + * @param value + * The value is first put then all values for that key, including the + * one just inserted will be sent back + * @param ttl + * The lifetime of the entry in seconds. The value will be removed + * automatically when it expires + */ + void atomic_put_and_get( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Periodically put and get the value + * + * @param key + * The key the value will be put under and retrieved from + * @param value + * The value that will be periodically put into the DHT. The value is + * first put then all values for that key, including the one just + * inserted will be sent back + * @param ttl + * How long should we try to put the value periodically (measured in + * seconds). 0 means putting the value until stop_meet() is called + */ + void meet( + const std::string& key, + const std::string& value, + uint16_t ttl); + + /** + * Stop periodically pushing the value + */ + void stop_meet(const std::string& key, const std::string& value); + + /** + * Remove the value under the specified key + */ + void remove(const std::string& key, const std::string& value); + + /** + * Register a listener which is called when an answer is received + */ + bool add_listener(DhtAnswerInterface* new_listener); + + /** + * Unregister a listener + * + * @returns true if the handler was successfully unregistered, false if the + * listener was not registered + */ + bool remove_listener(DhtAnswerInterface* new_listener); + +protected: + /*** CommunicationListener interface ***/ + + /** + * Called when a message is incoming + * @param msg The data message that is received + * @param remote The remote node that sent the message + * @param lnk The link id of the link where the message is received + */ + virtual void onMessage(const ariba::DataMessage& msg, const ariba::NodeID& source, + const ariba::LinkID& lnk = ariba::LinkID::UNSPECIFIED); + + + virtual void handle_dht_message(const DhtMessage& message, const NodeID& source); + + /*** SystemEventListener interface ***/ + virtual void handleSystemEvent( const SystemEvent& event ); + + +private: + class ValueEntry { + public: + ValueEntry(const std::string& value, uint16_t ttl = 0); + + void refresh(); + + const std::string& get_value() const; + + uint16_t get_age() const; + + uint16_t get_ttl() const; + void set_ttl(uint16_t ttl); + bool is_ttl_elapsed() const; + uint16_t get_remaining_ttl() const; + + bool operator<(const ValueEntry& rhs) const; + + private: + uint16_t ttl; + boost::posix_time::ptime last_update; + std::string value; + }; + + struct Key_Value + { + string key; + string value; + }; + + +private: + void answer_dht_request(const std::string& key, const NodeID& source); + void send_meet_message(const std::string& key, const std::string& value); + void meet_update_event(const std::string& key, const std::string& value); + + // just for debug purpose + void print_dht(); + + + ariba::ServiceID serviceID; + ariba::Node* node; + + typedef std::map< std::string, std::vector > DhtTableType; + DhtTableType table; + DhtTableType meet_store; + + void insert_into_table(DhtTableType& table, + const std::string& key, + const vector& values, + uint16_t ttl); + void remove_from_table(DhtTableType& table, + const std::string& key, + const vector& values); + void cleanup_table(DhtTableType& table); + void cleanup_entries(DhtTableType::mapped_type& entries); + + void schedule_cleanup_event(bool reschedule = false); + bool cleanup_running; + + // AnswerListener + DhtAnswerInterface* listener; +}; + + +}} /* namespace ariba_service::dht */ + + +#endif /* DHT_H_ */ Index: /source/services/dht/DhtAnswerInterface.h =================================================================== --- /source/services/dht/DhtAnswerInterface.h (revision 10653) +++ /source/services/dht/DhtAnswerInterface.h (revision 10653) @@ -0,0 +1,17 @@ +#ifndef DHT_ANSWER_INTERFACE_H_ +#define DHT_ANSWER_INTERFACE_H_ + +namespace ariba_service { +namespace dht { + +class DhtAnswerInterface +{ +public: + virtual void handle_dht_answer(const std::string& key, const std::vector& values) = 0; + + virtual ~DhtAnswerInterface() {} +}; + +}} /* namespace ariba_service::dht */ + +#endif /* DHT_ANSWER_INTERFACE_H_ */ Index: /source/services/dht/Makefile.am =================================================================== --- /source/services/dht/Makefile.am (revision 10653) +++ /source/services/dht/Makefile.am (revision 10653) @@ -0,0 +1,60 @@ +# sources ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +lib_LTLIBRARIES = libariba_dht.la + +# compiler flags ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +# read the autotools book 11.4 for setting the versioning number +# this is different from normal versioning schemes and important +# to set correctly as the runtime linker chooses the correct lib +# depending on the versioning information here! This is not the +# project version number!! + +libariba_dht_la_LDFLAGS = -version-info 1:0:0 + +# compiler flags ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +AM_CPPFLAGS = -DLINUX -D_LINUX -I../../ -D_REENTRANT +AM_CPPFLAGS += $(BOOST_CPPFLAGS) + +if DEBUG +AM_CPPFLAGS += -ggdb -DDEBUG -D_DEBUG -O0 +endif + +if PROFILING +AM_CPPFLAGS += -pg +endif + +if OMNET +AM_CPPFLAGS += -fPIC -DUNDERLAY_OMNET +endif + +# linker flags ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +AM_LDFLAGS = -L../../ariba -lariba +AM_LDFLAGS += $(BOOST_SYSTEM_LDFLAGS) +LIBS += $(BOOST_SYSTEM_LIBS) + +if PROFILING +AM_LDFLAGS += -pg +endif + +if OMNET +AM_LDFLAGS += -shared -rdynamic +endif + +# sources ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +libariba_dht_ladir = ${pkgincludedir} + +#needed to fix autotools bug +ariba_dht_CPPFLAGS = $(AM_CPPFLAGS) + +libariba_dht_la_SOURCES = \ + Dht.cpp \ + messages/DhtMessage.cpp + +nobase_libariba_dht_la_HEADERS = \ + Dht.h \ + messages/DhtMessage.h \ + DhtAnswerInterface.h + Index: /source/services/dht/messages/DhtMessage.cpp =================================================================== --- /source/services/dht/messages/DhtMessage.cpp (revision 10653) +++ /source/services/dht/messages/DhtMessage.cpp (revision 10653) @@ -0,0 +1,48 @@ +#include "DhtMessage.h" + +#include + +namespace ariba_service { +namespace dht { + +vsznDefault(DhtMessage); + +DhtMessage::DhtMessage() : + ttl( 0 ), + replace( false ) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key ) : + type( static_cast(type) ), + ttl( 0 ), + replace( false ), + key( key ) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key, + const std::string& value, uint16_t ttl ) : + type( static_cast(type) ), + ttl( ttl ), + replace( false ), + key( key ), + values(1, value) +{} + +DhtMessage::DhtMessage( DhtMessageType type, const std::string& key, + const vector& values, uint16_t ttl ) : + type( static_cast(type) ), + ttl( ttl ), + replace( false ), + key( key ) +{ + // preallocate enough room so we don't need to copy a lot + this->values.reserve(values.size()); + BOOST_FOREACH(const std::string value, values ) + this->values.push_back( value ); +} + +DhtMessage::~DhtMessage() { + // empty +} + +}} Index: /source/services/dht/messages/DhtMessage.h =================================================================== --- /source/services/dht/messages/DhtMessage.h (revision 10653) +++ /source/services/dht/messages/DhtMessage.h (revision 10653) @@ -0,0 +1,121 @@ +#ifndef DHTMESSAGE_H_ +#define DHTMESSAGE_H_ + +#include "ariba/utility/messages.h" +#include "ariba/utility/serialization.h" +#include "ariba/Name.h" + +namespace ariba_service { +namespace dht { + +using ariba::utility::Message; +using_serialization; + +class DhtMessage : public Message { VSERIALIZEABLE +public: + typedef enum { + DhtInvalid = 0, + DhtGet = 1, + DhtPut = 2, + DhtPutAndGet = 3, + DhtRemove = 4, + DhtRepublish = 5, + DhtAnswer = 8 + } DhtMessageType; + + DhtMessage(); + DhtMessage( DhtMessageType type, const std::string& key ); + DhtMessage( DhtMessageType type, const std::string& key, + const std::string& value, uint16_t ttl = 0 ); + + DhtMessage( DhtMessageType type, const std::string& key, + const vector& values, uint16_t ttl = 0 ); + + virtual ~DhtMessage(); + + DhtMessageType getType() const { + return static_cast(type); + } + + NodeID getHashedKey() const { + return ariba::Name(key).toNodeId(); + } + + const std::string& getKey() const { + return key; + } + + /// returns the first element of the key vector + const std::string& getValue() const { + return values.at(0); + } + + /// return all values for the key + const vector& getValues() const { + return values; + } + + /// return all values for the key + vector& getValues() { + return values; + } + + bool hasValues() const { + return values.size() != 0; + } + + uint16_t getTTL() const { + return ttl; + } + + void setTTL( uint16_t ttl ) { + this->ttl = ttl; + } + + void setReplace( bool replace ) { + this->replace = replace; + } + + bool doReplace() const { + return replace; + } + + +private: + uint8_t type; + uint16_t ttl; + bool replace; + std::string key; + vector values; +}; + +}} // namespace ariba_service::dht + +sznBeginDefault( ariba_service::dht::DhtMessage, X ) { + X && type; + + // serialize tll + X && ttl; + + // key serialization + X && T(key); + + // store number of values + uint16_t num_values = values.size(); + X && num_values; + + // value serialization + for (size_t i=0; i