Changeset 10653 for source/ariba/utility/transport

Timestamp:

Jul 25, 2012, 11:41:36 AM (12 years ago)

Author:

Michael Tänzer

Message:

Merge the ASIO branch back into trunk

Location:

source/ariba/utility/transport

Files:

• . (modified) (1 prop)
• asio/asio_io_service.cpp (deleted)
• asio/asio_io_service.h (deleted)
• asio/bluetooth_endpoint.hpp (deleted)
• asio/rfcomm.hpp (deleted)
• asio/unique_io_service.cpp (added)
• asio/unique_io_service.h (added)
• messages (added)
• messages/CHANGELOG (added)
• messages/LICENSE (added)
• messages/buffer.cpp (added)
• messages/buffer.hpp (added)
• messages/buffers.hpp (added)
• messages/message.cpp (added)
• messages/message.hpp (added)
• messages/shared_buffer.cpp (added)
• messages/shared_buffer.hpp (added)
• rfcomm/bluetooth_endpoint.hpp (added)
• rfcomm/bluetooth_rfcomm.hpp (added)
• rfcomm/rfcomm.cpp (deleted)
• rfcomm/rfcomm.hpp (deleted)
• rfcomm/rfcomm_transport.cpp (added)
• rfcomm/rfcomm_transport.hpp (added)
• tcpip/protlib (deleted)
• tcpip/tcpip.cpp (modified) (1 diff)
• tcpip/tcpip.hpp (modified) (1 diff)
• transport.hpp (modified) (1 diff)
• transport_connection.hpp (added)
• transport_listener.hpp (modified) (3 diffs)
• transport_peer.cpp (modified) (2 diffs)
• transport_peer.hpp (modified) (3 diffs)
• transport_protocol.hpp (modified) (2 diffs)

source/ariba/utility/transport/tcpip/tcpip.cpp

@@ -1 +1 @@
 #include "tcpip.hpp"
 
-#define _NO_LOGGING
-
-// std includes
-#include <unistd.h>
-#include <iostream>
-#include <string>
-#include <sstream>
-#include <boost/foreach.hpp>
-
-// protlib includes
-#include "protlib/network_message.h"
-#include "protlib/tp_over_tcp.h"
-#include "protlib/tperror.h"
-#include "protlib/logfile.h"
-#include "protlib/queuemanager.h"
-#include "protlib/threadsafe_db.h"
-#include "protlib/setuid.h"
-
-// protlib namespaces
-using namespace protlib;
-using namespace protlib::log;
-
-logfile commonlog;
-protlib::log::logfile& protlib::log::DefaultLog(commonlog);
+#include <boost/array.hpp>
+
+// interface discovery for link-local destinations
+#include <ifaddrs.h>
 
 namespace ariba {
 namespace transport {
 
+use_logging_cpp(tcpip)
+
 using namespace ariba::addressing;
 
-
-tcpip_endpoint convert( const appladdress* addr ) {
-        const char* ip_str = addr->get_ip_str();
-        tcpip_endpoint endpoint( std::string(ip_str), addr->get_port() );
-        return endpoint;
-}
-
-appladdress convert( const tcpip_endpoint& endpoint ) {
-        tcpip_endpoint* e = const_cast<tcpip_endpoint*>(&endpoint);
-        appladdress
-                peer(e->address().to_string().c_str(), "tcp", e->port().asio() );
-//      cout << endpoint.to_string() << " to " << peer.get_ip_str() << ":" << peer.get_port() << endl;
-        return peer;
-}
-
-tcpip::tcpip( uint16_t port ) : 
-        done ( false ),
-        running ( false ),
-        port( port ),
-        tpreceivethread ( NULL ),
-        tpthread ( NULL ),
-        listener ( NULL ) {
-}
-
-tcpip::~tcpip() {
-        if (running) stop();
-}
-
-bool get_message_length( NetMsg& m, uint32& clen_bytes ) {
-        clen_bytes = m.decode32();
-        m.set_pos_r(-4);
-        return true;
-}
-
-void tcpip::start() {
-        done = false;
-        running = false;
-
-        // initalize netdb and setuid
-        protlib::tsdb::init();
-        protlib::setuid::init();
-
-        // set tcp parameters
-        port_t port = this->port; // port
-        TPoverTCPParam tppar(4, get_message_length, port);
-
-        // create receiver thread
-        FastQueue* tpchecker_fq = new FastQueue("TCPTransport", true);
-        QueueManager::instance()->register_queue(tpchecker_fq,
-                        message::qaddr_signaling);
-
-        // start thread
-        pthread_create( &tpreceivethread, NULL, tcpip::receiverThread, this );
-        tpthread = new ThreadStarter<TPoverTCP, TPoverTCPParam> ( 1, tppar );
-        tpthread->start_processing();
-}
-
-void tcpip::stop() {
-        // stop receiver thread
-        done = true;
-
-        // stop TPoverTCP
-        tpthread->stop_processing();
-        tpthread->abort_processing(true);
-        tpthread->wait_until_stopped();
-
-        // unregister TPoverTCP
-        delete QueueManager::instance()->get_queue( message::qaddr_signaling );
-        QueueManager::instance()->unregister_queue( message::qaddr_signaling );
-
-        // destroy QueueManager
-        QueueManager::clear();
-
-        // de-initalize netdb and setuid
-        protlib::setuid::end();
-        protlib::tsdb::end();
-
-        // wait for thread to finish and delete
-        pthread_join(tpreceivethread, NULL);
-}
-
-void tcpip::send( const address_v* remote, const uint8_t* data, size_t size ) {
-
-        // prepare netmsg with length and and port
-        NetMsg* datamsg = new NetMsg(size + 6);
-        datamsg->encode32( size + 2,  true );
-        datamsg->encode16( this->port,true );
-
-        for (size_t i=0; i<size; i++)
-                datamsg->encode8( data[i],true );
-
-        // send message
-        tcpip_endpoint endpoint = *remote;
-        appladdress peer = convert(endpoint);
-
-        // add to output queue
-        tpthread->get_thread_object()->send( datamsg, peer, false );
-}
-
-void tcpip::send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) {
-        // send a message to each combination of ip-address and port
-        BOOST_FOREACH( const ip_address ip, endpoints.ip ) {
-                BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) {
-                        tcpip_endpoint endpoint(ip,port);
-                        address_vf vf = endpoint;
-                        send(vf,data,size);
-                }
-        }
-}
-
-void tcpip::terminate( const address_v* remote) {
-        tcpip_endpoint endpoint = *remote;
-        appladdress peer = convert(endpoint);
-        peer.convert_to_ipv6();
-        tpthread->get_thread_object()->terminate( peer );
-}
-
-void tcpip::register_listener( transport_listener* listener ) {
-        this->listener = listener;
-}
-
-void* tcpip::receiverThread( void* ptp ) {
-        // get reference to transport object
-        tcpip& tp = *((tcpip*)ptp);
-
-        // get queue
-        FastQueue* fq =
-                QueueManager::instance()->get_queue(message::qaddr_signaling);
-
-        // main processing loop
-        tp.running = true;
-        while (!tp.done) {
-
-                // wait for new message to approach
-                message* msg = fq->dequeue_timedwait(300);
-
-                // message has arrived? no-> continue
-                if (!msg) continue;
-
-                // handle transport message
-                TPMsg* tpmsg = dynamic_cast<TPMsg*> (msg);
-                if (!tpmsg) {
-                        delete msg;
-                        continue;
-                }
-
-                // get address & message
-                const appladdress* remote_peer = static_cast<const appladdress*>( tpmsg->get_peeraddress() );
-                const appladdress* local_peer  = static_cast<const appladdress*>( tpmsg->get_ownaddress() );
-                NetMsg* datamsg = tpmsg->get_message();
-
-                // not a data message? -> continue!
-                if (!datamsg) {
-                        delete tpmsg;
-                        continue;
-                }
-
-                // get length and remote endpoint port
-                datamsg->set_pos(0);
-                uint32_t message_size = datamsg->decode32(true)-2;
-                //uint16_t remote_port = datamsg->decode16(true);
-
-                // inform listener
-                if (tp.listener != NULL) {
-                        tcpip_endpoint remote = convert(remote_peer);
-                        tcpip_endpoint local  = convert(local_peer);
-                        tp.listener->receive_message(
-                                        &tp, local, remote, datamsg->get_buffer()+6, message_size );
-                }
-
-                tpmsg->set_message(NULL);
-                delete datamsg;
-                delete tpmsg;
-        }
-        // clean queue & stop
-        fq->cleanup();
-        tp.running = false;
-        return NULL;
+typedef boost::mutex::scoped_lock unique_lock;
+
+tcpip::tcpip( const tcp::endpoint& endp )  :
+        listener(NULL),
+        acceptor(u_io_service.get_asio_io_service(), endp)
+{
+}
+
+tcpip::~tcpip(){}
+
+void tcpip::start()
+{
+    // open server socket
+    accept();
+    
+    u_io_service.start();
+}
+
+
+void tcpip::stop()
+{
+    acceptor.close();
+    
+    u_io_service.stop();
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const tcp::endpoint& dest_addr,
+        reboost::message_t message,
+        uint8_t priority)
+{
+    ConnPtr conn;
+    bool need_to_connect = false;
+    
+    {
+        unique_lock lock(connections_lock);
+        
+        ConnectionMap::iterator it = connections.find(dest_addr);
+        if (it == connections.end())
+        {
+            ConnPtr tmp_ptr(
+                    new tcpip_connection(
+                            u_io_service.get_asio_io_service(),
+                            shared_from_this() )
+                    );
+            conn = tmp_ptr;
+            
+            conn->partner = dest_addr;
+            conn->remote = convert_address(dest_addr);
+            
+            // Note: starting the send is the obligation of the connect_handler
+            // (avoids trying to send while not connected yet)
+            conn->sending =  true;
+            need_to_connect = true;
+            
+            ConnectionMap::value_type item(dest_addr, conn);
+            connections.insert(item);
+            
+        } else {
+            conn = it->second;
+        }
+    }
+    
+    
+    // * the actual send *
+    conn->enqueue_for_sending(message, priority);
+    
+    // if new connection connect to the other party
+    if ( need_to_connect )
+    {
+        conn->sock.async_connect(
+                dest_addr,
+                boost::bind(
+                        &tcpip_connection::async_connect_handler,
+                        conn,
+                        boost::asio::placeholders::error));
+    }
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const address_v* remote,
+        reboost::message_t message,
+        uint8_t priority)
+{
+    send(convert_address(remote), message, priority);
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const endpoint_set& endpoints,
+        reboost::message_t message,
+        uint8_t priority )
+{
+    // network interfaces scope_ids, for link-local connections (lazy initialization)
+    vector<uint64_t> scope_ids;
+    
+    // send a message to each combination of address-address and port
+    BOOST_FOREACH( const ip_address address, endpoints.ip ) {
+        BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) {
+            tcp::endpoint endp(address.asio(), port.asio());
+            
+            // special treatment for link local addresses
+            //   ---> send over all (suitable) interfaces
+            if ( endp.address().is_v6() )
+            {
+                boost::asio::ip::address_v6 v6_addr = endp.address().to_v6();
+                
+                if ( v6_addr.is_link_local() )
+                {
+                    // initialize scope_ids
+                    if ( scope_ids.size() == 0 )
+                        scope_ids = get_interface_scope_ids();
+                    
+                    BOOST_FOREACH ( uint64_t id, scope_ids )
+                    {                        
+                        v6_addr.scope_id(id);
+                        endp.address(v6_addr);
+    
+                        logging_debug("------> SEND TO (link-local): " << endp);
+                        // * send *
+                        send(endp, message, priority);
+                    }
+                }
+                
+                continue;
+            }
+            
+            // * send *
+            send(endp, message, priority);
+        }
+    }
+}
+
+
+void tcpip::register_listener( transport_listener* listener )
+{
+    this->listener = listener;
+}
+
+
+void tcpip::terminate( const address_v* remote )
+{
+    terminate(convert_address(remote));
+}
+
+void tcpip::terminate( const tcp::endpoint& remote )
+{
+    ConnPtr conn;
+    
+    // find and forget connection
+    {
+        unique_lock lock(connections_lock);
+        
+        ConnectionMap::iterator it = connections.find(remote);
+        if (it == connections.end())
+        {
+            return;
+        }
+        
+        conn = it->second;
+        
+        connections.erase(it);
+    }
+
+    // close connection
+    boost::system::error_code ec;
+    conn->sock.shutdown(tcp::socket::shutdown_both, ec);
+    conn->sock.close(ec);
+}
+
+
+/* private */
+void tcpip::accept()
+{
+    // create new connection object
+    ConnPtr conn(
+            new tcpip_connection(
+                    u_io_service.get_asio_io_service(),
+                    shared_from_this()
+            )
+    );
+    
+    // wait for incoming connection
+    acceptor.async_accept(
+            conn->sock,
+            boost::bind(&self::async_accept_handler,
+                    this->shared_from_this(),
+                    conn,
+                    boost::asio::placeholders::error)
+    );
+}
+
+void tcpip::async_accept_handler(ConnPtr conn, const error_code& error)
+{
+    if ( ! error )
+    {
+        conn->partner = conn->sock.remote_endpoint();
+        conn->remote = convert_address(conn->partner);
+        conn->local = convert_address(conn->sock.local_endpoint());
+        
+        {
+            unique_lock lock(connections_lock);
+            
+            ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn);
+            connections.insert(item);
+        }
+        
+        // read
+        conn->listen();
+    }
+    
+    // accept further connections
+    accept();
+}
+
+inline tcp::endpoint tcpip::convert_address( const address_v* address )
+{
+    tcpip_endpoint endpoint = *address;
+    
+    return tcp::endpoint(
+        endpoint.address().asio(), endpoint.port().value()
+    );
+}
+
+
+inline tcpip_endpoint tcpip::convert_address(const tcp::endpoint& endpoint)
+{
+    ip_address address;
+    address.asio(endpoint.address());
+    tcp_port_address port;
+    port.value(endpoint.port());
+    return tcpip_endpoint(address, port);
+}
+
+
+vector<uint64_t> tcpip::get_interface_scope_ids()
+{
+    vector<uint64_t> ret;
+    
+    struct ifaddrs* ifaceBuffer = NULL;
+    void*           tmpAddrPtr  = NULL;
+    
+    int ok = getifaddrs( &ifaceBuffer );
+    if( ok != 0 ) return ret;
+
+    for( struct ifaddrs* i=ifaceBuffer; i != NULL; i=i->ifa_next ) {
+
+        // ignore devices that are disabled or have no ip
+        if(i == NULL) continue;
+        struct sockaddr* addr = i->ifa_addr;
+        if (addr==NULL) continue;
+
+        // only use ethX and wlanX devices
+        string device = string(i->ifa_name);
+        if ( (device.find("eth") == string::npos) &&
+              (device.find("wlan")  == string::npos) /* &&
+              (device.find("lo")  == string::npos) XXX */ )
+        {
+            continue;
+        }
+
+        // only use interfaces with ipv6 link-local addresses 
+        if (addr->sa_family == AF_INET6)
+        {
+            // convert address
+            // TODO should be possible without detour over strings
+            char straddr[INET6_ADDRSTRLEN];
+            tmpAddrPtr= &((struct sockaddr_in6*)addr)->sin6_addr;
+            inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
+
+            address_v6 v6addr = address_v6::from_string(straddr);
+            if ( v6addr.is_link_local() )
+            {
+                // * append the scope_id to the return vector *
+                ret.push_back(if_nametoindex(i->ifa_name));
+            }
+
+        }
+    }
+
+    freeifaddrs(ifaceBuffer);
+    
+    return ret;
+}
+
+
+/*****************
+ ** inner class **
+ *****************/
+
+tcpip::tcpip_connection::tcpip_connection(boost::asio::io_service & io_service, TcpIpPtr parent)  :
+        sock(io_service),
+        valid(true),
+        parent(parent),
+        out_queues(8), //TODO How much priorities shall we have?
+        sending(false)
+{
+        header.length = 0;
+        header.prot = 0;
+}
+
+/*-------------------------------------------
+ | implement transport_connection interface |
+ -------------------------------------------*/
+void tcpip::tcpip_connection::send(
+        reboost::message_t message,
+        uint8_t priority)
+{
+    enqueue_for_sending(message, priority);
+}
+
+
+address_vf tcpip::tcpip_connection::getLocalEndpoint()
+{
+    return local;
+}
+
+
+address_vf tcpip::tcpip_connection::getRemoteEndpoint()
+{
+    return remote;
+}
+
+
+void tcpip::tcpip_connection::terminate()
+{
+    parent->terminate(partner);
+}
+
+
+/*------------------------------
+ | things we defined ourselves |
+ ------------------------------*/
+void tcpip::tcpip_connection::async_connect_handler(const error_code& error)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+    
+    // save address in ariba format
+    local = parent->convert_address(sock.local_endpoint());
+    
+    // Note: sending has to be true at this point
+    send_next_package();
+    
+    listen();
+}
+
+
+void tcpip::tcpip_connection::listen()
+{
+    boost::asio::async_read(
+            this->sock,
+            boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)),
+            boost::bind(
+                    &tcpip::tcpip_connection::async_read_header_handler,
+                    this->shared_from_this(),
+                    boost::asio::placeholders::error,
+                    boost::asio::placeholders::bytes_transferred
+            )
+    );
+}
+
+
+void tcpip::tcpip_connection::async_read_header_handler(const error_code& error, size_t bytes_transferred)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+
+    // convert byte order
+    header.length = ntohl(header.length);
+    header.length -= 2;  // XXX protlib
+    
+    assert(header.length > 0);
+    
+    // new buffer for the new packet
+    buffy = shared_buffer_t(header.length);
+
+    // * read data *
+    boost::asio::async_read(
+            this->sock,
+            boost::asio::buffer(buffy.mutable_data(), buffy.size()),
+            boost::bind(
+                    &tcpip::tcpip_connection::async_read_data_handler,
+                    this->shared_from_this(),
+                    boost::asio::placeholders::error,
+                    boost::asio::placeholders::bytes_transferred
+            )
+    );
+}
+
+void tcpip::tcpip_connection::async_read_data_handler(
+        const error_code& error, size_t bytes_transferred)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+    
+    message_t msg;
+    msg.push_back(buffy);
+    buffy = shared_buffer_t();
+
+    if ( parent->listener )
+        parent->listener->receive_message(shared_from_this(), msg);
+    
+    listen();
+}
+
+/* see header file for comments */
+void tcpip::tcpip_connection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred)
+{
+    if ( error )
+    {        
+        // remove this connection
+        parent->terminate(partner); 
+
+        return;
+    }
+    
+    send_next_package();
+}
+
+
+
+void tcpip::tcpip_connection::enqueue_for_sending(Packet packet, uint8_t priority)
+{
+    bool restart_sending = false;
+    
+    // enqueue packet  [locked]
+    {
+        unique_lock(out_queues_lock);
+        
+        assert( priority < out_queues.size() );
+        out_queues[priority].push(packet);
+        
+        if ( ! sending )
+        {
+            restart_sending = true;
+            sending = true;
+        }
+    }
+    
+    // if sending was stopped, we have to restart it here
+    if ( restart_sending )
+    {
+        send_next_package();
+    }
+}
+
+/* see header file for comments */
+void tcpip::tcpip_connection::send_next_package()
+{
+    Packet packet;
+    bool found = false;
+
+    // find packet with highest priority  [locked]
+    {
+        unique_lock(out_queues_lock);
+        
+        for ( vector<OutQueue>::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<boost::asio::const_buffer> send_sequence(packet.length());
+        for ( int i=0; i < packet.length(); i++ )
+        {
+            shared_buffer_t b = packet.at(i);
+            send_sequence.push_back(boost::asio::buffer(b.data(), b.size()));
+        }
+        
+        // * async write *
+        boost::asio::async_write(
+                this->sock,
+                send_sequence,
+                boost::bind(
+                        &tcpip::tcpip_connection::async_write_handler,
+                        this->shared_from_this(),
+                        packet,  // makes sure our shared pointer lives long enough ;-)
+                        boost::asio::placeholders::error,
+                        boost::asio::placeholders::bytes_transferred)
+        );
+    }
 }
 

source/ariba/utility/transport/tcpip/tcpip.hpp

@@ -3 +3 @@
 
 #include "ariba/utility/transport/transport.hpp"
-#include <pthread.h>
-
-// forward declaration
-namespace protlib {
-template<class X, class Y>
-class ThreadStarter;
-class TPoverTCP;
-class TPoverTCPParam;
-}
+#include "ariba/utility/transport/asio/unique_io_service.h"
+#include "ariba/utility/transport/transport_connection.hpp"
+#include "ariba/utility/addressing/tcpip_endpoint.hpp"
+#include <boost/asio.hpp>
+#include <boost/shared_ptr.hpp>
+#include <boost/enable_shared_from_this.hpp>
+#include <queue>
+#include "ariba/utility/transport/messages/buffers.hpp"
+#include "ariba/utility/logging/Logging.h"
 
 namespace ariba {
 namespace transport {
 
-using namespace protlib;
+using namespace std;
+using ariba::transport::detail::unique_io_service;
+using ariba::addressing::tcpip_endpoint;
+using boost::asio::ip::tcp;
+using boost::asio::ip::address_v6;
+using boost::system::error_code;
+using reboost::shared_buffer_t;
+using reboost::message_t;
 
-/**
- * TODO: Doc
- *
- * @author Sebastian Mies <mies@tm.uka.de>
- */
-class tcpip : public transport_protocol {
+class tcpip;
+typedef boost::shared_ptr<tcpip> TcpIpPtr;
+
+class tcpip :
+    public transport_protocol,
+    public boost::enable_shared_from_this<tcpip>
+{
+    typedef tcpip self;
+use_logging_h(tcpip)
+
+private:
+    class tcpip_connection :
+        public transport_connection,
+        public boost::enable_shared_from_this<tcpip_connection>
+    {
+    public:
+        typedef reboost::message_t Packet;
+        typedef std::queue<Packet> OutQueue;
+        
+        struct header_t
+        {
+            uint32_t length;
+            uint16_t prot;  // XXX protlib
+        } __attribute__((packed));
+            
+        tcpip_connection(boost::asio::io_service& io_service, TcpIpPtr parent);
+        
+        /// Inherited from transport_connection
+        virtual void send(reboost::message_t message, uint8_t priority = 0);
+        virtual address_vf getLocalEndpoint();
+        virtual address_vf getRemoteEndpoint();
+        virtual void terminate();
+        
+        void listen();
+        
+        void async_connect_handler(const error_code& error);
+        
+        void async_read_header_handler(const error_code& error, size_t bytes_transferred);
+        void async_read_data_handler(const error_code& error, size_t bytes_transferred);
+        
+        /*
+         * is called from asio when write operation "returns",
+         * calls private function `send_next_package()`
+         */
+        void async_write_handler(
+                reboost::shared_buffer_t packet,
+                const error_code& error,
+                size_t bytes_transferred);
+
+        
+        void enqueue_for_sending(Packet packet, uint8_t priority);
+        
+    private:
+        /*
+         * is called from `send` or `async_write_handler` to begin/keep sending
+         * sends the next message with the highest priority in this connection
+         */
+        void send_next_package();
+
+
+    public:
+        tcp::socket sock;
+        bool valid;
+        TcpIpPtr parent;
+        
+        tcp::endpoint partner;
+        tcpip_endpoint remote;
+        tcpip_endpoint local;
+        
+        vector<OutQueue> 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<tcpip_connection> ConnPtr;
+    typedef std::map<tcp::endpoint, ConnPtr> ConnectionMap;
+    
 public:
-        tcpip( uint16_t port );
+        tcpip( const tcp::endpoint& endp );
         virtual ~tcpip();
         virtual void start();
         virtual void stop();
-        virtual void send( const address_v* remote, const uint8_t* data, size_t size );
-        virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size );
+        
+        /**
+     * enqueues message for sending
+     * create new connection if necessary
+     * starts sending mechanism (if not already running)
+     */
+    void send(
+            const tcp::endpoint&,
+            reboost::message_t message,
+            uint8_t priority = 0 );
+        
+        /**
+         * Converts address_v to tcp::endpoint and calls the real send() function
+         */
+        virtual void send(
+                const address_v* remote,
+                reboost::message_t message,
+                uint8_t priority = 0 );
+        
+        /**
+         * calls send for each destination endpoint in `endpoint_set& endpoints` 
+         */
+        virtual void send(
+                const endpoint_set& endpoints,
+                reboost::message_t message,
+                uint8_t priority = 0 );
+        
         virtual void terminate( const address_v* remote );
+        virtual void terminate( const tcp::endpoint& remote );
         virtual void register_listener( transport_listener* listener );
 
+        
+    /**
+     *  returns a vector of (interesting) network interfaces
+     *  
+     *  [NOTE: The current implementation returns the scope_ids of
+     *  all ethX and wlanX network interfaces, to be used for
+     *  connections to link-local ipv6 addresses.]
+     *  
+     *  TODO move to ariba/communication/networkinfo/AddressDiscovery ??
+     *  
+     */
+    static vector<uint64_t> get_interface_scope_ids();
+
 private:
-        volatile bool done, running;
-        uint16_t port;
-        pthread_t tpreceivethread;
-        ThreadStarter<TPoverTCP, TPoverTCPParam>* tpthread;
-        static void* receiverThread( void* ptp );
+        void accept();
+        void async_accept_handler(ConnPtr conn, const error_code& error);
+        tcp::endpoint convert_address(const address_v* endpoint);
+        tcpip_endpoint convert_address(const tcp::endpoint& endpoint);
+        
+private:
         transport_listener* listener;
+        unique_io_service u_io_service;
+        tcp::acceptor acceptor;
+        
+        ConnectionMap connections;
+        boost::mutex connections_lock;
 };
 

source/ariba/utility/transport/transport.hpp

@@ -8 +8 @@
 // transport protocol implementations
 #include "tcpip/tcpip.hpp"
-#include "rfcomm/rfcomm.hpp"
+#include "rfcomm/rfcomm_transport.hpp"
 
 // common transport peer using all known protocols

source/ariba/utility/transport/transport_listener.hpp

@@ -5 +5 @@
 
 #include "ariba/utility/addressing/addressing.hpp"
+#include "ariba/utility/transport/messages/buffers.hpp"
+#include "ariba/utility/transport/transport_connection.hpp"
 
 // namespace ariba::transport
@@ -11 +13 @@
 
 using namespace ariba::addressing;
-
-class transport_protocol;
 
 /**
@@ -21 +21 @@
 class transport_listener {
 public:
+    /// Allow deleting implementing classes by pointer
+    virtual ~transport_listener() {}
+    
         /// called when a message is received
         virtual void receive_message(
-                transport_protocol* transport,
-                const address_vf local, const address_vf remote,
-                const uint8_t* data, size_t size
+        transport_connection::sptr connection,
+                reboost::message_t msg
         ) {
                 std::cout << "transport_listener: not implemented" << std::endl;

source/ariba/utility/transport/transport_peer.cpp

@@ -3 +3 @@
 #include "transport_peer.hpp"
 #include "transport.hpp"
+#include "ariba/utility/logging/Logging.h"
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/error.hpp>
+#include <boost/foreach.hpp>
+
+#ifdef ECLIPSE_PARSER
+    #define foreach(a, b) for(a : b)
+#else
+    #define foreach(a, b) BOOST_FOREACH(a, b)
+#endif
 
 // namespace ariba::transport
@@ -9 +19 @@
 
 using namespace ariba::addressing;
+using boost::asio::ip::tcp;
+
+#ifdef HAVE_LIBBLUETOOTH
+using boost::asio::bluetooth::rfcomm;
+#endif
+
+use_logging_cpp(transport_peer);
 
 transport_peer::transport_peer( endpoint_set& local_set ) : local(local_set) {
-        // setup tcp transports
-        tcp = NULL;
-        //cout << "#tcpip_transports = " << local.tcp.size() << endl;
-        if (local.tcp.size()==1) {
-                tcp = new tcpip(local.tcp.begin()->value());
-                //cout << "Started tcpip_transport on port "  << local.tcp.begin()->value() << endl;
-        }
-
+    
+    // setup tcp transports
+    foreach(tcp_port_address port, local.tcp) {
+        
+        if (local.ip.size() > 0) {
+            foreach(ip_address ip_addr, local.ip) {
+                
+                tcp::endpoint endp(ip_addr.asio(), port.asio());
+                create_service(endp);
+            }
+        } else {
+            tcp::endpoint endp_v6(tcp::v6(), port.asio());
+            tcp::endpoint endp_v4(tcp::v4(), port.asio());
+            
+            create_service(endp_v6);
+            create_service(endp_v4);
+        }
+        
+    }
+    
         #ifdef HAVE_LIBBLUETOOTH
-        // setup rfcomm transports
-        rfc = NULL;
-        //cout << "#rfcomm_transports = " << local.rfcomm.size() << endl;
-        if ( local.rfcomm.size() == 1 ) {
-                rfc = new rfcomm( local.rfcomm.begin()->value() );
-                //cout << "Started rfcomm_transport on port "  << local.rfcomm.begin()->value() << endl;
-        }
+    foreach(rfcomm_channel_address channel, local.rfcomm) {
+        if (local.bluetooth.size() > 0) {
+                foreach(mac_address mac, local.bluetooth) {
+                        rfcomm::endpoint endp(mac.bluetooth(), channel.value());
+                        create_service(endp);
+                }
+        } else {
+                rfcomm::endpoint endp(channel.value());
+                create_service(endp);
+        }
+    }
         #endif
 }
 
+void transport_peer::create_service(tcp::endpoint endp) {
+    try {
+        TcpIpPtr tmp_ptr(new tcpip(endp));
+        tcps.push_back(tmp_ptr);
+        logging_info("Listening on IP/TCP " << endp);
+        
+    } catch (boost::system::system_error& e) {
+        if (e.code() == boost::asio::error::address_in_use) {
+            logging_warn("[WARN] Address already in use: "
+                    << endp << ". Endpoint will be ignored!");
+        } else {
+            // Rethrow
+            throw;
+        }
+    }
+}
+
+#ifdef HAVE_LIBBLUETOOTH
+void transport_peer::create_service(rfcomm::endpoint endp) {
+    try {
+        rfcomm_transport::sptr tmp_ptr(new rfcomm_transport(endp));
+        rfcomms.push_back(tmp_ptr);
+        logging_info("Listening on bluetooth/RFCOMM " << endp);
+        
+    } catch (boost::system::system_error& e) {
+        if (e.code() == boost::asio::error::address_in_use) {
+            logging_warn("[WARN] Address already in use: "
+                    << endp << ". Endpoint will be ignored!");
+        } else {
+            // Rethrow
+            throw;
+        }
+    }
+}
+#endif
+
 transport_peer::~transport_peer() {
-        if (tcp !=NULL ) delete tcp;
-#ifdef HAVE_LIBBLUETOOTH
-        if (rfc !=NULL ) delete rfc;
-#endif
 }
 
 void transport_peer::start() {
-        if (tcp!=NULL) tcp->start();
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->start();
+    }
+    
 #ifdef HAVE_LIBBLUETOOTH
-        if (rfc!=NULL) rfc->start();
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+        x->start();
+    }
 #endif
 }
 
 void transport_peer::stop() {
-        if (tcp!=NULL) tcp->stop();
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->stop();
+    }
+    
 #ifdef HAVE_LIBBLUETOOTH
-        if (rfc!=NULL) rfc->stop();
+        foreach(rfcomm_transport::sptr x, rfcomms) {
+                x->stop();
+        }
 #endif
 }
 
-void transport_peer::send( const address_v* remote, const uint8_t* data, size_t size ) {
-        if (remote->instanceof<tcpip_endpoint>() && tcp!=NULL) {
-                tcp->send(remote,data,size);
-        } else
+
+void transport_peer::send(
+        const endpoint_set& endpoints,
+        reboost::message_t message,
+        uint8_t priority)
+{
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->send(endpoints, message, priority);
+    }
+    
 #ifdef HAVE_LIBBLUETOOTH
-        if (remote->instanceof<rfcomm_endpoint>() && rfc!=NULL) {
-                rfc->send(remote,data,size);
-        } else
-#endif
-                cerr << "Could not send message to " << remote->to_string() << endl;
-}
-
-void transport_peer::send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) {
-        if (tcp!=NULL) tcp->send(endpoints,data,size);
-#ifdef HAVE_LIBBLUETOOTH
-        if (rfc!=NULL) rfc->send(endpoints,data,size);
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+                x->send(endpoints, message, priority);
+        }
 #endif
 }
 
 void transport_peer::terminate( const address_v* remote ) {
-        if (remote->instanceof<tcpip_endpoint>() && tcp!=NULL)
-                tcp->terminate(remote);
+        if (remote->instanceof<tcpip_endpoint>())// TODO direkt auf der richtigen verbindung
+        {
+            foreach(TcpIpPtr tcp, tcps) {
+                tcp->terminate(remote);
+            }
+        }
 #ifdef HAVE_LIBBLUETOOTH
-        if (remote->instanceof<rfcomm_endpoint>() && rfc!=NULL)
-                rfc->terminate(remote);
+        if (remote->instanceof<rfcomm_endpoint>()) {
+                foreach(rfcomm_transport::sptr x, rfcomms) {
+                        x->terminate(remote);
+                }
+        }
 #endif
 }
 
 void transport_peer::register_listener( transport_listener* listener ) {
-        if (tcp!=NULL) tcp->register_listener(listener);
+    foreach(TcpIpPtr tcp, tcps) {
+        tcp->register_listener(listener);
+    }
+    
 #ifdef HAVE_LIBBLUETOOTH
-        if (rfc!=NULL) rfc->register_listener(listener);
+    foreach(rfcomm_transport::sptr x, rfcomms) {
+        x->register_listener(listener);
+    }
 #endif
 }

source/ariba/utility/transport/transport_peer.hpp

@@ -5 +5 @@
 #include "transport_protocol.hpp"
 #include "ariba/utility/addressing/endpoint_set.hpp"
+#include <boost/shared_ptr.hpp>
+#include "rfcomm/bluetooth_rfcomm.hpp"
+
 
 // namespace ariba::transport
@@ -13 +16 @@
 
 class tcpip;
+
 #ifdef HAVE_LIBBLUETOOTH
-class rfcomm;
+class rfcomm_transport;
 #endif
 
@@ -30 +34 @@
         virtual void start();
         virtual void stop();
-        virtual void send( const address_v* remote, const uint8_t* data, size_t size );
-        virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size );
+        
+        virtual void send(
+                const endpoint_set& endpoints,
+                reboost::message_t message,
+                uint8_t priority = 0);
+        
+        /// @deprecated: Use terminate() from transport_connection instead
         virtual void terminate( const address_v* remote );
+        
         virtual void register_listener( transport_listener* listener );
 
 private:
+        void create_service(tcp::endpoint endp);
+#ifdef HAVE_LIBBLUETOOTH
+        void create_service(boost::asio::bluetooth::rfcomm::endpoint endp);
+#endif
+        
         endpoint_set&  local;
-        tcpip* tcp;
+        std::vector< boost::shared_ptr<tcpip> > tcps;
 #ifdef HAVE_LIBBLUETOOTH
-        rfcomm* rfc;
+        std::vector< boost::shared_ptr<rfcomm_transport> > rfcomms;
 #endif
 };

source/ariba/utility/transport/transport_protocol.hpp

@@ -3 +3 @@
 
 #include "ariba/utility/addressing/addressing.hpp"
-#include "transport_listener.hpp"
+#include "ariba/utility/transport/transport_listener.hpp"
+#include "ariba/utility/transport/messages/message.hpp"
 
 // namespace ariba::transport
@@ -18 +19 @@
 class transport_protocol {
 public:
+    /// Allow deleting implementing classes by pointer
+    virtual ~transport_protocol() {}
+    
         virtual void start() = 0;
         virtual void stop() = 0;
-        virtual void send( const address_v* remote, const uint8_t* data, size_t size ) = 0;
-        virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) = 0;
+        
+        virtual void send(
+                const endpoint_set& endpoints,
+                reboost::message_t message,
+                uint8_t priority = 0) = 0;
+        
+        /// @deprecated: Use terminate() from transport_connection instead
         virtual void terminate( const address_v* remote ) = 0;
+        
         virtual void register_listener( transport_listener* listener ) = 0;
 };