Index: source/ariba/utility/CMakeLists.txt =================================================================== --- source/ariba/utility/CMakeLists.txt (revision 10789) +++ source/ariba/utility/CMakeLists.txt (revision 12060) @@ -45,9 +45,9 @@ add_subdir_sources( addressing + addressing2 bootstrap configuration internal logging - measurement messages misc Index: source/ariba/utility/addressing/endpoint_set.hpp =================================================================== --- source/ariba/utility/addressing/endpoint_set.hpp (revision 10789) +++ source/ariba/utility/addressing/endpoint_set.hpp (revision 12060) @@ -1,4 +1,4 @@ -#ifndef ENDPOINT_SET_HPP_ -#define ENDPOINT_SET_HPP_ +#ifndef ENDPOINT_SET_HPP_DEPRECATED_ +#define ENDPOINT_SET_HPP_DEPRECATED_ #include "addressing.hpp" Index: source/ariba/utility/addressing/facades/address_v.hpp =================================================================== --- source/ariba/utility/addressing/facades/address_v.hpp (revision 10789) +++ source/ariba/utility/addressing/facades/address_v.hpp (revision 12060) @@ -13,4 +13,6 @@ #include "../detail/address_convenience.hpp" + +#include namespace ariba { @@ -26,4 +28,6 @@ class address_v: public detail::address_convenience { public: + typedef boost::shared_ptr shared_ptr; + virtual ~address_v() {} Index: source/ariba/utility/addressing/ip_address.hpp =================================================================== --- source/ariba/utility/addressing/ip_address.hpp (revision 10789) +++ source/ariba/utility/addressing/ip_address.hpp (revision 12060) @@ -1,4 +1,4 @@ -#ifndef IP_ADDRESS_HPP_ -#define IP_ADDRESS_HPP_ +#ifndef IP_ADDRESS_HPP_DEPRECATED_ +#define IP_ADDRESS_HPP_DEPRECATED_ #include Index: source/ariba/utility/addressing/tcpip_endpoint.hpp =================================================================== --- source/ariba/utility/addressing/tcpip_endpoint.hpp (revision 10789) +++ source/ariba/utility/addressing/tcpip_endpoint.hpp (revision 12060) @@ -1,4 +1,4 @@ -#ifndef TCPIP_ENDPOINT_HPP_ -#define TCPIP_ENDPOINT_HPP_ +#ifndef TCPIP_ENDPOINT_HPP_DEPRECATED_ +#define TCPIP_ENDPOINT_HPP_DEPRECATED_ #include Index: source/ariba/utility/addressing2/CMakeLists.txt =================================================================== --- source/ariba/utility/addressing2/CMakeLists.txt (revision 12060) +++ source/ariba/utility/addressing2/CMakeLists.txt (revision 12060) @@ -0,0 +1,48 @@ +# [License] +# The Ariba-Underlay Copyright +# +# Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +# +# Institute of Telematics +# UniversitÀt Karlsruhe (TH) +# Zirkel 2, 76128 Karlsruhe +# Germany +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OF TELEMATICS OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +# +# The views and conclusions contained in the software and documentation +# are those of the authors and should not be interpreted as representing +# official policies, either expressed or implied, of the Institute of +# Telematics. +# [License] + +add_headers( + endpoint.hpp + endpoint_set.hpp + tcpip_endpoint.hpp +) + +add_sources( + endpoint_set.cpp + tcpip_endpoint.cpp +) Index: source/ariba/utility/addressing2/endpoint.hpp =================================================================== --- source/ariba/utility/addressing2/endpoint.hpp (revision 12060) +++ source/ariba/utility/addressing2/endpoint.hpp (revision 12060) @@ -0,0 +1,79 @@ +/* + * endpoint.h + * + * Created on: 26.03.2013 + * Author: mario + */ + +#ifndef ENDPOINT_HPP_ +#define ENDPOINT_HPP_ + +// system +#include +#include + +// boost +#include + +namespace ariba { +namespace addressing2 { + +struct endpoint_category +{ + enum _ENDPOINT_CATEGORY { + INVALID = 0, + TCPIP = 1, + BLUETOOTH = 2 + }; + +// static const SEND_PRIORITY_INTERNAL OVERLAY = HIGHEST; +}; +typedef endpoint_category::_ENDPOINT_CATEGORY ENDPOINT_CATEGORY; + +struct endpoint_type +{ + enum _ENDPOINT_TYPE { + INVALID = 0, + TCPIPv4 = 1, + TCPIPv6 = 2 + }; +}; +typedef endpoint_type::_ENDPOINT_TYPE ENDPOINT_TYPE; + + +class endpoint +{ +public: + virtual ~endpoint() {} + + virtual ENDPOINT_CATEGORY get_category() const = 0; + virtual ENDPOINT_TYPE get_type() const = 0; + + /** + * @return Human readable string representation of this endpoint. + */ + virtual std::string to_string() const = 0; + + /** + * Serializes this »endpoint« into a byte_array. + * + * @param buffer: An array >= »endpoint::size()« + * @return »endpoint::size()« + */ + virtual size_t to_byte_array(uint8_t* buffer) const = 0; + + /** + * @return Number of bytes: + * - read in »endpoint::endpoint(const uint8_t* byte_array);« + * - to be written in »endpoint::to_byte_array(uint8_t* buffer)« + */ + virtual int size() const = 0; +}; + +typedef boost::shared_ptr EndpointPtr; +//typedef boost::shared_ptr const_EndpointPtr; + +// NOTE: An endpoint is designed as an unmutable object! + +}} /* namespace ariba::addressing2 */ +#endif /* ENDPOINT_HPP_ */ Index: source/ariba/utility/addressing2/endpoint_set.cpp =================================================================== --- source/ariba/utility/addressing2/endpoint_set.cpp (revision 12060) +++ source/ariba/utility/addressing2/endpoint_set.cpp (revision 12060) @@ -0,0 +1,344 @@ +/* + * endpointset.cpp + * + * Created on: 27.03.2013 + * Author: mario + */ + +#include "endpoint_set.hpp" + +// ariba endpoints +#include "tcpip_endpoint.hpp" + +// boost +#include +#include + +#include + +namespace ariba { +namespace addressing2 { + +using namespace std; +using boost::property_tree::ptree; + +// TODO maybe this function wants to go in a distinct header.. +template +T read_from_byte_array(const uint8_t*& buff, int& read_max) +{ + assert ( sizeof(T) <= read_max ); + + const uint8_t* b1 = buff; + buff += sizeof(T); + read_max -= sizeof(T); + + return *( reinterpret_cast(b1) ); +} + + +/* factories */ +//shared_ptr endpoint_set::create_EndpointSet(const std::string & str) +//{ +// EndpointSetPtr set(new endpoint_set(str)); +// +// return set; +//} + +shared_ptr endpoint_set::create_EndpointSet(const ptree& pt) +{ + EndpointSetPtr set(new endpoint_set(pt)); + + return set; +} + +shared_ptr endpoint_set::create_EndpointSet() +{ + EndpointSetPtr set(new endpoint_set()); + + return set; +} + + + +endpoint_set::endpoint_set() +{ +} + +endpoint_set::endpoint_set(const ptree& pt) +{ + /* create & add endpoints */ + BOOST_FOREACH( const ptree::value_type& child, pt ) + { + string cat = child.second.get("category"); + + // TCPIP + if ( cat == "TCPIP" ) + { + string addr = child.second.get("addr"); + int port = child.second.get("port"); + + TcpIP_EndpointPtr endp(new tcpip_endpoint(addr, port)); + + tcpip_endpoints.push_back(endp); + } + + // TODO else if bluetooth + } +} + +//endpoint_set::endpoint_set(const string& str) +//{ +// // TODO see endpoint_set(const ptree& pt) +// +// +// ptree pt; +// +// /* parse input string */ +// // input string format is: JSON +// if (str.substr(0, 4) == "JSON") +// { +// istringstream sstream(str.substr(4, string::npos)); +// boost::property_tree::json_parser::read_json(sstream, pt); +// } +// // --- other formats can be supported here (e.g. XML) --- +// else +// { +// throw invalid_argument("Could not parse endpoint_set from string."); +// } +// +// +// +// /* create & add endpoints */ +// BOOST_FOREACH(const ptree::value_type& child, pt.get_child("endpoint_set")) +// { +// string cat = child.second.get("category"); +// +// // TCPIP +// if ( cat == "TCPIP" ) +// { +// string addr = child.second.get("addr"); +// int port = child.second.get("port"); +// +// TcpIP_EndpointPtr endp(new tcpip_endpoint(addr, port)); +// +// tcpip_endpoints.push_back(endp); +// } +// +// // TODO else if bluetooth +// } +//} + +endpoint_set::~endpoint_set() +{ +} + +void endpoint_set::add_endpoint(EndpointPtr endpoint) +{ + switch ( endpoint->get_category() ) + { + case endpoint_category::TCPIP: + { + // TODO try-catch --> log a warning + shared_ptr tcpip_endp = + boost::dynamic_pointer_cast(endpoint); + + // no duplicates + bool duplicate = false; + BOOST_FOREACH(const shared_ptr& x, tcpip_endpoints) + { + if ( tcpip_endp->equals(x) ) + { + duplicate = true; + break; + } + } + + if ( ! duplicate ) + { + // * add * + tcpip_endpoints.push_back(tcpip_endp); + } + + break; + } + + default: + { + // TODO log a warning ^^ + + break; + } + } +} + +void endpoint_set::add_endpoints(const shared_ptr endpoints) +{ + // TODO bluetooth, etc.... + + BOOST_FOREACH( EndpointPtr endp, endpoints->get_tcpip_endpoints() ) + { + add_endpoint(endp); + } +} + + +const vector >& endpoint_set::get_tcpip_endpoints() const +{ + return tcpip_endpoints; +} + +//const vector > endpoint_set::get_tcpip_endpoints() const +//{ +// vector > ret; +// ret.reserve(tcpip_endpoints.size()); +// +// BOOST_FOREACH( const_TcpIP_EndpointPtr address, tcpip_endpoints ) +// { +// ret.push_back(address); +// } +// +// return ret; +//} + + +string endpoint_set::to_string() const +{ + ostringstream out; + + BOOST_FOREACH( EndpointPtr endp, tcpip_endpoints ) + { + out << endp->to_string() << "; "; + } + + return out.str(); +} + + +/** + * Format: + * + * | 16-bit: overall size | + ( | 8 bit: type | + | variable length: endpoint | ) * N + * + */ +reboost::shared_buffer_t endpoint_set::serialize() const +{ + // TODO bluetooth, etc.... + + /* calculate size */ + // size: two byte length field + size_t overall_size = sizeof(uint16_t); + + BOOST_FOREACH( EndpointPtr endp, tcpip_endpoints ) + { + // size: endpoint + type + overall_size += endp->size() + 1; + } + + // overall_size value must fit into 16 bits + assert ( overall_size <= numeric_limits::max() ); + + + /* serialize */ + reboost::shared_buffer_t buff(overall_size); + uint8_t* buff_ptr = buff.mutable_data(); + + // overall size + memcpy( buff_ptr, &overall_size, sizeof(uint16_t) ); + buff_ptr += sizeof(uint16_t); + + // tcpip_endpoints + BOOST_FOREACH( EndpointPtr endp, tcpip_endpoints ) + { +// // XXX AKTUELL BUG FINDING... +// cout << " - SERIALIZE: (" << (int) (buff_ptr - buff.mutable_data()) << ")"; + + // type + *buff_ptr = static_cast(endp->get_type()); + buff_ptr++; + + // serialize tcpip_endpoint + buff_ptr += endp->to_byte_array(buff_ptr); + +// // XXX AKTUELL BUG FINDING... +// cout << endp->to_string() << " (" << (int) (buff_ptr - buff.mutable_data()) << ")" << endl; + } + + // boundary check + assert( buff_ptr <= buff.mutable_data() + buff.size() ); + + return buff; +} + +reboost::shared_buffer_t endpoint_set::deserialize(reboost::shared_buffer_t buff) +{ + assert( tcpip_endpoints.size() == 0); + + const uint8_t* buff_ptr = buff.data(); // NOTE: the data is const, the pointer is not. + + // read overall size (16 bit value) + int bytes_left = sizeof(uint16_t); + uint16_t overall_size = read_from_byte_array(buff_ptr, bytes_left); + + // check claimed overall size + if ( overall_size > buff.size() ) + { + // todo throw + cout << endl << "FATAL ERROR in »endpoint_set::deserialize«: overall_size > buff.size()" << endl; + assert ( false ); + } + + // calculate bytes to read + bytes_left = overall_size - sizeof(uint16_t); + + + // read endpoints + while ( bytes_left > 0 ) + { +// // XXX AKTUELL BUG FINDING... +// cout << " - DESERIALIZE: (" << (int) (buff_ptr - buff.data()) << ")"; + + ENDPOINT_TYPE type = static_cast( + read_from_byte_array(buff_ptr, bytes_left) ); + + switch (type) + { + case endpoint_type::TCPIPv4: + case endpoint_type::TCPIPv6: + { + // TODO try catch + TcpIP_EndpointPtr endp(new tcpip_endpoint(type, buff_ptr, bytes_left)); + add_endpoint(endp); + + // move pointers + const int bytes_read = endp->size(); + buff_ptr += bytes_read; + bytes_left -= bytes_read; + + +// // XXX AKTUELL BUG FINDING... +// cout << endp->to_string() << " (" << (int) (buff_ptr - buff.data()) << ")" << endl; + + break; + } + + // TODO case endpoint_type::bluetooth...: + + default: + { + // TODO throw + cout << endl << "FATAL ERROR in »endpoint_set::deserialize«: Unknown type (" << type << ")" << endl; + cout << "BUFFER (size = " << buff.size() << "):" << endl; + cout << buff << endl << "-------------------------" << endl; + assert(false); + break; + } + } + + assert( bytes_left >= 0 ); + } + + // return sub-buffer + return buff(overall_size); +} + +}} /* namespace addressing2::ariba */ Index: source/ariba/utility/addressing2/endpoint_set.hpp =================================================================== --- source/ariba/utility/addressing2/endpoint_set.hpp (revision 12060) +++ source/ariba/utility/addressing2/endpoint_set.hpp (revision 12060) @@ -0,0 +1,110 @@ +/* + * endpoint_set.hpp + * + * Created on: 27.03.2013 + * Author: mario + */ + +#ifndef ENDPOINT_SET_HPP_ +#define ENDPOINT_SET_HPP_ + +// ariba +#include "endpoint.hpp" +#include "ariba/utility/transport/messages/shared_buffer.hpp" + +// boost +#include +#include + +// system +#include +#include + +namespace ariba { + +using boost::property_tree::ptree; +using boost::shared_ptr; + +namespace addressing2 { + +/// forward declarations +class tcpip_endpoint; + +/** + * Holds a number of endpoints. (Usually of one particular node.) + */ +class endpoint_set +{ +public: + + // factories + static shared_ptr create_EndpointSet(const boost::property_tree::ptree& pt); +// static shared_ptr create_EndpointSet(const std::string& str); + static shared_ptr create_EndpointSet(); + + endpoint_set(); + +// endpoint_set(const std::string& str); + endpoint_set(const boost::property_tree::ptree& pt); + virtual ~endpoint_set(); + + + /** + * Adds an endpoint to this set. + * + * (Ignores duplicates) + */ + void add_endpoint(EndpointPtr endpoint); + + /** + * Adds all endpoints from the given set to this set. + * + * (Ignores duplicates) + */ + void add_endpoints(const boost::shared_ptr endpoints); + + /** + * @return Human readable string representation of this endpoint-set. + */ + std::string to_string() const; + + /** + * @return a vector of all tcpip_endpoints in this set + */ + const std::vector >& get_tcpip_endpoints() const; + + /** + * This function serializes this endpoint_set with all its endpoints + * into a shared buffer, which then can be sent to another node. + * + * @return A byte-representation of this set saved into a shared_buffer. + */ + reboost::shared_buffer_t serialize() const; + + /** + * Recreates an endpoint_set from a shared buffer. + * + * ---> Complementary to »endpoint_set::serialize_into_shared_buffer()« + * + * @return remaining sub-buffer + */ + reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff); + + + /** + * @return total number of endpoints in this set + */ + int count() const + { + return tcpip_endpoints.size(); + } + +private: + std::vector > tcpip_endpoints; +}; + +typedef boost::shared_ptr EndpointSetPtr; +typedef boost::shared_ptr const_EndpointSetPtr; + +}} /* namespace addressing2::ariba */ +#endif /* ENDPOINT_SET_HPP_ */ Index: source/ariba/utility/addressing2/tcpip_endpoint.cpp =================================================================== --- source/ariba/utility/addressing2/tcpip_endpoint.cpp (revision 12060) +++ source/ariba/utility/addressing2/tcpip_endpoint.cpp (revision 12060) @@ -0,0 +1,182 @@ +/* + * tcpip_endpoint.cpp + * + * Created on: 26.03.2013 + * Author: mario + */ + +#include "tcpip_endpoint.hpp" + +// system +#include +#include + +using namespace std; +using namespace boost::asio::ip; + +namespace ariba { +namespace addressing2 { + +#define TCPIPv6_SIZE 18 +#define TCPIPv4_SIZE 6 + + +/// factories +TcpIP_EndpointPtr tcpip_endpoint::create_TcpIP_Endpoint( + const tcp::endpoint& asio_endpoint) +{ + TcpIP_EndpointPtr ptr(new tcpip_endpoint(asio_endpoint)); + + return ptr; +} + + +/// constructors +tcpip_endpoint::tcpip_endpoint(const std::string& ip_addr, const int port) +{ + address addr = address::from_string(ip_addr); + asio_endpoint = tcp::endpoint(addr, static_cast(port)); +} + +tcpip_endpoint::tcpip_endpoint(const ENDPOINT_TYPE type, const uint8_t* const byte_array, const int read_max) +{ + // IPv4 + if ( type == endpoint_type::TCPIPv4 ) + { + // boundary check + if ( read_max < TCPIPv4_SIZE ) + throw invalid_argument("Not enough bytes to read an TCPIPv4 endpoint."); + + asio_endpoint = tcp::endpoint( address_v4( *((uint32_t*) byte_array) ), + *((uint16_t*) (byte_array+sizeof(uint32_t))) ); + } + + // IPv6 + else if ( type == endpoint_type::TCPIPv6 ) + { + // boundary check + if ( read_max < TCPIPv6_SIZE ) + throw invalid_argument("Not enough bytes to read an TCPIPv6 endpoint."); + + address_v6::bytes_type bytes_; + for (size_t i=0; iasio_endpoint; +} + +}} /* namespace ariba::addressing2 */ Index: source/ariba/utility/addressing2/tcpip_endpoint.hpp =================================================================== --- source/ariba/utility/addressing2/tcpip_endpoint.hpp (revision 12060) +++ source/ariba/utility/addressing2/tcpip_endpoint.hpp (revision 12060) @@ -0,0 +1,92 @@ +/* + * tcpip_endpoint.h + * + * Created on: 26.03.2013 + * Author: mario + */ + +#ifndef TCPIP_ENDPOINT_HPP_ +#define TCPIP_ENDPOINT_HPP_ + +// TODO: maybe move this class ariba/utility/transport.. ? + +#include "endpoint.hpp" + +// boost +#include +#include + +using boost::shared_ptr; + +namespace ariba { +namespace addressing2 { + +class tcpip_endpoint: public endpoint +{ +public: + /** + * Implements »endpoint« + * @see endpoint.h + * + * NOTE: An endpoint is designed as an unmutable object! + */ + + /// factories + static boost::shared_ptr create_TcpIP_Endpoint( + const boost::asio::ip::tcp::endpoint& asio_endpoint); + + + /// constructors + + /** + * Creates a new »tcpip_endpoint« from a string representation. + * ---> NOT complementary to »to_string()« + */ + tcpip_endpoint(const std::string& ip_addr, const int port); + + /** + * Creates a new »tcpip_endpoint« from a byte array. + * ---> Complementary to »to_byte_array()« + * + * May throw »invalid_argument« exception. + */ + tcpip_endpoint(const ENDPOINT_TYPE type, const uint8_t* const byte_array, const int read_max); + + /** + * Creates a new »tcpip_endpoint« from an ASIO tcp::endpoint. + * ---> Complementary to »tcpip_endpoint::to_asio()« + */ + tcpip_endpoint(const boost::asio::ip::tcp::endpoint& asio_endpoint); + + virtual ~tcpip_endpoint(); + + + /// from superclass + virtual ENDPOINT_CATEGORY get_category() const; + virtual ENDPOINT_TYPE get_type() const; + virtual std::string to_string() const; + virtual size_t to_byte_array(uint8_t* buffer) const; + virtual int size() const; + + + /** + * @return An equivalent ASIO tcp::endpoint + */ + const boost::asio::ip::tcp::endpoint& to_asio() const; + + /** + * @return »true« if rhs points to the same Address:Port pair, + * otherwise returns »false«. + */ + bool equals(const shared_ptr& rhs) const; + + +private: + boost::asio::ip::tcp::endpoint asio_endpoint; +}; + +typedef boost::shared_ptr TcpIP_EndpointPtr; +//typedef boost::shared_ptr const_TcpIP_EndpointPtr; + +}} /* namespace ariba::addressing2 */ +#endif /* TCPIP_ENDPOINT_HPP_ */ Index: source/ariba/utility/bootstrap/modules/periodicbroadcast/PeriodicBroadcast.h =================================================================== --- source/ariba/utility/bootstrap/modules/periodicbroadcast/PeriodicBroadcast.h (revision 10789) +++ source/ariba/utility/bootstrap/modules/periodicbroadcast/PeriodicBroadcast.h (revision 12060) @@ -40,21 +40,28 @@ #define __PERIODIC_BROADCAST_H +// ariba #include "ariba/config.h" - +#include "ariba/utility/bootstrap/modules/BootstrapModule.h" +#include "ariba/utility/logging/Logging.h" +#include "ariba/utility/system/Timer.h" + +// ariba messages +#include "PeriodicBroadcastMessage.h" + +// (ariba) link-local +#include "ariba/utility/transport/StreamTransport/StreamTransport.hpp" + +// system #include #include #include #include + +// boost #include #include #include #include -#include "ariba/utility/bootstrap/modules/BootstrapModule.h" -#include "ariba/utility/logging/Logging.h" -#include "ariba/utility/system/Timer.h" -#include "PeriodicBroadcastMessage.h" - -//link-local -#include "ariba/utility/transport/tcpip/tcpip.hpp" + using std::map; @@ -301,5 +308,5 @@ // include all link-local interfaces - vector scope_ids = ariba::transport::tcpip::get_interface_scope_ids(); + vector scope_ids = ariba::transport::get_interface_scope_ids(); BOOST_FOREACH ( uint64_t id, scope_ids ) Index: source/ariba/utility/logging/Logging.h =================================================================== --- source/ariba/utility/logging/Logging.h (revision 10789) +++ source/ariba/utility/logging/Logging.h (revision 12060) @@ -97,4 +97,5 @@ static int __loglevel__ = 2; //default is info +// static int __loglevel__ = 1; // XXX use higher log level #define logging_trace(x) { logging_stdout(x); } Index: source/ariba/utility/messages.h =================================================================== --- source/ariba/utility/messages.h (revision 10789) +++ source/ariba/utility/messages.h (revision 12060) @@ -40,10 +40,13 @@ #define MESSAGES_H_ +// TODO wÌrde sagen das brauchen wir nicht mehr +// ---> Ìberall dieses include rausnehmen und ggf #include "messages/Message.h" einfÌgen.. + #include "messages/Message.h" -#include "messages/MessageSender.h" -#include "messages/MessageReceiver.h" -#include "messages/MessageUtilities.h" -#include "messages/MessageProvider.h" -#include "messages/TextMessage.h" +//#include "messages/MessageSender.h" // TODO wird das noch genutzt..? Wenn nein, sollte es weg! +//#include "messages/MessageReceiver.h" +//#include "messages/MessageUtilities.h" +////#include "messages/MessageProvider.h" // DEPRECATED +//#include "messages/TextMessage.h" #endif /* MESSAGES_H_ */ Index: source/ariba/utility/messages/Message.cpp =================================================================== --- source/ariba/utility/messages/Message.cpp (revision 10789) +++ source/ariba/utility/messages/Message.cpp (revision 12060) @@ -41,4 +41,6 @@ #include "ariba/utility/serialization/DataStream.hpp" +#include "ariba/utility/logging/Logging.h" + NAMESPACE_BEGIN @@ -80,15 +82,59 @@ } + +reboost::message_t Message::wrap_up_for_sending() +{ + assert( ! wrapped_up ); + wrapped_up = true; + + //// Adapt to new message system //// + Data data = data_serialize(this, DEFAULT_V); + reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8); + + newstyle_payload.push_front(buf); + + return newstyle_payload; +} + +reboost::shared_buffer_t Message::serialize_into_shared_buffer() +{ + assert ( newstyle_payload.length() == 0 ); + + //// Adapt to new message system //// + Data data = data_serialize(this, DEFAULT_V); + reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8); + + return buf; +} + + +reboost::shared_buffer_t Message::deserialize_from_shared_buffer(reboost::shared_buffer_t buff) +{ + // NOTE: legacy payload is not allowed when using shared buffers + this->legacy_payload_disabled = true; + + assert( buff.size() > 0 ); + + // const_cast is necessary here, but without legacy payload we should be save here (more or less) + Data dat(const_cast(buff.data()), buff.size() * 8); + + size_t len = this->SERIALIZATION_METHOD_NAME(DESERIALIZE, dat) / 8; + + // return remaining sub-buffer + return buff(len); +} + + + NAMESPACE_END std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg ) { using_serialization; - stream << "msg(type=" << typeid(msg).name(); + stream << "msg(type=" << typeid(msg).name() << ","; stream << "len=" << (data_length(&msg)/8) << ","; Data data = data_serialize(&msg); - stream << ",data=" << data; + stream << "data=" << data; data.release(); stream << ")"; return stream; } - Index: source/ariba/utility/messages/Message.h =================================================================== --- source/ariba/utility/messages/Message.h (revision 10789) +++ source/ariba/utility/messages/Message.h (revision 12060) @@ -62,4 +62,8 @@ #include "ariba/utility/serialization.h" +// reboost messages +#include "ariba/utility/transport/messages/message.hpp" + + std::ostream& operator<<(std::ostream& stream, const ariba::utility::Message& msg ); @@ -82,10 +86,12 @@ friend std::ostream& ::operator<<(std::ostream& stream, const ariba::utility::Message& msg ); - // root binary data - shared_array root; - // payload + bool legacy_payload_disabled; bool releasePayload; Data payload; //< messages binary data + + // XXX testing... + reboost::message_t newstyle_payload; + bool wrapped_up; // addresses and control info @@ -98,5 +104,6 @@ */ inline Message() : - root(), releasePayload(true), payload(), srcAddr(NULL),destAddr(NULL) { + legacy_payload_disabled(false), releasePayload(true), payload(), + newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) { } @@ -106,5 +113,6 @@ */ explicit inline Message( const Data& data ) : - releasePayload(true), srcAddr(NULL),destAddr(NULL) { + legacy_payload_disabled(false), releasePayload(true), + newstyle_payload(), wrapped_up(false), srcAddr(NULL),destAddr(NULL) { // FIXME newstyle_payload..? this->payload = data.clone(); // this->root = shared_array((uint8_t*)data.getBuffer()); @@ -225,4 +233,37 @@ return decapsulate(); } + + + // XXX testing + void set_payload_message(reboost::message_t msg) + { + newstyle_payload = msg; + } + + void append_buffer(reboost::shared_buffer_t buff) + { + newstyle_payload.push_back(buff); + } + + + // XXX testing... packs this message into the payload message (do not use twice!!) + virtual reboost::message_t wrap_up_for_sending(); + + + /** + * Uses the old serialization system to serialize itself into a (new style) shared buffer. + */ + virtual reboost::shared_buffer_t serialize_into_shared_buffer(); + + /* + * XXX experimental + * + * Uses the old serialization system to deserialize itself out of a (new style) shared buffer. + * @return remaining sub-buffer (the "payload") + * + * Note: This is some kind of a hack! handle with care. + */ + virtual reboost::shared_buffer_t deserialize_from_shared_buffer(reboost::shared_buffer_t buff); + protected: @@ -262,5 +303,8 @@ * @return A explicit payload serializer */ - finline PayloadSerializer Payload( size_t length = ~0 ) { + finline PayloadSerializer Payload( size_t length = ~0 ) + { +// assert( ! legacy_payload_disabled ); // FIXME aktuell + return PayloadSerializer( this, length ); } Index: source/ariba/utility/messages/MessageProvider.cpp =================================================================== --- source/ariba/utility/messages/MessageProvider.cpp (revision 10789) +++ source/ariba/utility/messages/MessageProvider.cpp (revision 12060) @@ -37,32 +37,32 @@ // [License] -#include "MessageProvider.h" - -NAMESPACE_BEGIN - -MessageProvider::MessageProvider() { -} - -MessageProvider::~MessageProvider() { -} - -bool MessageProvider::sendMessageToReceivers( const Message* message ) { - bool sent = false; - for (size_t i=0; ireceiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true; - return sent; -} - -void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) { - receivers.push_back(receiver); -} - -void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) { - for (size_t i=0; ireceiveMessage(message, LinkID::UNSPECIFIED, NodeID::UNSPECIFIED)) sent = true; +// return sent; +//} +// +//void MessageProvider::addMessageReceiver( MessageReceiver* receiver ) { +// receivers.push_back(receiver); +//} +// +//void MessageProvider::removeMessageReceiver( MessageReceiver* receiver ) { +// for (size_t i=0; i - -using std::vector; -using ariba::utility::LinkID; -using ariba::utility::NodeID; - -NAMESPACE_BEGIN - - -/** - * This class defines an interface for message providers. - * Implementing classes must allow receivers to register themselves. - * - * @author Sebastian Mies - */ -class MessageProvider { -private: - vector receivers; - -protected: - bool sendMessageToReceivers( const Message* message ); - -public: - /** - * Constructor. - */ - MessageProvider(); - - /** - * Destructor. - */ - ~MessageProvider(); - - /** - * Adds a message receiver. - * - * @param receiver The receiver. - */ - void addMessageReceiver( MessageReceiver* receiver ); - - /** - * Removes a message receiver. - * - * @param receiver The receiver. - */ - void removeMessageReceiver( MessageReceiver* receiver ); -}; - -NAMESPACE_END +//#include "_namespace.h" +//#include "MessageReceiver.h" +//#include "ariba/utility/types/LinkID.h" +//#include "ariba/utility/types/NodeID.h" +//#include +// +//using std::vector; +//using ariba::utility::LinkID; +//using ariba::utility::NodeID; +// +//NAMESPACE_BEGIN +// +// +///** +// * This class defines an interface for message providers. +// * Implementing classes must allow receivers to register themselves. +// * +// * @author Sebastian Mies +// */ +//class MessageProvider { +//private: +// vector receivers; +// +//protected: +// bool sendMessageToReceivers( const Message* message ); +// +//public: +// /** +// * Constructor. +// */ +// MessageProvider(); +// +// /** +// * Destructor. +// */ +// ~MessageProvider(); +// +// /** +// * Adds a message receiver. +// * +// * @param receiver The receiver. +// */ +// void addMessageReceiver( MessageReceiver* receiver ); +// +// /** +// * Removes a message receiver. +// * +// * @param receiver The receiver. +// */ +// void removeMessageReceiver( MessageReceiver* receiver ); +//}; +// +//NAMESPACE_END #endif /* MESSAGEPROVIDER_H_ */ Index: source/ariba/utility/messages/MessageReceiver.cpp =================================================================== --- source/ariba/utility/messages/MessageReceiver.cpp (revision 10789) +++ source/ariba/utility/messages/MessageReceiver.cpp (revision 12060) @@ -49,8 +49,8 @@ } -bool MessageReceiver::receiveMessage( const Message* message, const LinkID& link, const NodeID& node ) { - //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl; - return false; -} +//bool MessageReceiver::receiveMessage( reboost::shared_buffer_t message, const LinkID& link, const NodeID& node ) { +// //std::cout << "UNIMPLEMENTED MessageReceiver got Message:" << (Message*)message << std::endl; +// return false; +//} NAMESPACE_END Index: source/ariba/utility/messages/MessageReceiver.h =================================================================== --- source/ariba/utility/messages/MessageReceiver.h (revision 10789) +++ source/ariba/utility/messages/MessageReceiver.h (revision 12060) @@ -40,5 +40,7 @@ #define MESSAGERECEIVER_H__ -#include "ariba/utility/messages/Message.h" +//#include "ariba/utility/messages/Message.h" +// reboost messages +#include "ariba/utility/transport/messages/message.hpp" #include "ariba/utility/types/LinkID.h" #include "ariba/utility/types/NodeID.h" @@ -73,5 +75,8 @@ * @return True, when the message has been accepted. */ - virtual bool receiveMessage( const Message* message, const LinkID& link, const NodeID& node ); + virtual bool receiveMessage( reboost::shared_buffer_t message, + const LinkID& link, + const NodeID& node, + bool bypass_overlay ) = 0; }; Index: source/ariba/utility/misc/Demultiplexer.hpp =================================================================== --- source/ariba/utility/misc/Demultiplexer.hpp (revision 10789) +++ source/ariba/utility/misc/Demultiplexer.hpp (revision 12060) @@ -136,5 +136,5 @@ SERVICE_LISTENER_MAP_CITERATOR it = mapServiceListener.find( id ); - if( it == mapServiceListener.end() ) return NULL; + if( it == mapServiceListener.end() ) return 0; else return it->second; } Index: source/ariba/utility/system/StartupWrapper.cpp =================================================================== --- source/ariba/utility/system/StartupWrapper.cpp (revision 10789) +++ source/ariba/utility/system/StartupWrapper.cpp (revision 12060) @@ -36,4 +36,6 @@ // Telematics. // [License] + +// XXX NOTE: Use this class with caution! Config support is outdated. #include "StartupWrapper.h" Index: source/ariba/utility/system/StartupWrapper.h =================================================================== --- source/ariba/utility/system/StartupWrapper.h (revision 10789) +++ source/ariba/utility/system/StartupWrapper.h (revision 12060) @@ -36,4 +36,6 @@ // Telematics. // [License] + +// XXX NOTE: Use this class with caution! Config support is outdated. #ifndef __STARTUP_WRAPPER_H Index: source/ariba/utility/system/SystemQueue.cpp =================================================================== --- source/ariba/utility/system/SystemQueue.cpp (revision 10789) +++ source/ariba/utility/system/SystemQueue.cpp (revision 12060) @@ -69,4 +69,16 @@ } +// maps to function call internally to the Event-system +void SystemQueue::scheduleCall( const boost::function0& function, uint32_t delay) +{ + // copy function object + boost::function0* function_ptr = new boost::function0(); + (*function_ptr) = function; + + // schedule special call-event + scheduleEvent( SystemEvent(&internal_function_caller, SystemEventType::DEFAULT, function_ptr), delay ); + +} + #ifdef UNDERLAY_OMNET void SystemQueue::handleMessage(cMessage* msg){ Index: source/ariba/utility/system/SystemQueue.h =================================================================== --- source/ariba/utility/system/SystemQueue.h (revision 10789) +++ source/ariba/utility/system/SystemQueue.h (revision 12060) @@ -60,4 +60,7 @@ #endif +#include + + using std::vector; using boost::posix_time::ptime; @@ -108,4 +111,13 @@ */ void scheduleEvent( const SystemEvent& event, uint32_t delay = 0 ); + + /** + * This method schedules a function call in the SystemQueue. + * (Like scheduleEvent, but to be used with boost::bind.) + * + * @param function: The function to be called [void function()] + * @param The delay in milli-seconds + */ + void scheduleCall( const boost::function0& function, uint32_t delay = 0 ); /** @@ -170,4 +182,6 @@ #endif + + private: @@ -235,5 +249,22 @@ volatile bool systemQueueRunning; #endif - + + +private: + /** + * This inner class handles the function-call events. + * @see SystemQueue::scheduleCall + */ + class FunctionCaller : public SystemEventListener + { + void handleSystemEvent(const SystemEvent& event) + { + boost::function0* function_ptr = event.getData< boost::function0 >(); + (*function_ptr)(); + delete function_ptr; + } + }; + + FunctionCaller internal_function_caller; }; // class SystemQueue Index: source/ariba/utility/transport/CMakeLists.txt =================================================================== --- source/ariba/utility/transport/CMakeLists.txt (revision 10789) +++ source/ariba/utility/transport/CMakeLists.txt (revision 12060) @@ -38,12 +38,8 @@ add_headers( - test_transport.hpp - transport_connection.hpp - transport.hpp - transport_listener.hpp transport_peer.cpp transport_peer.hpp - transport_protocol.hpp ) -add_subdir_sources(asio messages rfcomm tcpip) +add_subdir_sources(asio messages rfcomm StreamTransport) + Index: source/ariba/utility/transport/StreamTransport/CMakeLists.txt =================================================================== --- source/ariba/utility/transport/StreamTransport/CMakeLists.txt (revision 12060) +++ source/ariba/utility/transport/StreamTransport/CMakeLists.txt (revision 12060) @@ -0,0 +1,42 @@ +# [License] +# The Ariba-Underlay Copyright +# +# Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +# +# Institute of Telematics +# UniversitÀt Karlsruhe (TH) +# Zirkel 2, 76128 Karlsruhe +# Germany +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OF TELEMATICS OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +# +# The views and conclusions contained in the software and documentation +# are those of the authors and should not be interpreted as representing +# official policies, either expressed or implied, of the Institute of +# Telematics. +# [License] + +add_headers(StreamTransport.hpp) + +add_sources(StreamTransport.cpp) + Index: source/ariba/utility/transport/StreamTransport/StreamTransport.cpp =================================================================== --- source/ariba/utility/transport/StreamTransport/StreamTransport.cpp (revision 12060) +++ source/ariba/utility/transport/StreamTransport/StreamTransport.cpp (revision 12060) @@ -0,0 +1,814 @@ +#include "StreamTransport.hpp" + +// ariba +#include "ariba/utility/addressing2/tcpip_endpoint.hpp" + +// boost-adaption +#include "ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp" + +// boost +#include +#include +#include + +// interface discovery for link-local destinations (tcp-only) +#include + + +namespace ariba { +namespace transport { + +//use_logging_cpp(StreamTransport) + +#ifdef HAVE_LIBBLUETOOTH + using boost::asio::bluetooth::rfcomm; +#endif /* HAVE_LIBBLUETOOTH */ + +using namespace ariba::addressing2; +using ariba::utility::LinkID; + +using boost::asio::ip::tcp; +using boost::asio::ip::address_v6; +using boost::shared_ptr; + +typedef boost::mutex::scoped_lock unique_lock; + + +template +StreamTransport::StreamTransport( const typename T::endpoint& endp ) : + listener(NULL), + acceptor(u_io_service.get_asio_io_service(), endp) +{ +} + +template +StreamTransport::~StreamTransport(){} + +template +void StreamTransport::start() +{ + // open server socket + accept(); + + u_io_service.start(); +} + + +template +void StreamTransport::stop() +{ + acceptor.close(); + + u_io_service.stop(); +} + + +/* see header file for comments */ +template +void StreamTransport::send( + const typename T::endpoint& dest_addr, + reboost::message_t message, + uint8_t priority) +{ + ConnPtr conn; + bool need_to_connect = false; + + { + unique_lock lock(connections_lock); + + // try to find existing connection to this endpoint + typename ConnectionMap::iterator it = connections.find(dest_addr); + + // BRANCH: create new connection + if (it == connections.end()) + { + ConnPtr tmp_ptr( + new StreamConnection( + u_io_service.get_asio_io_service(), + this->shared_from_this() ) + ); + conn = tmp_ptr; + + // save partner endpoint + conn->partner_endpoint = 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; + + typename ConnectionMap::value_type item(dest_addr, conn); + connections.insert(item); + + } + // BRANCH: use existing connection + 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( + &StreamConnection::async_connect_handler, + conn, + boost::asio::placeholders::error)); + } +} + + +// TODO is this a private function? ... and still needed? +/* see header file for comments */ +template +void StreamTransport::send( + const EndpointPtr remote, + reboost::message_t message, + uint8_t priority) +{ + if ( remote->get_category() == endpoint_category::TCPIP ) + { + const addressing2::tcpip_endpoint* endp = + static_cast(remote.get()); + + send(endp->to_asio(), message, priority); + } + + // else + // TODO what now? +} + + +template +void StreamTransport::register_listener( transport_listener* listener ) +{ + this->listener = listener; +} + + +// XXX DEPRECATED +template +void StreamTransport::terminate( const EndpointPtr remote ) +{ + if ( remote->get_category() == endpoint_category::TCPIP ) + { + const addressing2::tcpip_endpoint* endp = + static_cast(remote.get()); + + + this->terminate(endp->to_asio()); + } +} + +template +void StreamTransport::terminate( const typename T::endpoint& remote ) +{ + ConnPtr conn; + + // find and forget connection + { + unique_lock lock(connections_lock); + + typename ConnectionMap::iterator it = connections.find(remote); + if (it == connections.end()) + { + return; + } + + conn = it->second; + + // prevent upper layers from using this link + conn->valid = false; + + connections.erase(it); + } + + // XXX aktuell +// cout << "/// MARIO: TCP/IP TERMINATE: " << conn->partner_endpoint << endl; + + // notify higher layers + if ( listener ) + { + listener->connection_terminated(conn); + } + + + // XXX debug aktuell +// cout << "/// MARIO Open connections:" << endl; +// for ( typename ConnectionMap::iterator it = connections.begin(); it != connections.end(); ++it ) +// { +// cout << " CONNECTION: " << it->second->local_endpoint << " <---> " << it->second->partner_endpoint << endl; +// cout << " used by: " << endl; +// +// int usecount = 0; +// ConnPtr xx; +// std::vector links = it->second->get_communication_links(); +// for ( std::vector::iterator it2 = links.begin(); it2 != links.end(); ++it2 ) +// { +// cout << " - " << *it2 << endl; +// usecount++; +// } +// if ( usecount == 0 ) +// { +// cout << " ---> NOBODY !!" << endl; +// } +// } +// cout << "/// -------" << endl; + + + // close connection + boost::system::error_code ec; + conn->sock.shutdown(T::socket::shutdown_both, ec); + conn->sock.close(ec); +} + + +/* private */ +template +void StreamTransport::accept() +{ + // create new connection object + ConnPtr conn( + new StreamConnection( + u_io_service.get_asio_io_service(), + this->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) + ); +} + +template +void StreamTransport::async_accept_handler(ConnPtr conn, const error_code& error) +{ + if ( ! error ) + { + // save partner endpoint + conn->partner_endpoint = conn->sock.remote_endpoint(); + + { + unique_lock lock(connections_lock); + + typename ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn); + connections.insert(item); + } + + // read + conn->listen(); + } + + // accept further connections + accept(); +} + + + +/*------------------ + | specializations | + ------------------*/ +/* TCP */ +template <> +void StreamTransport::send( + const addressing2::const_EndpointSetPtr 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 TcpIP_EndpointPtr address, endpoints->get_tcpip_endpoints() ) +// vector endpoint_vec = endpoints->get_tcpip_endpoints(); +// for ( vector::iterator it = endpoint_vec.begin(); it != endpoint_vec.end(); ++it ) + { + tcp::endpoint endp = address->to_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); + } +} + + +/* RFCOMM */ +#ifdef HAVE_LIBBLUETOOTH + +// TODO + +// template <> +// void StreamTransport::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); +// } +// } +// } + +#endif /* HAVE_LIBBLUETOOTH */ + + + +/***************** + ** inner class ** + *****************/ + +template +StreamTransport::StreamConnection::StreamConnection(boost::asio::io_service & io_service, StreamTransportPtr parent) : + sock(io_service), + valid(true), + parent(parent), + out_queues(8), //TODO How much priorities shall we have? + sending(false) +{ + header.length = 0; +} + +/*------------------------------------------- + | implement transport_connection interface | + -------------------------------------------*/ +template +bool StreamTransport::StreamConnection::send( + reboost::message_t message, + uint8_t priority) +{ + if ( ! valid ) + { + // XXX aktuell +// cout << "/// MARIO: USED INVALID LINK: " << this->partner_endpoint << endl; + + return false; + } + + // * enqueue data * + enqueue_for_sending(message, priority); + + return true; +} + + +template +EndpointPtr StreamTransport::StreamConnection::getLocalEndpoint() +{ + EndpointPtr ret(new addressing2::tcpip_endpoint(local_endpoint)); + + return ret; +} + + +template +EndpointPtr StreamTransport::StreamConnection::getRemoteEndpoint() +{ + EndpointPtr ret(new addressing2::tcpip_endpoint(partner_endpoint)); + + return ret; +} + +template +void StreamTransport::StreamConnection::register_communication_link(LinkID* link) +{ + if ( ! link ) + return; + + // add link + communication_links.push_back(link); +} + +template +void StreamTransport::StreamConnection::unregister_communication_link(LinkID* link) +{ + if ( ! link ) + return; + + + // remove link + { + std::vector::iterator it = communication_links.begin(); + + while ( it != communication_links.end() ) + { + if ( (*it) == link ) + { + it = communication_links.erase(it); + } + else + { + ++it; + } + } + } + + // this connection is no longer used by any link + if ( communication_links.empty() ) + { + //XXX +// cout << "communication_links.empty() " << getLocalEndpoint()->to_string() << " - " << getRemoteEndpoint()->to_string() << endl; + + // terminate connection + this->terminate(); // TODO aktuell + + /* + * TODO racecondition: + * When receiving a link request, the connection could closed + * before the request is handled. + * + * ---> Maybe wait a timeout before actually terminate the connection. + * (e.g. record last-used time: + * if last used > timeout and communication_links.empty() + * then terminate the connection) + */ + } +} + +template +std::vector StreamTransport::StreamConnection::get_communication_links() +{ + return communication_links; +} + + +template +void StreamTransport::StreamConnection::terminate() +{ + parent->terminate(partner_endpoint); +} + + +/*------------------------------ + | things we defined ourselves | + ------------------------------*/ +template +void StreamTransport::StreamConnection::async_connect_handler(const error_code& error) +{ + if (error) + { + parent->terminate(partner_endpoint); + + return; + } + + // save local endpoint + local_endpoint = sock.local_endpoint(); + + // Note: sending has to be true at this point + send_next_package(); + + listen(); +} + + +template +void StreamTransport::StreamConnection::listen() +{ + boost::asio::async_read( + this->sock, + boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)), + boost::bind( + &StreamTransport::StreamConnection::async_read_header_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + + +template +void StreamTransport::StreamConnection::async_read_header_handler(const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner_endpoint); + + return; + } + + // sanity checks + // NOTE: max size 8k (may be changed later..) + if ( header.length == 0 || header.length > 8192 ) + { + parent->terminate(partner_endpoint); + } + + + // 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( + &StreamTransport::StreamConnection::async_read_data_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + +template +void StreamTransport::StreamConnection::async_read_data_handler( + const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner_endpoint); + + return; + } + + if ( parent->listener ) + parent->listener->receive_message(this->shared_from_this(), buffy); + + buffy = shared_buffer_t(); + listen(); +} + +/* see header file for comments */ +template +void StreamTransport::StreamConnection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred) +{ + if ( error ) + { + // remove this connection + parent->terminate(partner_endpoint); + + return; + } + + send_next_package(); +} + + + +template +void StreamTransport::StreamConnection::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 */ +template +void StreamTransport::StreamConnection::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 = packet.size(); + + 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( + &StreamTransport::StreamConnection::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) + ); + } +} +/********************* + ** [ inner class ] ** + *********************/ + + + +// explicitly tell the compiler to create a »tcp« (and »rfcomm«) specialization +// --> (needed since hpp and cpp are separated) +template class StreamTransport; + +#ifdef HAVE_LIBBLUETOOTH + template class StreamTransport; +#endif /* HAVE_LIBBLUETOOTH */ + + + +///////////////////////////////////////////////////////////////////////////////////// + + +// XXX testing +///** +// * Conversion between ASIO Adresses and ARIBA adresses +// */ +///* TCP */ +//template <> +//inline typename tcp::endpoint convert_address( const address_v* address ) +//{ +// tcpip_endpoint endpoint = *address; +// +// return typename tcp::endpoint( +// endpoint.address().asio(), endpoint.port().value() +// ); +//} +// +//template <> +//inline EndpointPtr convert_address(const typename tcp::endpoint& endpoint) +//{ +// ip_address address; +// address.asio(endpoint.address()); +// tcp_port_address port; +// port.value(endpoint.port()); +// +//// new tcpip_endpoint(address, port); +// tcpip_endpoint xx; +// address_vf yy; +// address_v* zz = yy->clone(); +// address_v::shared_ptr endp(zz); // XXX +// +// return endp; +//} +// +///* RFCOMM */ +//#ifdef HAVE_LIBBLUETOOTH +// template <> +// inline typename rfcomm::endpoint convert_address( const address_v* address ) +// { +// rfcomm_endpoint endpoint = *address; +// +// return rfcomm::endpoint( +// endpoint.mac().bluetooth(), endpoint.channel().value() +// ); +// } +// +// template <> +// inline address_v::shared_ptr convert_address(const typename rfcomm::endpoint& endpoint) +// { +// mac_address mac; +// mac.bluetooth(endpoint.address()); +// rfcomm_channel_address channel; +// channel.value(endpoint.channel()); +// +// address_v::shared_ptr endp((ariba::addressing::address_v*) new rfcomm_endpoint(mac, channel)); +// +// return endp; +// } +//#endif /* HAVE_LIBBLUETOOTH */ + + + +///////////////////////////////////////////////////////////////////////////////////// + + +/* + * Get Ethernet scope ids (for link-local) + */ +vector 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; +} + + +}} // namespace ariba::transport Index: source/ariba/utility/transport/StreamTransport/StreamTransport.hpp =================================================================== --- source/ariba/utility/transport/StreamTransport/StreamTransport.hpp (revision 12060) +++ source/ariba/utility/transport/StreamTransport/StreamTransport.hpp (revision 12060) @@ -0,0 +1,227 @@ +#ifndef STREAM_TRANSPORT_HPP_ +#define STREAM_TRANSPORT_HPP_ + +// ariba +#include "ariba/utility/transport/asio/unique_io_service.h" +#include "ariba/utility/transport/messages/buffers.hpp" +//#include "ariba/utility/logging/Logging.h" +#include "ariba/utility/addressing2/endpoint.hpp" + +// ariba (transport) interfaces +#include "ariba/utility/transport/interfaces/transport_connection.hpp" +#include "ariba/utility/transport/interfaces/transport_protocol.hpp" +#include "ariba/utility/transport/interfaces/transport_listener.hpp" + +// system +#include + +// boost +#include +#include +#include + + +namespace ariba { +namespace transport { + +using namespace std; +using boost::shared_ptr; +using ariba::transport::detail::unique_io_service; +using boost::system::error_code; +using reboost::shared_buffer_t; +using reboost::message_t; + +template +class StreamTransport : + public transport_protocol, + public boost::enable_shared_from_this > +{ + typedef StreamTransport self; + typedef shared_ptr StreamTransportPtr; +//use_logging_h(StreamTransport) + + + + +private: + +/***************** + ** inner class ** + *****************/ + class StreamConnection : + public transport_connection, + public boost::enable_shared_from_this + { + public: + typedef reboost::message_t Packet; + typedef std::queue OutQueue; + + struct header_t + { + uint16_t length; + } __attribute__((packed)); + + StreamConnection(boost::asio::io_service& io_service, StreamTransportPtr parent); + + virtual ~StreamConnection() {} + + /// Inherited from transport_connection + // Thread: ARIBA + virtual bool send(reboost::message_t message, uint8_t priority = 0); + // Thread: ARIBA + virtual ariba::addressing2::EndpointPtr getLocalEndpoint(); + // Thread: ARIBA + virtual ariba::addressing2::EndpointPtr getRemoteEndpoint(); + // Thread: ARIBA + virtual void register_communication_link(ariba::utility::LinkID* link); + // Thread: ARIBA + virtual void unregister_communication_link(ariba::utility::LinkID* link); + // Thread: ARIBA + virtual std::vector get_communication_links(); + + // Thread: BOTH + 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: + typename T::socket sock; + bool valid; + StreamTransportPtr parent; + + typename T::endpoint partner_endpoint; + typename T::endpoint local_endpoint; +// address_v* remote; +// address_v* 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; + + private: + std::vector communication_links; + }; + typedef boost::shared_ptr ConnPtr; + typedef std::map ConnectionMap; +/********************* + ** [ inner class ] ** + *********************/ + + + + +public: + StreamTransport( const typename T::endpoint& endp ); + virtual ~StreamTransport(); + 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 typename T::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 addressing2::EndpointPtr remote, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * calls send for each destination endpoint in `endpoints` + */ + virtual void send( + const addressing2::const_EndpointSetPtr endpoints, + reboost::message_t message, + uint8_t priority = 0 ); + + // XXX DEPRECATED + virtual void terminate( addressing2::EndpointPtr remote ); + + virtual void terminate( const typename T::endpoint& remote ); + virtual void register_listener( transport_listener* listener ); + +private: + void accept(); + void async_accept_handler(ConnPtr conn, const error_code& error); + +private: + transport_listener* listener; + unique_io_service u_io_service; + typename T::acceptor acceptor; + + ConnectionMap connections; + boost::mutex connections_lock; +}; + + +// aliases +//typedef StreamTransport tcpip; +//#ifdef HAVE_LIBBLUETOOTH +// typedef StreamTransport rfcomm_transport; +//#endif /* HAVE_LIBBLUETOOTH */ + + +// XXX testing TODO natÌrlich brauchen wir das noch... oO +///** +// * Conversion between ASIO Adresses and ARIBA adresses +// */ +//template +//typename T::endpoint convert_address(const address_v* endpoint); +// +//template +//address_v::shared_ptr convert_address(const typename T::endpoint& endpoint); + + + +/** + * 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 ?? + * + */ +vector get_interface_scope_ids(); + + +}} // namespace ariba::transport + +#endif /* STREAM_TRANSPORT_HPP_ */ Index: source/ariba/utility/transport/interfaces/CMakeLists.txt =================================================================== --- source/ariba/utility/transport/interfaces/CMakeLists.txt (revision 12060) +++ source/ariba/utility/transport/interfaces/CMakeLists.txt (revision 12060) @@ -0,0 +1,45 @@ +# [License] +# The Ariba-Underlay Copyright +# +# Copyright (c) 2008-2012, Institute of Telematics, UniversitÀt Karlsruhe (TH) +# +# Institute of Telematics +# UniversitÀt Karlsruhe (TH) +# Zirkel 2, 76128 Karlsruhe +# Germany +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR +# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OF TELEMATICS OR +# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, +# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR +# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +# +# The views and conclusions contained in the software and documentation +# are those of the authors and should not be interpreted as representing +# official policies, either expressed or implied, of the Institute of +# Telematics. +# [License] + +add_headers( + transport_connection.hpp + transport_listener.hpp + transport_protocol.hpp +) + +#add_sources() Index: source/ariba/utility/transport/interfaces/transport_connection.hpp =================================================================== --- source/ariba/utility/transport/interfaces/transport_connection.hpp (revision 12060) +++ source/ariba/utility/transport/interfaces/transport_connection.hpp (revision 12060) @@ -0,0 +1,60 @@ +#ifndef TRANSPORT_CONNECTION_HPP +#define TRANSPORT_CONNECTION_HPP + +// ariba +#include "ariba/utility/transport/messages/message.hpp" +#include "ariba/utility/addressing2/endpoint.hpp" + +#include +#include + +namespace ariba { + +// forward declaration +namespace utility { + class LinkID; +} + + +namespace transport { + +class transport_connection +{ +public: + typedef boost::shared_ptr sptr; + + /// Allow deleting implementing classes by pointer + virtual ~transport_connection() {} + + virtual bool send(reboost::message_t message, uint8_t priority = 1) = 0; + + virtual ariba::addressing2::EndpointPtr getLocalEndpoint() = 0; + virtual ariba::addressing2::EndpointPtr getRemoteEndpoint() = 0; + + + /** + * Tells this transport_connection, that it's used by the given + * communication link. + */ + virtual void register_communication_link(ariba::utility::LinkID* link) = 0; + + /** + * Tells this transport_connection, that it's no longer used by the given + * communication link. + * + * A transport_connection may terminate itself, when it's no longer used by + * any communication link. + */ + virtual void unregister_communication_link(ariba::utility::LinkID* link) = 0; + + /** + * @return A vector of all registered communication links. + */ + virtual std::vector get_communication_links() = 0; + + virtual void terminate() = 0; +}; + +} /* namespace transport */ +} /* namespace ariba */ +#endif /* TRANSPORT_CONNECTION_HPP */ Index: source/ariba/utility/transport/interfaces/transport_listener.hpp =================================================================== --- source/ariba/utility/transport/interfaces/transport_listener.hpp (revision 12060) +++ source/ariba/utility/transport/interfaces/transport_listener.hpp (revision 12060) @@ -0,0 +1,39 @@ +// transport_listener.hpp, created on 01.07.2009 by Sebastian Mies + +#ifndef TRANSPORT_LISTENER_HPP +#define TRANSPORT_LISTENER_HPP + +// ariba +#include "ariba/utility/transport/messages/buffers.hpp" + +// ariba interfaces +#include "transport_connection.hpp" + +// namespace ariba::transport +namespace ariba { +namespace transport { + +/** + * TODO: Doc + * + * @author Sebastian Mies + */ +class transport_listener { +public: + /// Allow deleting implementing classes by pointer + virtual ~transport_listener() {} + + /// called when a message is received + virtual void receive_message( + transport_connection::sptr connection, + reboost::shared_buffer_t msg + ) = 0; + + /// called when a connection is terminated (e.g. TCP close) + virtual void connection_terminated(transport_connection::sptr connection) = 0; +}; + +}} // namespace ariba::transport + + +#endif /* TRANSPORT_LISTENER_HPP */ Index: source/ariba/utility/transport/interfaces/transport_protocol.hpp =================================================================== --- source/ariba/utility/transport/interfaces/transport_protocol.hpp (revision 12060) +++ source/ariba/utility/transport/interfaces/transport_protocol.hpp (revision 12060) @@ -0,0 +1,56 @@ +#ifndef TRANSPORT_PROTOCOL_HPP_ +#define TRANSPORT_PROTOCOL_HPP_ + +// ariba +#include "ariba/utility/addressing2/endpoint_set.hpp" +#include "ariba/utility/transport/messages/message.hpp" + +// ariba interfaces +#include "transport_listener.hpp" + +// namespace ariba::transport +namespace ariba { +namespace transport { + +struct system_priority { + enum SEND_PRIORITY_INTERNAL { + HIGHEST = 0, + HIGH = 1, + ON_IDLE = 7 + }; + + static const SEND_PRIORITY_INTERNAL OVERLAY = HIGHEST; +}; + + + +/** + * TODO: Doc + * + * @author Sebastian Mies , Mario Hock + */ +class transport_protocol { +public: + /// Allow deleting implementing classes by pointer + virtual ~transport_protocol() {} + + virtual void start() = 0; + virtual void stop() = 0; + + /** + * @returns whether this connection is assumed to be valid + */ + virtual void send( + const addressing2::const_EndpointSetPtr endpoints, + reboost::message_t message, + uint8_t priority = system_priority::HIGH) = 0; + + /// @deprecated: Use terminate() from transport_connection instead +// virtual void terminate( const address_v* remote ) = 0; + + virtual void register_listener( transport_listener* listener ) = 0; +}; + +}} // namespace ariba::transport + +#endif /* TRANSPORT_PROTOCOL_HPP_ */ Index: source/ariba/utility/transport/messages/message.cpp =================================================================== --- source/ariba/utility/transport/messages/message.cpp (revision 10789) +++ source/ariba/utility/transport/messages/message.cpp (revision 12060) @@ -24,5 +24,5 @@ os << "message({size=" << m.size() << ",buffers=" << (int) m.length() << ",hash=" << m.hash() << "},"; - m.foreach(ts); + m.msg_foreach(ts); os << ")"; return os; Index: source/ariba/utility/transport/messages/message.hpp =================================================================== --- source/ariba/utility/transport/messages/message.hpp (revision 10789) +++ source/ariba/utility/transport/messages/message.hpp (revision 12060) @@ -17,8 +17,11 @@ /// message size type -typedef signed char mlength_t; +//typedef signed char mlength_t; // <--- don't do this!! +//typedef size_t mlength_t; +typedef int mlength_t; // signed int seems necessary /// maximum number of buffers per message (default is 8) const mlength_t message_max_buffers = (1L << 3); +//const mlength_t message_max_buffers = (1L << 4); //! A Copy-on-Write Message with Shared Buffers. @@ -70,6 +73,8 @@ /// Copy message inline message_t(const message_t& msg) : - imsg(msg.imsg) { - imsg->owner = NULL; + imsg(msg.imsg) + { + if ( imsg ) + imsg->owner = NULL; } @@ -142,4 +147,7 @@ /// Returns the number of buffers inside this message. inline mlength_t length() const { + if ( ! imsg ) + return 0; + return (imsg->length); } @@ -167,5 +175,5 @@ /// Iterates over a partial set of buffers. template - inline void foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const { + inline void msg_foreach(const T& work, size_t index_ = 0, size_t size_ = 0) const { T op = work; if (size_ == 0) size_ = size() - index_; @@ -192,5 +200,5 @@ inline void read(boctet_t* mem, size_t idx = 0, size_t size_ = 0) const { struct read_buffer rb = { mem }; - foreach(rb, idx, size_); + msg_foreach(rb, idx, size_); } @@ -198,5 +206,5 @@ inline void write(const boctet_t* mem, size_t idx = 0, size_t size_ = 0) { struct write_buffer wb = { mem }; - foreach(wb, idx, size_); + msg_foreach(wb, idx, size_); } @@ -227,5 +235,5 @@ message_t m; struct sub_message sm = { &m }; - foreach(sm, index, size); + msg_foreach(sm, index, size); return m; } Index: source/ariba/utility/transport/messages/shared_buffer.hpp =================================================================== --- source/ariba/utility/transport/messages/shared_buffer.hpp (revision 10789) +++ source/ariba/utility/transport/messages/shared_buffer.hpp (revision 12060) @@ -9,4 +9,5 @@ #include +#include #include @@ -18,5 +19,19 @@ #include "buffer.hpp" +#include + namespace reboost { + +class illegal_sub_buffer: public std::runtime_error +{ +public: + /** Takes a character string describing the error. */ + explicit illegal_sub_buffer(const std::string& __arg) : + std::runtime_error(__arg) + { + } + + virtual ~illegal_sub_buffer() throw() {} +}; /** @@ -104,8 +119,13 @@ parent(new deleteable_buffer(buffer, size)) { + } + +// /// XXX debug... copy! +// /// create shared buffer from buffer +// inline shared_buffer_t(const char* buffer, bsize_t size) : +// buffer_t(), parent(new deleteable_buffer(size)) { // memcpy(parent->mutable_data(), buffer, parent->size()); -// data(parent->mutable_data()); -// this->size(parent->size()); - } +// data(parent->mutable_data()); this->size(parent->size()); +// } /// clone data from a normal buffer @@ -129,5 +149,21 @@ /// return sub-buffer. - inline self operator()(bsize_t index, bsize_t size = 0) const { + inline self operator()(bsize_t index, bsize_t size = 0) const + { + // special cases + if ( index + size > size_ ) + { + // empty sub-buffer + if ( index == size_ ) + { + self n; + return n; + } + + // ERROR: index out of bounds + throw illegal_sub_buffer("Index or size out of bounds in shared_buffer"); + } + + // regular case self n(*this); n.data_ += index; Index: source/ariba/utility/transport/rfcomm/CMakeLists.txt =================================================================== --- source/ariba/utility/transport/rfcomm/CMakeLists.txt (revision 10789) +++ source/ariba/utility/transport/rfcomm/CMakeLists.txt (revision 12060) @@ -40,6 +40,6 @@ bluetooth_endpoint.hpp bluetooth_rfcomm.hpp - rfcomm_transport.hpp ) -add_sources(rfcomm_transport.cpp) +#add_sources() + Index: source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp =================================================================== --- source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp (revision 10789) +++ (revision ) @@ -1,464 +1,0 @@ -#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 10789) +++ (revision ) @@ -1,170 +1,0 @@ -#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/test_transport.hpp =================================================================== --- source/ariba/utility/transport/test_transport.hpp (revision 10789) +++ (revision ) @@ -1,82 +1,0 @@ -// test_transport.hpp, created on 01.07.2009 by Sebastian Mies - -#ifndef TEST_TRANSPORT_HPP_ -#define TEST_TRANSPORT_HPP_ - -#include "transport_peer.hpp" -#include "transport_listener.hpp" - -#include "rfcomm/rfcomm.hpp" - -#include -#include -#include -#include - -namespace ariba { -namespace transport { -namespace detail { - -using namespace std; -using namespace ariba::transport; -using namespace ariba::addressing; - -class listener : public transport_listener { -public: - virtual void receive_message( - transport_protocol* transport, - const address_vf local, const address_vf remote, - const uint8_t* data, size_t size - ) { - cout << "transport_listener: " << endl; - cout << "received message data='" << data << "' local=" << local->to_string() << " remote=" << remote->to_string() << endl; - } -}; - -void test_transport_process( endpoint_set& local, endpoint_set& remote ) { - cout << "started " << local.to_string() << endl; - transport_peer* peer = new transport_peer( local ); - peer->register_listener( new listener() ); - peer->start(); - peer->send( remote, (uint8_t*)"Hello!", 7 ); - getchar(); -} - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -void tcp_test() { - - endpoint_set local = string("tcp{5001};ip{127.0.0.1 | 2001:638:204:6:216:d3ff:fece:1070}"); - endpoint_set remote = string("tcp{5002};ip{127.0.0.1 | 2001:638:204:6:216:d3ff:fece:1070}"); - - pid_t pID = fork(); - if (pID < 0) { - cerr << "Failed to fork" << endl; - } else - if (pID == 0) { - test_transport_process(local,remote); - } else { - getchar(); - test_transport_process(remote,local); - } - getchar(); -} - -void bluetooth_test( string& endp_str ) { - cout << endp_str << endl; - rfcomm* rfc = new rfcomm( 3 ); - rfc->register_listener( new listener() ); - rfc->start(); - if (endp_str.size()!=0) { - rfcomm_endpoint endp = endp_str; - rfc->send( endp, (uint8_t*)"Hello!", 7 ); - } - getchar(); -} - -}}} // namespace ariba::transport::detail - -#endif /* TEST_TRANSPORT_HPP_ */ Index: source/ariba/utility/transport/transport.hpp =================================================================== --- source/ariba/utility/transport/transport.hpp (revision 10789) +++ (revision ) @@ -1,15 +1,0 @@ -#ifndef TRANSPORT_HPP_ -#define TRANSPORT_HPP_ - -// abstract classes -#include "transport_protocol.hpp" -#include "transport_listener.hpp" - -// transport protocol implementations -#include "tcpip/tcpip.hpp" -#include "rfcomm/rfcomm_transport.hpp" - -// common transport peer using all known protocols -#include "transport_peer.hpp" - -#endif /* TRANSPORT_HPP_ */ Index: source/ariba/utility/transport/transport_connection.hpp =================================================================== --- source/ariba/utility/transport/transport_connection.hpp (revision 10789) +++ (revision ) @@ -1,32 +1,0 @@ - -#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 10789) +++ (revision ) @@ -1,38 +1,0 @@ -// transport_listener.hpp, created on 01.07.2009 by Sebastian Mies - -#ifndef TRANSPORT_LISTENER_HPP_ -#define TRANSPORT_LISTENER_HPP_ - -#include "ariba/utility/addressing/addressing.hpp" -#include "ariba/utility/transport/messages/buffers.hpp" -#include "ariba/utility/transport/transport_connection.hpp" - -// namespace ariba::transport -namespace ariba { -namespace transport { - -using namespace ariba::addressing; - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -class transport_listener { -public: - /// Allow deleting implementing classes by pointer - virtual ~transport_listener() {} - - /// called when a message is received - virtual void receive_message( - transport_connection::sptr connection, - reboost::message_t msg - ) { - std::cout << "transport_listener: not implemented" << std::endl; - } -}; - -}} // namespace ariba::transport - - -#endif /* TRANSPORT_LISTENER_HPP_ */ Index: source/ariba/utility/transport/transport_peer.cpp =================================================================== --- source/ariba/utility/transport/transport_peer.cpp (revision 10789) +++ source/ariba/utility/transport/transport_peer.cpp (revision 12060) @@ -1,15 +1,11 @@ - -#include "ariba/config.h" #include "transport_peer.hpp" -#include "transport.hpp" -#include + +// ariba +#include "StreamTransport/StreamTransport.hpp" +#include "ariba/utility/addressing2/tcpip_endpoint.hpp" + +// boost #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 @@ -17,5 +13,5 @@ namespace transport { -using namespace ariba::addressing; +using namespace addressing2; using boost::asio::ip::tcp; @@ -26,74 +22,136 @@ use_logging_cpp(transport_peer); -transport_peer::transport_peer( endpoint_set& local_set ) : local(local_set) { - - // setup tcp transports - foreach(tcp_port_address port, local.tcp) { +transport_peer::transport_peer() : + local(new addressing2::endpoint_set()) +{ +} + +EndpointSetPtr transport_peer::add_listenOn_endpoints(EndpointSetPtr endpoints) +{ + // TCP Endpoints + BOOST_FOREACH( shared_ptr endp, endpoints->get_tcpip_endpoints() ) + { + // automatic port detection + bool port_detection = false; + uint16_t try_port = 41322; - if (local.ip.size() > 0) { - foreach(ip_address ip_addr, local.ip) { - - tcp::endpoint endp(ip_addr.asio(), port.asio()); - create_service(endp); - } - } else { - tcp::endpoint endp_v6(tcp::v6(), port.asio()); - tcp::endpoint endp_v4(tcp::v4(), port.asio()); - - create_service(endp_v6); - create_service(endp_v4); + tcp::endpoint asio_endp = endp->to_asio(); + if ( asio_endp.port() == 0 ) + { + port_detection = true; } - } - + + // create new server socket + do + { + try + { + // automatic port detection + if ( port_detection ) + { + asio_endp.port(try_port); + endp = tcpip_endpoint::create_TcpIP_Endpoint(asio_endp); + } + + TransportProtocolPtr tmp_ptr(new StreamTransport(endp->to_asio())); + transport_streams.push_back(tmp_ptr); + logging_info("Listening on IP/TCP " << endp->to_string()); + + local->add_endpoint(endp); + port_detection = false; + } + + catch (boost::system::system_error& e) + { + // address in use + if (e.code() == boost::asio::error::address_in_use) + { + // BRANCH: automatic port detection + if ( port_detection ) + { + // give up ? + if ( try_port > 41422 ) + { + logging_warn("[WARN] Unable to find free port. Giving up. :-( Last try was: " + << endp->to_string() << ". Endpoint will be ignored!"); + + port_detection = false; + } + else + { + // try next port + try_port++; + } + } + // BRANCH: explicit given port --> error + else + { + logging_warn("[WARN] Address already in use: " + << endp->to_string() << ". Endpoint will be ignored!"); + } + } + + // Rethrow + else + { + throw; + } + } + } while ( port_detection ); + } + + // TODO Bluetooth Endpoints #ifdef HAVE_LIBBLUETOOTH - foreach(rfcomm_channel_address channel, local.rfcomm) { - if (local.bluetooth.size() > 0) { - foreach(mac_address mac, local.bluetooth) { - rfcomm::endpoint endp(mac.bluetooth(), channel.value()); - create_service(endp); - } - } else { - rfcomm::endpoint endp(channel.value()); - create_service(endp); - } - } +// foreach(rfcomm_channel_address channel, local.rfcomm) { +// if (local.bluetooth.size() > 0) { +// foreach(mac_address mac, local.bluetooth) { +// rfcomm::endpoint endp(mac.bluetooth(), channel.value()); +// create_service(endp); +// } +// } else { +// rfcomm::endpoint endp(channel.value()); +// create_service(endp); +// } +// } #endif -} - -void transport_peer::create_service(tcp::endpoint endp) { - try { - TcpIpPtr tmp_ptr(new tcpip(endp)); - tcps.push_back(tmp_ptr); - logging_info("Listening on IP/TCP " << endp); - - } catch (boost::system::system_error& e) { - if (e.code() == boost::asio::error::address_in_use) { - logging_warn("[WARN] Address already in use: " - << endp << ". Endpoint will be ignored!"); - } else { - // Rethrow - throw; - } - } -} + + return local; +} + +//void transport_peer::create_service(tcp::endpoint endp) { +// try { +// TransportProtocolPtr tmp_ptr(new StreamTransport(endp)); +// tcps.push_back(tmp_ptr); +// logging_info("Listening on IP/TCP " << endp); +// +// } catch (boost::system::system_error& e) { +// if (e.code() == boost::asio::error::address_in_use) { +// logging_warn("[WARN] Address already in use: " +// << endp << ". Endpoint will be ignored!"); +// } else { +// // Rethrow +// throw; +// } +// } +//} #ifdef HAVE_LIBBLUETOOTH -void transport_peer::create_service(rfcomm::endpoint endp) { - try { - rfcomm_transport::sptr tmp_ptr(new rfcomm_transport(endp)); - rfcomms.push_back(tmp_ptr); - logging_info("Listening on bluetooth/RFCOMM " << endp); - - } catch (boost::system::system_error& e) { - if (e.code() == boost::asio::error::address_in_use) { - logging_warn("[WARN] Address already in use: " - << endp << ". Endpoint will be ignored!"); - } else { - // Rethrow - throw; - } - } -} +//void transport_peer::create_service(rfcomm::endpoint endp) { +// try { +// TransportProtocolPtr tmp_ptr(new StreamTransport(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 @@ -101,71 +159,55 @@ } -void transport_peer::start() { - foreach(TcpIpPtr tcp, tcps) { - tcp->start(); - } - -#ifdef HAVE_LIBBLUETOOTH - foreach(rfcomm_transport::sptr x, rfcomms) { - x->start(); - } -#endif -} - -void transport_peer::stop() { - foreach(TcpIpPtr tcp, tcps) { - tcp->stop(); - } - -#ifdef HAVE_LIBBLUETOOTH - foreach(rfcomm_transport::sptr x, rfcomms) { - x->stop(); - } -#endif +void transport_peer::start() +{ + BOOST_FOREACH(TransportProtocolPtr stream, transport_streams) + { + stream->start(); + } +} + +void transport_peer::stop() +{ + BOOST_FOREACH(TransportProtocolPtr stream, transport_streams) + { + stream->stop(); + } } void transport_peer::send( - const endpoint_set& endpoints, + const const_EndpointSetPtr endpoints, reboost::message_t message, uint8_t priority) { - foreach(TcpIpPtr tcp, tcps) { - tcp->send(endpoints, message, priority); - } - -#ifdef HAVE_LIBBLUETOOTH - foreach(rfcomm_transport::sptr x, rfcomms) { - x->send(endpoints, message, priority); - } -#endif -} - -void transport_peer::terminate( const address_v* remote ) { - if (remote->instanceof())// TODO direkt auf der richtigen verbindung - { - foreach(TcpIpPtr tcp, tcps) { - tcp->terminate(remote); - } - } -#ifdef HAVE_LIBBLUETOOTH - if (remote->instanceof()) { - foreach(rfcomm_transport::sptr x, rfcomms) { - x->terminate(remote); - } - } -#endif -} - -void transport_peer::register_listener( transport_listener* listener ) { - foreach(TcpIpPtr tcp, tcps) { - tcp->register_listener(listener); - } - -#ifdef HAVE_LIBBLUETOOTH - foreach(rfcomm_transport::sptr x, rfcomms) { - x->register_listener(listener); - } -#endif + BOOST_FOREACH(TransportProtocolPtr stream, transport_streams) + { + stream->send(endpoints, message, priority); + } +} + +// XXX DEPRECATED +//void transport_peer::terminate( const address_v* remote ) { +// if (remote->instanceof())// TODO direkt auf der richtigen verbindung +// { +// foreach(TransportProtocolPtr tcp, tcps) { +// tcp->terminate(remote); +// } +// } +//#ifdef HAVE_LIBBLUETOOTH +// if (remote->instanceof()) { +// foreach(TransportProtocolPtr x, rfcomms) { +// x->terminate(remote); +// } +// } +//#endif +//} + +void transport_peer::register_listener( transport_listener* listener ) +{ + BOOST_FOREACH(TransportProtocolPtr stream, transport_streams) + { + stream->register_listener(listener); + } } Index: source/ariba/utility/transport/transport_peer.hpp =================================================================== --- source/ariba/utility/transport/transport_peer.hpp (revision 10789) +++ source/ariba/utility/transport/transport_peer.hpp (revision 12060) @@ -2,10 +2,17 @@ #define TRANSPORT_PEER_HPP_ +// ariba #include "ariba/config.h" #include "ariba/utility/logging/Logging.h" -#include "transport_protocol.hpp" -#include "ariba/utility/addressing/endpoint_set.hpp" +#include "ariba/utility/addressing2/endpoint_set.hpp" + +// ariba interfaces +#include "interfaces/transport_protocol.hpp" + +// boost #include -#include "rfcomm/bluetooth_rfcomm.hpp" + +// boost-adaption +//#include "rfcomm/bluetooth_rfcomm.hpp" @@ -14,23 +21,26 @@ namespace transport { -using namespace ariba::addressing; - -class tcpip; - -#ifdef HAVE_LIBBLUETOOTH -class rfcomm_transport; -#endif - /** - * TODO: Doc + * This class allocates implementations of various transport + * protocols and can send messages to an entire set of endpoints * - * @author Sebastian Mies + * @author Sebastian Mies , Mario Hock */ -/// this transport peer allocates implementations of various transport -/// protocols and can send messages to an entire set of endpoints -class transport_peer : public transport_protocol { +class transport_peer : + public transport_protocol +{ use_logging_h(transport_peer); + typedef boost::shared_ptr TransportProtocolPtr; + public: - transport_peer( endpoint_set& local_set ); + transport_peer(); + + /** + * Adds endpoints on which ariba should listen ("server"-sockets) + * + * @return An endpoint_set holding all active endpoints ariba is listening on. + */ + addressing2::EndpointSetPtr add_listenOn_endpoints(addressing2::EndpointSetPtr endpoints); + virtual ~transport_peer(); virtual void start(); @@ -38,24 +48,14 @@ virtual void send( - const endpoint_set& endpoints, + const addressing2::const_EndpointSetPtr endpoints, reboost::message_t message, - uint8_t priority = 0); - - /// @deprecated: Use terminate() from transport_connection instead - virtual void terminate( const address_v* remote ); - + uint8_t priority = system_priority::OVERLAY); + virtual void register_listener( transport_listener* listener ); + private: - void create_service(tcp::endpoint endp); -#ifdef HAVE_LIBBLUETOOTH - void create_service(boost::asio::bluetooth::rfcomm::endpoint endp); -#endif - - endpoint_set& local; - std::vector< boost::shared_ptr > tcps; -#ifdef HAVE_LIBBLUETOOTH - std::vector< boost::shared_ptr > rfcomms; -#endif + addressing2::EndpointSetPtr local; + std::vector transport_streams; }; Index: source/ariba/utility/transport/transport_protocol.hpp =================================================================== --- source/ariba/utility/transport/transport_protocol.hpp (revision 10789) +++ (revision ) @@ -1,40 +1,0 @@ -#ifndef TRANSPORT_PROTOCOL_HPP_ -#define TRANSPORT_PROTOCOL_HPP_ - -#include "ariba/utility/addressing/addressing.hpp" -#include "ariba/utility/transport/transport_listener.hpp" -#include "ariba/utility/transport/messages/message.hpp" - -// namespace ariba::transport -namespace ariba { -namespace transport { - -using namespace ariba::addressing; - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -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 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; -}; - -}} // namespace ariba::transport - -#endif /* TRANSPORT_PROTOCOL_HPP_ */ Index: source/ariba/utility/types/Identifier.h =================================================================== --- source/ariba/utility/types/Identifier.h (revision 10789) +++ source/ariba/utility/types/Identifier.h (revision 12060) @@ -50,4 +50,7 @@ #include "ariba/utility/serialization.h" +// XXX EXPERIMENTAL +#include "ariba/utility/transport/messages/shared_buffer.hpp" + /**< maximum length of the key */ #define MAX_KEYLENGTH 192 @@ -69,4 +72,24 @@ use_logging_h( Identifier ); public: + + // XXX EXPERIMENTAL + reboost::shared_buffer_t serialize() const + { + Data data = data_serialize(this, DEFAULT_V); + reboost::shared_buffer_t buf(data.getBuffer(), data.getLength() / 8); + + return buf; + } + + reboost::shared_buffer_t deserialize(reboost::shared_buffer_t buff) + { + Data dat(const_cast(buff.data()), buff.size() * 8); + + size_t len = this->SERIALIZATION_METHOD_NAME(DESERIALIZE, dat) / 8; + + // return remaining sub-buffer + return buff(len); + } + //------------------------------------------------------------------------- Index: source/ariba/utility/types/OverlayParameterSet.h =================================================================== --- source/ariba/utility/types/OverlayParameterSet.h (revision 10789) +++ source/ariba/utility/types/OverlayParameterSet.h (revision 12060) @@ -53,5 +53,5 @@ typedef enum _OverlayStructure { - OverlayStructureOneHop = 0, + OverlayStructureOneHop = 0, // DEPRECATED, DO NOT USE OverlayStructureChord = 1, } OverlayStructure; Index: source/ariba/utility/types/ServiceID.cpp =================================================================== --- source/ariba/utility/types/ServiceID.cpp (revision 10789) +++ source/ariba/utility/types/ServiceID.cpp (revision 12060) @@ -76,5 +76,9 @@ } -string ServiceID::toString() const { +string ServiceID::toString() const +{ + if ( *this == ServiceID::UNSPECIFIED ) + return "UNSPEC"; + return ariba::utility::Helper::ultos( id ); }