Working with boost::asio::streambuf

Looking for a boost::asio (and with himself boost) decided to write asynchronous server. To store incoming data I use boost::asio::streambuf. Here I have a problem. When I receive a second message from the client and subsequent I see that in the buffer contains a data from previous messages. Although I call Consume method at the input buffer. What's wrong?

class tcp_connection
// Using shared_ptr and enable_shared_from_this 
// because we want to keep the tcp_connection object alive 
// as long as there is an operation that refers to it.
: public boost::enable_shared_from_this<tcp_connection>
{
...

boost::asio::streambuf receive_buffer;

boost::asio::io_service::strand strand;
}

...

void tcp_connection::receive()
{
// Read the response status line. The response_ streambuf will
// automatically grow to accommodate the entire line. The growth may be
// limited by passing a maximum size to the streambuf constructor.
boost::asio::async_read_until(m_socket, receive_buffer, "\r\n",
    strand.wrap(boost::bind(&tcp_connection::handle_receive, shared_from_this()/*this*/,
    boost::asio::placeholders::error,
    boost::asio::placeholders::bytes_transferred)));

}


void tcp_connection::handle_receive(const boost::system::error_code& error, 
std::size_t bytes_transferred)
{

if (!error)
{
    // process the data

    /*  boost::asio::async_read_until remarks

    After a successful async_read_until operation, 
    the streambuf may contain additional data beyond the delimiter.
    An application will typically leave that data in the streambuf for a
    subsequent async_read_until operation to examine.
    */

    /* didn't work      
    std::istream is(&receive_buffer);
    std::string line;
    std::getline(is, line); 
    */


    // clean up incomming buffer but it didn't work 
    receive_buffer.consume(bytes_transferred);  

    receive(); 

}
else if (error != boost::asio::error::operation_aborted)
{
    std::cout << "Client Disconnected\n";

    m_connection_manager.remove(shared_from_this());
}
}

Either using a std::istream and reading from it, such as by std::getline(), or explicitly invoking boost::asio::streambuf::consume(n), will remove data from the input sequence.
If the application is performing either of these and subsequent read_until() operations results in duplicated data in receive_buffer's input sequence, then the duplicated data is likely originating from the remote peer. If the remote peer is writing to the socket and directly using a streambuf's input sequence, then the remote peer needs to explicitly invoke consume() after each successful write operation.

As noted in the documentation, successful read_until() operations may contain additional data beyond the delimiter, including additional delimiters. For instance, if "a@b@" is written to a socket, a read_until() operation using '@' as a delimiter may read and commit "a@b@" to the streambuf's input sequence. However, the operation will indicate that the amount of bytes transferred is that up to and including the first delimiter. Thus, bytes_transferred would be 2 and streambuf.size() would be 4. After 2 bytes have been consumed, the streambuf's input sequence would contain "b@", and a subsequent call to read_until() will return immediately, as the streambuf already contains the delimiter.

Here is a complete example demonstrating streambuf usage for reading and writing, and how the input sequence is consumed:

#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>

// This example is not interested in the handlers, so provide a noop function
// that will be passed to bind to meet the handler concept requirements.
void noop() {}

std::string make_string(boost::asio::streambuf& streambuf)
{
 return {buffers_begin(streambuf.data()), 
         buffers_end(streambuf.data())};
}

int main()
{
  using boost::asio::ip::tcp;
  boost::asio::io_service io_service;

  // Create all I/O objects.
  tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), 0));
  tcp::socket server_socket(io_service);
  tcp::socket client_socket(io_service);

  // Connect client and server sockets.
  acceptor.async_accept(server_socket, boost::bind(&noop));
  client_socket.async_connect(acceptor.local_endpoint(), boost::bind(&noop));
  io_service.run();

  // Write to server.
  boost::asio::streambuf write_buffer;
  std::ostream output(&write_buffer);
  output << "a@"
            "b@";
  write(server_socket, write_buffer.data());
  std::cout << "Wrote: " << make_string(write_buffer) << std::endl;
  assert(write_buffer.size() == 4);  // Data not consumed.

  // Read from the client.
  boost::asio::streambuf read_buffer;

  // Demonstrate consuming via istream.
  {
    std::cout << "Read" << std::endl;
    auto bytes_transferred = read_until(client_socket, read_buffer, '@');
    // Verify that the entire write_buffer (data pass the first delimiter) was
    // read into read_buffer.
    auto initial_size = read_buffer.size();
    assert(initial_size == write_buffer.size());

    // Read from the streambuf.
    std::cout << "Read buffer contains: " << make_string(read_buffer)
              << std::endl;
    std::istream input(&read_buffer);
    std::string line;
    getline(input, line, '@'); // Consumes from the streambuf.
    assert("a" == line); // Note getline discards delimiter.
    std::cout << "Read consumed: " << line << "@" << std::endl;
    assert(read_buffer.size() == initial_size - bytes_transferred);
  }

  // Write an additional message to the server, but only consume 'a@'
  // from write buffer.  The buffer will contain 'b@c@'.
  write_buffer.consume(2);
  std::cout << "Consumed write buffer, it now contains: " <<
                make_string(write_buffer) << std::endl;
  assert(write_buffer.size() == 2);
  output << "c@";
  assert(write_buffer.size() == 4);
  write(server_socket, write_buffer.data());
  std::cout << "Wrote: " << make_string(write_buffer) << std::endl;

  // Demonstrate explicitly consuming via the streambuf.
  {
    std::cout << "Read" << std::endl;
    auto initial_size = read_buffer.size();
    auto bytes_transferred = read_until(client_socket, read_buffer, '@');
    // Verify that the read operation did not attempt to read data from
    // the socket, as the streambuf already contained the delimiter.
    assert(initial_size == read_buffer.size());

    // Read from the streambuf.
    std::cout << "Read buffer contains: " << make_string(read_buffer)
              << std::endl;
    std::string line(
        boost::asio::buffers_begin(read_buffer.data()),
        boost::asio::buffers_begin(read_buffer.data()) + bytes_transferred);
    assert("b@" == line);
    assert(read_buffer.size() == initial_size); // Nothing consumed.
    read_buffer.consume(bytes_transferred); // Explicitly consume.
    std::cout << "Read consumed: " << line << std::endl;
    assert(read_buffer.size() == 0);
  }

  // Read again.
  {
    std::cout << "Read" << std::endl;
    read_until(client_socket, read_buffer, '@');

    // Read from the streambuf.
    std::cout << "Read buffer contains: " << make_string(read_buffer)
              << std::endl;
    std::istream input(&read_buffer);
    std::string line;
    getline(input, line, '@'); // Consumes from the streambuf.
    assert("b" == line); // Note "b" is expected and not "c".
    std::cout << "Read consumed: " << line << "@" << std::endl;
    std::cout << "Read buffer contains: " << make_string(read_buffer)
              << std::endl;
  }
}

Output:

Wrote: a@b@
Read
Read buffer contains: a@b@
Read consumed: a@
Consumed write buffer, it now contains: b@
Wrote: b@c@
Read
Read buffer contains: b@
Read consumed: b@
Read
Read buffer contains: b@c@
Read consumed: b@
Read buffer contains: c@

Recommended topics

Hot tags