/* -*- c++ -*- */ /* * Copyright 2012 Free Software Foundation, Inc. * * This file is part of GNU Radio * * GNU Radio is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 3, or (at your option) * any later version. * * GNU Radio is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with GNU Radio; see the file COPYING. If not, write to * the Free Software Foundation, Inc., 51 Franklin Street, * Boston, MA 02110-1301, USA. */ #ifndef INCLUDED_GR_SOCKET_PDU_H #define INCLUDED_GR_SOCKET_PDU_H #include #include #include #include #include #include #include #include class gr_socket_pdu; typedef boost::shared_ptr gr_socket_pdu_sptr; GR_CORE_API gr_socket_pdu_sptr gr_make_socket_pdu (std::string type, std::string addr, std::string port, int MTU=10000); class tcp_connection : public boost::enable_shared_from_this { public: typedef boost::shared_ptr pointer; gr_socket_pdu *d_block; boost::array buf; static pointer create(boost::asio::io_service& io_service) { return pointer(new tcp_connection(io_service)); } boost::asio::ip::tcp::socket& socket() { return socket_; } void start(gr_socket_pdu* parent) { d_block = parent; // message_ = "connected to gr_socket_pdu\n"; // boost::asio::async_write(socket_, boost::asio::buffer(message_), // boost::bind(&tcp_connection::handle_write, shared_from_this(), // boost::asio::placeholders::error, // boost::asio::placeholders::bytes_transferred)); socket_.async_read_some( boost::asio::buffer(buf), boost::bind(&tcp_connection::handle_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } void send(pmt::pmt_t vector){ size_t len = pmt::pmt_length(vector); size_t offset(0); boost::array txbuf; memcpy(&txbuf[0], pmt::pmt_uniform_vector_elements(vector, offset), len); boost::asio::async_write(socket_, boost::asio::buffer(txbuf, len), boost::bind(&tcp_connection::handle_write, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } ~tcp_connection(){ // std::cout << "tcp_connection destroyed\n"; } private: tcp_connection(boost::asio::io_service& io_service) : socket_(io_service) { } void handle_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred); void handle_write(const boost::system::error_code& /*error*/, size_t /*bytes_transferred*/) { } boost::asio::ip::tcp::socket socket_; std::string message_; }; /*! * \brief Gather received items into messages and insert into msgq * \ingroup sink_blk */ class GR_CORE_API gr_socket_pdu : public gr_stream_pdu_base { private: friend GR_CORE_API gr_socket_pdu_sptr gr_make_socket_pdu(std::string type, std::string addr, std::string port, int MTU); boost::asio::io_service _io_service; boost::array rxbuf; // tcp specific boost::asio::ip::tcp::endpoint _tcp_endpoint; // specific to tcp server boost::shared_ptr _acceptor_tcp; std::vector d_tcp_connections; void tcp_server_send(pmt::pmt_t msg); void tcp_client_send(pmt::pmt_t msg); void udp_send(pmt::pmt_t msg); // specific to tcp client boost::shared_ptr _tcp_socket; // specific to udp client/server boost::asio::ip::udp::endpoint _udp_endpoint; boost::asio::ip::udp::endpoint _udp_endpoint_other; boost::shared_ptr _udp_socket; void handle_receive(const boost::system::error_code& error, std::size_t ){ } void start_tcp_accept(){ tcp_connection::pointer new_connection = tcp_connection::create(_acceptor_tcp->get_io_service()); _acceptor_tcp->async_accept(new_connection->socket(), boost::bind(&gr_socket_pdu::handle_tcp_accept, this, new_connection, boost::asio::placeholders::error)); } void handle_tcp_accept(tcp_connection::pointer new_connection, const boost::system::error_code& error){ if (!error) { new_connection->start(this); d_tcp_connections.push_back(new_connection); start_tcp_accept(); } else { std::cout << error << std::endl; } } void run_io_service(){ _io_service.run(); } void handle_udp_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred){ if(!error){ pmt::pmt_t vector = pmt::pmt_init_u8vector(bytes_transferred, (const uint8_t*)&rxbuf[0]); pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector); message_port_pub( pmt::mp("pdus"), pdu ); _udp_socket->async_receive_from( boost::asio::buffer(rxbuf), _udp_endpoint_other, boost::bind(&gr_socket_pdu::handle_udp_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } else { throw boost::system::system_error(error); // std::cout << "error occurred\n"; } } void handle_tcp_read(const boost::system::error_code& error/*error*/, size_t bytes_transferred){ if(!error) { pmt::pmt_t vector = pmt::pmt_init_u8vector(bytes_transferred, (const uint8_t*)&rxbuf[0]); pmt::pmt_t pdu = pmt::pmt_cons( pmt::PMT_NIL, vector); message_port_pub( pmt::mp("pdus"), pdu ); _tcp_socket->async_read_some( boost::asio::buffer(rxbuf), boost::bind(&gr_socket_pdu::handle_tcp_read, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } else { //std::cout << "error occurred\n"; throw boost::system::system_error(error); } } protected: gr_socket_pdu (std::string type, std::string addr, std::string port, int MTU=10000); public: ~gr_socket_pdu () {} }; #endif /* INCLUDED_GR_TUNTAP_PDU_H */