Boost::Asio::Ip::Tcp::Iostream questions

Question

Hey all, I'm new to asio and boost, I've been trying to implement a TCP Server & Client so that I could transmit an std::vector - but I've failed so far. I'm finding the boost documentation of Asio lacking (to say the least) and hard to understand (english is not my primary language).

In any case, I've been looking at the iostreams examples and I've been trying to implement an object oriented solution - but I've failed.

The server that I'm trying to implement should be able to accept connections from multiple clients (How do I do that ?)

The server should receive the std::vector, /* Do something */ and then return it to the client so that the client can tell that the server received the data intact.

*.h file

class TCP_Server : private boost::noncopyable
     {
      typedef boost::shared_ptr<TCP_Connection> tcp_conn_pointer;

      public :
       TCP_Server(ba::io_service &io_service, int port);
       virtual ~TCP_Server() {}
       virtual void Start_Accept();
      private:
       virtual void Handle_Accept(const boost::system::error_code& e);
      private :
       int     m_port;
       ba::io_service&  m_io_service;    // IO Service
       bi::tcp::acceptor m_acceptor;   // TCP Connections acceptor
       tcp_conn_pointer m_new_tcp_connection; // New connection pointer
     };

*.cpp file

TCP_Server::TCP_Server(boost::asio::io_service &io_service, int port) : 
      m_io_service(io_service), 
      m_acceptor(io_service, bi::tcp::endpoint(bi::tcp::v4(), port)), 
      m_new_tcp_connection(TCP_Connection::Create(io_service))
     {
      m_port = port;
      Start_Accept();
     }

     void TCP_Server::Start_Accept()
     {
      std::cout << "[TCP_Server][Start_Accept] => Listening on port : " << m_port << std::endl;
      //m_acceptor.async_accept(m_new_tcp_connection->Socket(),
            //                        boost::bind(&TCP_Server::Handle_Accept, this,
            //                                    ba::placeholders::error));


       m_acceptor.async_accept(*m_stream.rdbuf(),
       boost::bind(&TCP_Server::Handle_Accept, 
       this,
       ba::placeholders::error));
     }

     void TCP_Server::Handle_Accept(const boost::system::error_code &e)
     {
      if(!e)
      {

       /*boost::thread T(boost::bind(&TCP_Connection::Run, m_new_tcp_connection));
       std::cout << "[TCP_Server][Handle_Accept] => Accepting incoming connection. Launching Thread " << std::endl;
       m_new_tcp_connection = TCP_Connection::Create(m_io_service);
       m_acceptor.async_accept(m_new_tcp_connection->Socket(), 
             boost::bind(&TCP_Server::Handle_Accept, 
                this, 
                ba::placeholders::error));*/
       m_stream << "Server Response..." << std::endl;
      }
     }

How should the client look ? How do I keep the connection alive while both apps "talk" ?

Answer 1

I believe the code you have posted is a little incomplete/incorrect. Nonetheless, here is some guidance..

1) Your async_accept() call seems wrong. It should be something like,

m_acceptor.async_accept(m_new_tcp_connection->socket(),...)

2) Take note that the Handle_Accept() function will be called after the socket is accepted. In other words, when control reaches Handle_Accept(), you simply have to write to the socket. Something like

void TCP_Server::Handle_Accept(const system::error_code& error)
{
  if(!error)
  {
    //send data to the client
    string message = "hello there!\n";

    //Write data to the socket and then call the handler AFTER that
    //Note, you will need to define a Handle_Write() function in your TCP_Connection class.
async_write(m_new_tcp_connection->socket(),buffer(message),bind(&TCP_Connection::Handle_Write, this,placeholders::error,placeholders::bytes_transferred)); 

    //accept the next connection
    Start_Accept();
  }
}

3) As for the client, you should take a look here: http://www.boost.org/doc/libs/1_39_0/doc/html/boost_asio/tutorial/tutdaytime1.html

Answer 2

If your communication on both ends is realized in C++ you can use Boost Serialization library to sezilize the vector into bytes and transfer these to the other machine. On the opposite end you will use boost serialization lib to desirialize the object. I saw at least two approaches doing so.

Advantage of Boost Serialization: this approach works when transferring objects between 32bit and 64bit systems as well.

Below are the links:
code project article
boost mailing list ideas

Regards,
Ovanes

Answer 3

AFAIK ASIO iostreams are only for synchronous I/O. But your example gives me a hint that you want to use asynchronous I/O. Here is a small example of a server which uses async I/O to read a request comprising of an array of integers preceded by 4 byte count of the integers in the request. So in effect I am serializing a vector of integerss as count(4 bytes) int int ... etc if reading the vector of ints is successful, the server will write a 4 byte response code(=1) and then issue a read for a new request from the client. Enough said, Code follows.

#include <iostream>
#include <vector>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/asio.hpp>

using namespace boost::asio;
using boost::asio::ip::tcp;

class Connection
{
public:
    Connection(tcp::acceptor& acceptor)
     : acceptor_(acceptor), socket_(acceptor.get_io_service(), tcp::v4())
    {
    }
    void start()
    {
     acceptor_.get_io_service().post(boost::bind(&Connection::start_accept, this));
    }
private:
    void start_accept()
    {
     acceptor_.async_accept(socket_,boost::bind(&Connection::handle_accept, this, 
      placeholders::error));
    }
    void handle_accept(const boost::system::error_code& err)
    {
     if (err)
     {
      //Failed to accept the incoming connection.
      disconnect();
     }
     else
     {
      count_ = 0;
      async_read(socket_, buffer(&count_, sizeof(count_)),
       boost::bind(&Connection::handle_read_count,
       this, placeholders::error, placeholders::bytes_transferred));
     }
    }
    void handle_read_count(const boost::system::error_code& err, std::size_t bytes_transferred)
    {
     if (err || (bytes_transferred != sizeof(count_))
     {
      //Failed to read the element count.
      disconnect();
     }
     else
     {
      elements_.assign(count_, 0);
      async_read(socket_, buffer(elements_), 
       boost::bind(&Connection::handle_read_elements, this,
       placeholders::error, placeholders::bytes_transferred));
     }
    }

    void handle_read_elements(const boost::system::error_code& err, std::size_t bytes_transferred)
    {
     if (err || (bytes_transferred != count_ * sizeof(int)))
     {
      //Failed to read the request elements.
      disconnect();
     }
     else
     {
      response_ = 1;
      async_write(socket_, buffer(&response_, sizeof(response_)),
       boost::bind(&Connection::handle_write_response, this,
       placeholders::error, placeholders::bytes_transferred));
     }
    }
    void handle_write_response(const boost::system::error_code& err, std::size_t bytes_transferred)
    {
     if (err)
      disconnect();
     else
     {
      //Start a fresh read 
      count_ = 0;
      async_read(socket_, buffer(&count_, sizeof(count_)),
       boost::bind(&Connection::handle_read_count,
       this, placeholders::error, placeholders::bytes_transferred));
     }
    }
    void disconnect()
    {
     socket_.shutdown(tcp::socket::shutdown_both);
     socket_.close();
     socket_.open(tcp::v4());
     start_accept();
    }
    tcp::acceptor& acceptor_;
    tcp::socket socket_;
    std::vector<int> elements_;
    long count_;
    long response_;
};

class Server : private boost::noncopyable
{
public:
    Server(unsigned short port, unsigned short thread_pool_size, unsigned short conn_pool_size)
     : acceptor_(io_service_, tcp::endpoint(tcp::v4(), port), true)
    {
     unsigned short i = 0;
     for (i = 0; i < conn_pool_size; ++i)
     {
      ConnectionPtr conn(new Connection(acceptor_));
      conn->start();
      conn_pool_.push_back(conn);
     }

     // Start the pool of threads to run all of the io_services.
     for (i = 0; i < thread_pool_size; ++i)
     {
      thread_pool_.create_thread(boost::bind(&io_service::run, &io_service_));
     }
    } 
    ~Server()
    {
     io_service_.stop();
     thread_pool_.join_all();
    }

private:
    io_service io_service_;
    tcp::acceptor acceptor_;
    typedef boost::shared_ptr<Connection> ConnectionPtr;
    std::vector<ConnectionPtr> conn_pool_;
    boost::thread_group thread_pool_;
};

boost::function0<void> console_ctrl_function;

BOOL WINAPI console_ctrl_handler(DWORD ctrl_type)
{
  switch (ctrl_type)
  {
  case CTRL_C_EVENT:
  case CTRL_BREAK_EVENT:
  case CTRL_CLOSE_EVENT:
  case CTRL_SHUTDOWN_EVENT:
    console_ctrl_function();
    return TRUE;
  default:
    return FALSE;
  }
}

void stop_server(Server* pServer)
{
    delete pServer;
    pServer = NULL;
}

int main()
{
    Server *pServer = new Server(10255, 4, 20);
    console_ctrl_function = boost::bind(stop_server, pServer);
    SetConsoleCtrlHandler(console_ctrl_handler, TRUE);
    while(true)
    {
     Sleep(10000);
    }
}