Index: source/ariba/utility/transport/CMakeLists.txt =================================================================== --- source/ariba/utility/transport/CMakeLists.txt (revision 10700) +++ 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 10700) +++ 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 10700) +++ 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 10700) +++ 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 10700) +++ 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 10700) +++ (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 10700) +++ (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 10700) +++ (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 10700) +++ (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 10700) +++ (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 10700) +++ (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 10700) +++ 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 10700) +++ 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 10700) +++ (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_ */