Consume only part of data in boost::asio basic_stream_socket::async_read_some handler

Question

I am new into boost::asio so my question maight be dumb - sorry if it is such. I am writing asynchronous server application with keepalive (multiple requests may be sent on single connection).

Connection handling routine is simple:

In a loop:

• schedule read request with socket->async_read_some(buffer, handler)

• from handler schedule write response with async_write.

The problem I am facing is that when handler passed to async_read_some is called by on of io_service threads, buffers may actually contain more data than single request (e.g. part of next request sent by client).

I do not want to (and cannot if it is only part of request) handle this remaining bytes at the moment. I would like to do it after handling previous request is finished.

It would be easy to address this if I had the possiblity to reinject unnecessary remainging data back to the socket. So it is handled on next async_read_some call.

Is there such possiblity in boost::asio or do I have to store the remaining data somewhere aside, and handle it myself with extra code.

Answer 1

I think what you are looking for is asio::streambuf.

Basically, you can inspect your seeded streambuf as a char*, read as much as you see fit, and then inform how much was actually processed by consume(amount).

Working code-example to parse HTTP-header as a client:

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

namespace asio = boost::asio;

std::string LINE_TERMINATION = "\r\n";

class Connection {
  asio::streambuf _buf;
  asio::ip::tcp::socket _socket;
public:

  Connection(asio::io_service& ioSvc, asio::ip::tcp::endpoint server)
    : _socket(ioSvc)
  {
    _socket.connect(server);
    _socket.send(boost::asio::buffer("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"));
    readMore();
  }

  void readMore() {
    // Allocate 13 bytes space on the end of the buffer. Evil prime number to prove algorithm works.
    asio::streambuf::mutable_buffers_type buf = _buf.prepare(13);

    // Perform read
    _socket.async_read_some(buf,  boost::bind(
          &Connection::onRead, this,
          asio::placeholders::bytes_transferred, asio::placeholders::error
    ));
  }

  void onRead(size_t read, const boost::system::error_code& ec) {
    if ((!ec) && (read > 0)) {
      // Mark to buffer how much was actually read
      _buf.commit(read);

      // Use some ugly parsing to extract whole lines.
      const char* data_ = boost::asio::buffer_cast<const char*>(_buf.data());
      std::string data(data_, _buf.size());
      size_t start = 0;
      size_t end = data.find(LINE_TERMINATION, start);
      while (end < data.size()) {
        std::cout << "LINE:" << data.substr(start, end-start) << std::endl;
        start = end + LINE_TERMINATION.size();
        end = data.find(LINE_TERMINATION, start);
      }
      _buf.consume(start);

      // Wait for next data
      readMore();
    }
  }
};

int main(int, char**) {
  asio::io_service ioSvc;

  // Setup a connection and run 
  asio::ip::address localhost = asio::ip::address::from_string("127.0.0.1");
  Connection c(ioSvc, asio::ip::tcp::endpoint(localhost, 80));

  ioSvc.run();
}

Answer 2

If you know the messages are going to be of a fixed length, you can do something like the following:

//-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
void
Connection::readMore()
{
    if (m_connected)
    {
        // Asynchronously read some data from the connection into the buffer.
        // Using shared_from_this() will prevent this Connection object from
        // being destroyed while data is being read.
        boost::asio::async_read(
            m_socket, 
            boost::asio::buffer(
                m_readMessage.getData(), 
                MessageBuffer::MESSAGE_LENGTH
            ),
            boost::bind(
                &Connection::messageBytesRead, 
                shared_from_this(), 
                boost::asio::placeholders::error, 
                boost::asio::placeholders::bytes_transferred
            ),
            boost::bind(
                &Connection::handleRead, 
                shared_from_this(),
                boost::asio::placeholders::error
            )
        );
    }
}

//-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
std::size_t
Connection::messageBytesRead(const boost::system::error_code& _errorCode, 
                             std::size_t _bytesRead)
{
    return MessageBuffer::MESSAGE_LENGTH - _bytesRead;
}

//-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
void
Connection::handleRead(const boost::system::error_code& _errorCode)
{
    if (!_errorCode)
    {
        /// Do something with the populated m_readMessage here.
        readMore();
    }
    else
    {
        disconnect();
    }
}

The messageBytesRead callback will indicate to boost::asio::async_read when a complete message has been read. This snippet was pulled from an existing Connection object from running code, so I know it works...

Answer 3

One way of tackling this when using a reliable and ordered transport like TCP is to:

1. Write a header of known size, containing the size of the rest of the message
2. Write the rest of the message

And on the receiving end:

1. Read just enough bytes to get the header
2. Read the rest of the message and no more