Index: source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp =================================================================== --- source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp (revision 10653) +++ source/ariba/utility/transport/rfcomm/bluetooth_endpoint.hpp (revision 10653) @@ -0,0 +1,195 @@ +#include "ariba/config.h" + +#ifdef HAVE_LIBBLUETOOTH + +#ifndef BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ +#define BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ + +#include +#include +#include + +#include + +namespace boost { +namespace asio { +namespace bluetooth { + +/** + * Describes an endpoint for a RFCOMM Bluetooth socket. + * + * @author Martin Florian + */ +template +class bluetooth_endpoint { +private: + static const bdaddr_t addr_any; + +public: + /// The protocol type associated with the endpoint. + typedef BluetoothProtocol protocol_type; + + /// The type of the endpoint structure. This type is dependent on the + /// underlying implementation of the socket layer. + typedef boost::asio::detail::socket_addr_type data_type; // <-- Do I need this? + //typedef sockaddr_rc data_type; + + /// Default constructor. + bluetooth_endpoint() : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = 0; + } + + bluetooth_endpoint(const BluetoothProtocol& protocol, + uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = channel; + } + + /// Construct an endpoint using a port number, specified in the host's byte + /// order. The IP address will be the any address (i.e. INADDR_ANY or + /// in6addr_any). This constructor would typically be used for accepting new + /// connections. + bluetooth_endpoint(uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_bdaddr = addr_any; + data_.rc_channel = channel; + } + + /// Construct an endpoint using a port number and an BT address. + /// The address is in human readable form as a string. + bluetooth_endpoint(const char *addr, uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_channel = channel; + str2ba(addr, &data_.rc_bdaddr); + } + + /// Construct an endpoint using a port number and an BT address. + /// The address is given in the bluetooth-internal format. + bluetooth_endpoint(const bdaddr_t& addr, uint8_t channel) : + data_() { + data_.rc_family = AF_BLUETOOTH; + data_.rc_channel = channel; + data_.rc_bdaddr = addr; + } + + /// Copy constructor. + bluetooth_endpoint(const bluetooth_endpoint& other) : + data_(other.data_) { + } + + /// Assign from another endpoint. + bluetooth_endpoint& operator=(const bluetooth_endpoint& other) { + data_ = other.data_; + return *this; + } + + /// The protocol associated with the endpoint. + protocol_type protocol() const { + return protocol_type::get(); + } + + /// Get the underlying endpoint in the native type. + /// TODO: make this nice and generic -> union like in tcp + data_type* data() { + return (boost::asio::detail::socket_addr_type*) &data_; + } + + /// Get the underlying endpoint in the native type. + const data_type* data() const { + return (boost::asio::detail::socket_addr_type*) &data_; + } + + /// Get the underlying size of the endpoint in the native type. + std::size_t size() const { + return sizeof(data_type); + } + + /// Set the underlying size of the endpoint in the native type. + void resize(std::size_t size) { + if (size > sizeof(data_type)) { + boost::system::system_error e(boost::asio::error::invalid_argument); + boost::throw_exception(e); + } + } + + /// Get the capacity of the endpoint in the native type. + std::size_t capacity() const { + return sizeof(data_type); + } + + /// Get the channel associated with the endpoint. The port number is always in + /// the host's byte order. + uint8_t channel() const { + return data_.rc_channel; + } + + /// Set the channel associated with the endpoint. The port number is always in + /// the host's byte order. + void channel(uint8_t channel_num) { + data_.rc_channel = channel_num; + } + + /// Get the Bluetooth address associated with the endpoint. + const bdaddr_t& address() const { + return data_.rc_bdaddr; + } + + /// Set the Bluetooth address associated with the endpoint. + void address(const boost::asio::ip::address& addr) { + bluetooth_endpoint tmp_endpoint(addr, channel()); + data_ = tmp_endpoint.data_; + } + + /// Compare two endpoints for equality. + friend bool operator==(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + return e1.address() == e2.address() && e1.channel() == e2.channel(); + } + + /// Compare two endpoints for inequality. + friend bool operator!=(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + return e1.address() != e2.address() || e1.channel() != e2.channel(); + } + + /// Compare endpoints for ordering. + friend bool operator<(const bluetooth_endpoint& e1, + const bluetooth_endpoint& e2) { + int cmp = bacmp(&e1.address(), &e2.address()); + + if (cmp == 0) { + return e1.channel() < e2.channel(); + } else { + return cmp < 0; + } + } + + friend ostream& operator<<( + ostream& out, + const bluetooth_endpoint& endp) { + char* mac_str = batostr(&endp.data_.rc_bdaddr); + out << "[" << mac_str << "]:" << (int)endp.data_.rc_channel; + free(mac_str); + return out; + } + +private: + // The underlying rfcomm socket address structure thingy. + //struct data_type data_; + struct sockaddr_rc data_; +}; + +template +const bdaddr_t bluetooth_endpoint::addr_any = { {0u, 0u, 0u, 0u, 0u, 0u} }; + +}}} // namespace boost::asio::bluetooth + +#endif /* HAVE_LIBBLUETOOTH */ +#endif /* BOOST_ASIO_BLUETOOTH_BLUETOOTH_ENDPOINT_HPP__ */ Index: source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp =================================================================== --- source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp (revision 10653) +++ source/ariba/utility/transport/rfcomm/bluetooth_rfcomm.hpp (revision 10653) @@ -0,0 +1,54 @@ +#ifndef BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ +#define BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ + +#include "bluetooth_endpoint.hpp" + +#include +#include + +namespace boost { +namespace asio { +namespace bluetooth { + +/** + * The rfcomm class contains flags necessary for RFCOMM sockets. + * + * @author Martin Florian + */ +class rfcomm { +public: + /// The type of endpoint. + typedef bluetooth_endpoint endpoint; + + /// Get an Instance. + /// We need this to fulfill the asio Endpoint requirements, I think. + static rfcomm get() { + return rfcomm(); + } + + /// Obtain an identifier for the type of the protocol. + int type() const { + return SOCK_STREAM; + } + + /// Obtain an identifier for the protocol. + int protocol() const { + return BTPROTO_RFCOMM; + } + + /// Obtain an identifier for the protocol family. + int family() const { + return AF_BLUETOOTH; + } + + /// The RFCOMM socket type, lets pray that this will work. + typedef basic_stream_socket socket; + + /// The RFCOMM acceptor type. + typedef basic_socket_acceptor acceptor; + +}; + +}}} // namespace boost::asio::bluetooth + +#endif /* BOOST_ASIO_BLUETOOTH_RFCOMM_HPP__ */ Index: source/ariba/utility/transport/rfcomm/rfcomm.cpp =================================================================== --- source/ariba/utility/transport/rfcomm/rfcomm.cpp (revision 7464) +++ (revision ) @@ -1,439 +1,0 @@ -#include "ariba/config.h" - -#ifdef HAVE_LIBBLUETOOTH - -#include "rfcomm.hpp" - -#include "ariba/utility/transport/asio/asio_io_service.h" -#include "ariba/utility/transport/asio/rfcomm.hpp" -#include "ariba/utility/transport/asio/bluetooth_endpoint.hpp" - -#include -#include -#include -#include -#include -#include - -namespace ariba { -namespace transport { - -use_logging_cpp(rfcomm) - -using namespace boost::asio; -using namespace detail; -using namespace std; - -using boost::system::error_code; - -class link_data { -public: - uint8_t size_[4]; - size_t size; - uint8_t* buffer; -}; - -class link_info { -public: - link_info(io_service& io ) : - io(io), up(false), connecting(false), local(), remote(), socket(new bluetooth::rfcomm::socket(io)), connect_retries(0), - size(0), buffer(NULL), sending(false) { - } - - ~link_info() { - delete socket; - } - - void reinit() { - delete socket; - socket = new bluetooth::rfcomm::socket(io); - up = false; - } - - io_service& io; - - // state - bool up; - bool connecting; - rfcomm_endpoint local, remote; - bluetooth::rfcomm::socket* socket; - int connect_retries; - - // read buffer - uint8_t size_[4]; - size_t size; - uint8_t* buffer; - - // send buffer - bool sending; - boost::mutex mutex; - std::deque send_buffer; -}; - -void rfcomm::shutdown(link_info* info) { - if (info != NULL && info->up) { - info->up = false; - info->socket->shutdown( bluetooth::rfcomm::socket::shutdown_both ); - } -} - - -inline bluetooth::rfcomm::endpoint convert( const rfcomm_endpoint& endpoint ) { - return bluetooth::rfcomm::endpoint( - endpoint.mac().bluetooth(), endpoint.channel().value() - ); -} - -inline rfcomm_endpoint convert( const bluetooth::rfcomm::endpoint& endpoint ) { - mac_address mac; - mac.bluetooth( endpoint.address() ); - rfcomm_channel_address channel; - channel.value( endpoint.channel() ); - return rfcomm_endpoint( mac, channel ); -} - - -rfcomm::rfcomm(uint16_t channel) : - channel(channel), io(asio_io_service::alloc()) { - accept_retries = 0; -} - -rfcomm::~rfcomm() { - asio_io_service::free(); -} - -void rfcomm::start() { - - // start io service - asio_io_service::start(); - - // create acceptor - logging_info( "Binding to channel " << channel ); - acceptor = new bluetooth::rfcomm::acceptor(io, - bluetooth::rfcomm::endpoint(bluetooth::rfcomm::get(), channel ) - ); - - send_data = new link_data(); - - // start accepting - start_accept(); -} - -void rfcomm::stop() { - logging_info( "Stopping asio rfcomm" ); -} - -void rfcomm::send(const address_v* remote, const uint8_t* data, size_t size) { - - // get end-point - rfcomm_endpoint endpoint = *remote; - endpoint = convert(convert(endpoint)); - - // try to find established connector - logging_debug("Trying to find a already existing link."); - link_info* info = NULL; - for (size_t i = 0; i < links.size(); i++) - if (links[i]->remote.mac() == endpoint.mac()) { - logging_debug("Using already established link"); - info = links[i]; - break; - } - - // not found, or not up? ->try to (re-)connect - if (info==NULL || ((!info->up || !info->socket->is_open()) && !info->connecting) ) { - logging_debug( "Connecting to " << endpoint.to_string() ); - if (info != NULL && (!info->up || !info->socket->is_open())) { - logging_error("Old link is down. Trying to re-establish link."); - info->reinit(); - } else { - info = new link_info(io); - } - info->connect_retries = 0; - info->remote = endpoint; - info->connecting = true; - info->socket->async_connect( convert(endpoint), boost::bind( - &rfcomm::handle_connect, this, - boost::asio::placeholders::error, info - )); - links.push_back(info); - } - - // copy message - link_data ldata; - ldata.size = size; - ldata.size_[0] = (size >> 24) & 0xFF; - ldata.size_[1] = (size >> 16) & 0xFF; - ldata.size_[2] = (size >> 8) & 0xFF; - ldata.size_[3] = (size >> 0) & 0xFF; - ldata.buffer = new uint8_t[size]; - memcpy(ldata.buffer, data, size); - - // enqueue message - info->mutex.lock(); - info->send_buffer.push_back(ldata); - info->mutex.unlock(); - - // start writing - io.post( boost::bind( &rfcomm::start_write, this, info) ); -} - -void rfcomm::send(const endpoint_set& endpoints, const uint8_t* data, size_t size) { - // send a message to each combination of mac-address and channel - BOOST_FOREACH( const mac_address mac, endpoints.bluetooth ) { - BOOST_FOREACH( const rfcomm_channel_address channel, endpoints.rfcomm ) { - rfcomm_endpoint endpoint(mac, channel); - address_vf vf = endpoint; - send(vf,data,size); - } - } -} - -void rfcomm::terminate( const address_v* remote) { - // get end-point - rfcomm_endpoint endpoint = *remote; - - for (size_t i = 0; i < links.size(); i++) - if (links[i]->remote.mac() == endpoint.mac()) { - // close socket - shutdown(links[i]); - break; - } -} - -void rfcomm::register_listener(transport_listener* listener) { - this->listener = listener; -} - -void rfcomm::start_accept() { - - logging_info( "Waiting for connections ..." ); - - // start accepting a connection - link_info* info = new link_info(io); - acceptor->async_accept(*info->socket, boost::bind( - // bind parameters - &rfcomm::handle_accept, this, - - // handler parameters - boost::asio::placeholders::error, info - )); - asio_io_service::start(); -} - -void rfcomm::handle_accept(const error_code& error, link_info* info) { - if (error) { - logging_error( "Error waiting for new connections. Error code: "<< error.message() - << ", trying to recover (attempt " << accept_retries << ")"); - - // restart accepting - if (accept_retries<3) { - accept_retries++; - start_accept(); - } else - delete info; - - return; - } - - links_mutex.lock(); - - // convert endpoints - info->up = true; - info->local = convert( info->socket->local_endpoint() ); - info->remote = convert( info->socket->remote_endpoint() ); - - logging_debug("Accepted incoming connection from " - << info->remote.to_string() ); - - // add to list - links.push_back(info); - links_mutex.unlock(); - - // start i/o - start_read(info); - start_write(info); - - // restart accept - start_accept(); -} - -void rfcomm::handle_connect( const error_code& error, link_info* info ) { - if (error) { - logging_error( "Can not connect. Error code: " - << error.message() << " Retrying ... " - "(attempt " << info->connect_retries << ")" ); - - // do we retry this connection? yes-> - if (info->connect_retries<3) { - // increase counter - info->connecting = false; - info->connect_retries++; - info->reinit(); - - // retry connection attempt - info->socket->async_connect( convert(info->remote), boost::bind( - &rfcomm::handle_connect, this, - boost::asio::placeholders::error, info - )); - - } else { // no-> delete link and stop - return; - } - } - - // convert endpoints - info->up = true; - info->connecting = false; - info->local = convert( info->socket->local_endpoint() ); - info->remote = convert( info->socket->remote_endpoint() ); - - logging_debug( "Connected to " << info->remote.to_string() ); - - // add to list - links_mutex.lock(); - links.push_back(info); - links_mutex.unlock(); - - // start i/o - start_read(info); - start_write(info); -} - -void rfcomm::start_read(link_info* info) { - logging_debug("Start reading ..."); - - // start read - boost::asio::async_read(*info->socket, - - // read size of packet - boost::asio::buffer(info->size_, 4), - - // bind handler - boost::bind( - - // bind parameters - &rfcomm::handle_read_header, this, - - // handler parameters - placeholders::error, placeholders::bytes_transferred, info - ) - ); -} - -void rfcomm::handle_read_header(const error_code& error, size_t bytes, - link_info* info) { - - // handle error - if (error) { - logging_error("Failed to receive message payload. Error code: " - << error.message() ); - shutdown(info); - return; - } - - // ignore errors and wait for all data to be received - if (bytes != 4) return; - - // get size - info->size = (info->size_[0]<<24) + (info->size_[1] << 16) + - (info->size_[2]<< 8) + (info->size_[3] << 0); - - // allocate buffer - info->buffer = new uint8_t[info->size]; - - // start read - boost::asio::async_read(*info->socket, - // read size of packet - boost::asio::buffer(info->buffer, info->size), - // bind handler - boost::bind( - // bind parameters - &rfcomm::handle_read_data, this, - // handler parameters - placeholders::error, placeholders::bytes_transferred, info - ) - ); -} - -void rfcomm::handle_read_data(const error_code& error, size_t bytes, - link_info* info) { - - // check error - if (error) { - logging_error("Failed to receive message payload. Error: " << error.message() ); - shutdown(info); - return; - } - - // wait for all data to be received - if (bytes != info->size) - return; - - // deliver data - listener->receive_message(this, info->local, info->remote, info->buffer, info->size ); - - // free buffers and reset size buffer - delete [] info->buffer; - for (size_t i=0; i<4; i++) info->size_[i] = 0; - - start_read(info); -} - -void rfcomm::start_write( link_info* info ) { - // do not start writing if sending is in progress - if (info->sending || !info->up || info->send_buffer.size()==0) return; - - // set sending flag - info->sending = true; - - // safely remove data from deque - *send_data = info->send_buffer.front(); - info->send_buffer.pop_front(); - - boost::array buffer; - buffer[0] = boost::asio::buffer(send_data->size_,4); - buffer[1] = boost::asio::buffer(send_data->buffer,send_data->size); - - // start writing - boost::asio::async_write(*info->socket, - // read size of packet - buffer, - // bind handler - boost::bind( - // bind parameters - &rfcomm::handle_write_data, this, - // handler parameters - placeholders::error, placeholders::bytes_transferred, - info, send_data->size, send_data->buffer - ) - ); -} - -void rfcomm::handle_write_data(const error_code& error, size_t bytes, - link_info* info, size_t size, uint8_t* buffer ) { - - // handle error - if (error) { - logging_error( "Message sent error. Error: " << error.message() ); - info->sending = false; - shutdown(info); - return; - } - - // wait for all data to be sent - if (bytes != (size+4) ) - return; - - logging_debug( "Message sent" ); - - // free buffer - delete [] buffer; - info->sending = false; - - // restart-write - start_write(info); -} - -}} // namespace ariba::transport - -#endif Index: source/ariba/utility/transport/rfcomm/rfcomm.hpp =================================================================== --- source/ariba/utility/transport/rfcomm/rfcomm.hpp (revision 7464) +++ (revision ) @@ -1,78 +1,0 @@ -#include "ariba/config.h" - -#ifdef HAVE_LIBBLUETOOTH - -#ifndef RFCOMM_HPP_ -#define RFCOMM_HPP_ - -#include "ariba/utility/transport/transport.hpp" -#include "ariba/utility/transport/asio/asio_io_service.h" -#include "ariba/utility/transport/asio/bluetooth_endpoint.hpp" -#include "ariba/utility/transport/asio/rfcomm.hpp" - -#include -#include -#include - -#include "ariba/utility/logging/Logging.h" - -namespace ariba { -namespace transport { - -using boost::system::error_code; -using namespace boost::asio; - -class link_info; -class link_data; - -/** - * TODO: Doc - * - * @author Sebastian Mies - */ -class rfcomm : public transport_protocol { - use_logging_h(rfcomm) -public: - rfcomm( uint16_t channel ); - virtual ~rfcomm(); - virtual void start(); - virtual void stop(); - virtual void send( const address_v* remote, const uint8_t* data, size_t size ); - virtual void send( const endpoint_set& endpoints, const uint8_t* data, size_t size ); - virtual void terminate( const address_v* remote ); - virtual void register_listener( transport_listener* listener ); - -private: - uint16_t channel; - boost::mutex links_mutex; - vector links; - io_service& io; - transport_listener* listener; - bluetooth::rfcomm::acceptor* acceptor; - int accept_retries; - link_data* send_data; - - void start_accept(); - - void handle_accept( const error_code& error, link_info* info ); - - void start_read( link_info* info ); - - void handle_read_header( const error_code& error, size_t bytes, link_info* info ); - - void handle_read_data( const error_code& error, size_t bytes, link_info* info ); - - void handle_connect( const error_code& error, link_info* info ); - - void start_write( link_info* info ); - - void handle_write_data(const error_code& error, size_t bytes, - link_info* info, size_t size, uint8_t* buffer ); - - void shutdown(link_info* info); -}; - -}} // namespace ariba::transport - -#endif /* RFCOMM_HPP_ */ -#endif Index: source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp =================================================================== --- source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp (revision 10653) +++ source/ariba/utility/transport/rfcomm/rfcomm_transport.cpp (revision 10653) @@ -0,0 +1,464 @@ +#include "rfcomm_transport.hpp" + +#ifdef HAVE_LIBBLUETOOTH +#include + +namespace ariba { +namespace transport { + +use_logging_cpp(rfcomm_transport) + +using namespace ariba::addressing; + +typedef boost::mutex::scoped_lock unique_lock; + +/* constructor */ +rfcomm_transport::rfcomm_transport( const rfcomm_transport::rfcomm::endpoint& endp ) : + listener(NULL), + acceptor(u_io_service.get_asio_io_service(), endp) +{ +} + +rfcomm_transport::~rfcomm_transport(){} + +void rfcomm_transport::start() +{ + // open server socket + accept(); + + u_io_service.start(); +} + + +void rfcomm_transport::stop() +{ + acceptor.close(); + + u_io_service.stop(); +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const rfcomm::endpoint& dest_addr, + reboost::message_t message, + uint8_t priority) +{ + ConnPtr conn; + bool need_to_connect = false; + + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(dest_addr); + if (it == connections.end()) + { + ConnPtr tmp_ptr( + new rfcomm_connection( + u_io_service.get_asio_io_service(), + shared_from_this() ) + ); + conn = tmp_ptr; + + conn->partner = dest_addr; + conn->remote = convert_address(dest_addr); + + // Note: starting the send is the obligation of the connect_handler + // (avoids trying to send while not connected yet) + conn->sending = true; + need_to_connect = true; + + ConnectionMap::value_type item(dest_addr, conn); + connections.insert(item); + + } else { + conn = it->second; + } + } + + + // * the actual send * + conn->enqueue_for_sending(message, priority); + + // if new connection connect to the other party + if ( need_to_connect ) + { + conn->sock.async_connect( + dest_addr, + boost::bind( + &rfcomm_connection::async_connect_handler, + conn, + boost::asio::placeholders::error)); + } +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const address_v* remote, + reboost::message_t message, + uint8_t priority) +{ + send(convert_address(remote), message, priority); +} + + +/* see header file for comments */ +void rfcomm_transport::send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority ) +{ + // send a message to each combination of address-address and port + BOOST_FOREACH( const mac_address mac, endpoints.bluetooth ) { + BOOST_FOREACH( const rfcomm_channel_address channel, endpoints.rfcomm ) { + rfcomm::endpoint endp(mac.bluetooth(), channel.value()); + + // * send * + send(endp, message, priority); + } + } +} + + +void rfcomm_transport::register_listener( transport_listener* listener ) +{ + this->listener = listener; +} + + +void rfcomm_transport::terminate( const address_v* remote ) +{ + terminate(convert_address(remote)); +} + +void rfcomm_transport::terminate( const rfcomm::endpoint& remote ) +{ + ConnPtr conn; + + // find and forget connection + { + unique_lock lock(connections_lock); + + ConnectionMap::iterator it = connections.find(remote); + if (it == connections.end()) + { + return; + } + + conn = it->second; + + connections.erase(it); + } + + // close connection + boost::system::error_code ec; + conn->sock.shutdown(tcp::socket::shutdown_both, ec); + conn->sock.close(ec); +} + + +/* private */ +void rfcomm_transport::accept() +{ + // create new connection object + ConnPtr conn( + new rfcomm_connection( + u_io_service.get_asio_io_service(), + shared_from_this() + ) + ); + + // wait for incoming connection + acceptor.async_accept( + conn->sock, + boost::bind(&self::async_accept_handler, + this->shared_from_this(), + conn, + boost::asio::placeholders::error) + ); +} + +void rfcomm_transport::async_accept_handler(ConnPtr conn, const error_code& error) +{ + if ( ! error ) + { + conn->partner = conn->sock.remote_endpoint(); + conn->remote = convert_address(conn->partner); + conn->local = convert_address(conn->sock.local_endpoint()); + + { + unique_lock lock(connections_lock); + + ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn); + connections.insert(item); + } + + // read + conn->listen(); + } + + // accept further connections + accept(); +} + +inline rfcomm_transport::rfcomm::endpoint +rfcomm_transport::convert_address( const address_v* address ) +{ + rfcomm_endpoint endpoint = *address; + + return rfcomm::endpoint( + endpoint.mac().bluetooth(), endpoint.channel().value() + ); +} + + +inline rfcomm_endpoint rfcomm_transport::convert_address(const rfcomm::endpoint& endpoint) +{ + mac_address mac; + mac.bluetooth(endpoint.address()); + rfcomm_channel_address channel; + channel.value(endpoint.channel()); + return rfcomm_endpoint(mac, channel); +} + + +/***************** + ** inner class ** + *****************/ + +rfcomm_transport::rfcomm_connection::rfcomm_connection( + boost::asio::io_service & io_service, + rfcomm_transport::sptr parent) : + sock(io_service), + valid(true), + parent(parent), + out_queues(8), //TODO How much priorities shall we have? + sending(false) +{ + header.length = 0; + header.prot = 0; +} + +/*------------------------------------------- + | implement transport_connection interface | + -------------------------------------------*/ +void rfcomm_transport::rfcomm_connection::send( + reboost::message_t message, + uint8_t priority) +{ + enqueue_for_sending(message, priority); +} + + +address_vf rfcomm_transport::rfcomm_connection::getLocalEndpoint() +{ + return local; +} + + +address_vf rfcomm_transport::rfcomm_connection::getRemoteEndpoint() +{ + return remote; +} + + +void rfcomm_transport::rfcomm_connection::terminate() +{ + parent->terminate(partner); +} + + +/*------------------------------ + | things we defined ourselves | + ------------------------------*/ +void rfcomm_transport::rfcomm_connection::async_connect_handler(const error_code& error) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // save address in ariba format + local = parent->convert_address(sock.local_endpoint()); + + // Note: sending has to be true at this point + send_next_package(); + + listen(); +} + + +void rfcomm_transport::rfcomm_connection::listen() +{ + boost::asio::async_read( + this->sock, + boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)), + boost::bind( + &rfcomm_transport::rfcomm_connection::async_read_header_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + + +void rfcomm_transport::rfcomm_connection::async_read_header_handler(const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + // convert byte order + header.length = ntohl(header.length); + header.length -= 2; // XXX protlib + + assert(header.length > 0); + + // new buffer for the new packet + buffy = shared_buffer_t(header.length); + + // * read data * + boost::asio::async_read( + this->sock, + boost::asio::buffer(buffy.mutable_data(), buffy.size()), + boost::bind( + &rfcomm_transport::rfcomm_connection::async_read_data_handler, + this->shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred + ) + ); +} + +void rfcomm_transport::rfcomm_connection::async_read_data_handler( + const error_code& error, size_t bytes_transferred) +{ + if (error) + { + parent->terminate(partner); + + return; + } + + message_t msg; + msg.push_back(buffy); + buffy = shared_buffer_t(); + + if ( parent->listener ) + parent->listener->receive_message(shared_from_this(), msg); + + listen(); +} + +/* see header file for comments */ +void rfcomm_transport::rfcomm_connection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred) +{ + if ( error ) + { + // remove this connection + parent->terminate(partner); + + return; + } + + send_next_package(); +} + + + +void rfcomm_transport::rfcomm_connection::enqueue_for_sending(Packet packet, uint8_t priority) +{ + bool restart_sending = false; + + // enqueue packet [locked] + { + unique_lock(out_queues_lock); + + assert( priority < out_queues.size() ); + out_queues[priority].push(packet); + + if ( ! sending ) + { + restart_sending = true; + sending = true; + } + } + + // if sending was stopped, we have to restart it here + if ( restart_sending ) + { + send_next_package(); + } +} + +/* see header file for comments */ +void rfcomm_transport::rfcomm_connection::send_next_package() +{ + Packet packet; + bool found = false; + + // find packet with highest priority [locked] + { + unique_lock(out_queues_lock); + + for ( vector::iterator it = out_queues.begin(); + it != out_queues.end(); it++ ) + { + if ( !it->empty() ) + { + packet = it->front(); + it->pop(); + found = true; + + break; + } + } + + // no packets waiting --> stop sending + if ( ! found ) + { + sending = false; + } + } + + // * send * + if ( found ) + { + reboost::shared_buffer_t header_buf(sizeof(header_t)); + header_t* header = (header_t*)(header_buf.mutable_data()); + header->length = htonl(packet.size()+2); // XXX protlib + + packet.push_front(header_buf); + + // "convert" message to asio buffer sequence + vector send_sequence(packet.length()); + for ( int i=0; i < packet.length(); i++ ) + { + shared_buffer_t b = packet.at(i); + send_sequence.push_back(boost::asio::buffer(b.data(), b.size())); + } + + // * async write * + boost::asio::async_write( + this->sock, + send_sequence, + boost::bind( + &rfcomm_transport::rfcomm_connection::async_write_handler, + this->shared_from_this(), + packet, // makes sure our shared pointer lives long enough ;-) + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred) + ); + } +} + +}} // namespace ariba::transport + +#endif /* HAVE_LIBBLUETOOTH */ Index: source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp =================================================================== --- source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp (revision 10653) +++ source/ariba/utility/transport/rfcomm/rfcomm_transport.hpp (revision 10653) @@ -0,0 +1,170 @@ +#include "ariba/config.h" + +#ifdef HAVE_LIBBLUETOOTH + +#ifndef RFCOMM_TRANSPORT_HPP_ +#define RFCOMM_TRANSPORT_HPP_ + +#include "ariba/utility/transport/transport.hpp" +#include "ariba/utility/transport/asio/unique_io_service.h" +#include "ariba/utility/transport/transport_connection.hpp" +#include "ariba/utility/addressing/rfcomm_endpoint.hpp" +#include +#include +#include +#include +#include "ariba/utility/transport/messages/buffers.hpp" +#include "ariba/utility/logging/Logging.h" +#include "bluetooth_rfcomm.hpp" + +namespace ariba { +namespace transport { + +using namespace std; +using ariba::transport::detail::unique_io_service; +using ariba::addressing::rfcomm_endpoint; +using boost::system::error_code; +using reboost::shared_buffer_t; +using reboost::message_t; + + +class rfcomm_transport : + public transport_protocol, + public boost::enable_shared_from_this +{ +public: + typedef boost::shared_ptr sptr; + +private: + typedef rfcomm_transport self; + typedef boost::asio::bluetooth::rfcomm rfcomm; + use_logging_h(rfcomm_transport) + + class rfcomm_connection : + public transport_connection, + public boost::enable_shared_from_this + { + public: + typedef reboost::message_t Packet; + typedef std::queue OutQueue; + + struct header_t + { + uint32_t length; + uint16_t prot; // XXX protlib + } __attribute__((packed)); + + rfcomm_connection(boost::asio::io_service& io_service, + rfcomm_transport::sptr parent); + + /// Inherited from transport_connection + virtual void send(reboost::message_t message, uint8_t priority = 0); + virtual address_vf getLocalEndpoint(); + virtual address_vf getRemoteEndpoint(); + virtual void terminate(); + + void listen(); + + void async_connect_handler(const error_code& error); + + void async_read_header_handler(const error_code& error, size_t bytes_transferred); + void async_read_data_handler(const error_code& error, size_t bytes_transferred); + + /* + * is called from asio when write operation "returns", + * calls private function `send_next_package()` + */ + void async_write_handler( + reboost::shared_buffer_t packet, + const error_code& error, + size_t bytes_transferred); + + + void enqueue_for_sending(Packet packet, uint8_t priority); + + private: + /* + * is called from `send` or `async_write_handler` to begin/keep sending + * sends the next message with the highest priority in this connection + */ + void send_next_package(); + + + public: + rfcomm::socket sock; + bool valid; + rfcomm_transport::sptr parent; + + rfcomm::endpoint partner; + rfcomm_endpoint remote; + rfcomm_endpoint local; + + vector out_queues; // to be locked with out_queues_lock + boost::mutex out_queues_lock; + + bool sending; // to be locked with out_queues_lock + + header_t header; + shared_buffer_t buffy; + }; + typedef boost::shared_ptr ConnPtr; + typedef std::map ConnectionMap; + +public: + /* constructor */ + rfcomm_transport( const rfcomm::endpoint& endp ); + virtual ~rfcomm_transport(); + + virtual void start(); + virtual void stop(); + + /** + * enqueues message for sending + * create new connection if necessary + * starts sending mechanism (if not already running) + */ + void send( + const rfcomm::endpoint&, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * Converts address_v to rfcomm::endpoint and calls the real send() function + */ + virtual void send( + const address_v* remote, + reboost::message_t message, + uint8_t priority = 0 ); + + /** + * calls send for each destination endpoint in `endpoint_set& endpoints` + */ + virtual void send( + const endpoint_set& endpoints, + reboost::message_t message, + uint8_t priority = 0 ); + + virtual void terminate( const address_v* remote ); + virtual void terminate( const rfcomm::endpoint& remote ); + virtual void register_listener( transport_listener* listener ); + + +private: + void accept(); + void async_accept_handler(ConnPtr conn, const error_code& error); + rfcomm::endpoint convert_address(const address_v* endpoint); + rfcomm_endpoint convert_address(const rfcomm::endpoint& endpoint); + +private: + transport_listener* listener; + unique_io_service u_io_service; + rfcomm::acceptor acceptor; + + ConnectionMap connections; + boost::mutex connections_lock; +}; + +}} // namespace ariba::transport + +#endif /* RFCOMM_TRANSPORT_HPP_ */ +#endif /* HAVE_LIBBLUETOOTH */