Lavorare con boost :: asio :: streambuf

Question

Alla ricerca di una spinta :: asio (e con se stesso boost) ha deciso di scrivere server asincrono. Per memorizzare i dati in arrivo, utilizzo boost :: asio :: streambuf. Qui ho un problema. Quando ricevo un secondo messaggio dal client e successivamente vedo che nel buffer contiene dati provenienti da messaggi precedenti. Anche se chiamo il metodo Consume nel buffer di input. Cosa c'è che non va?

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()); 
} 
} 

Answer 1

Sia utilizzando un std::istream e la lettura da esso, come mediante std::getline(), o invocando esplicitamente boost::asio::streambuf::consume(n), rimuove i dati dalla sequenza di ingresso.
Se l'applicazione sta eseguendo una di queste operazioni e successive operazioni read_until(), i risultati sono duplicati nella sequenza di input di receive_buffer, quindi è probabile che i dati duplicati provengano dal peer remoto. Se il peer remoto sta scrivendo sul socket e utilizzando direttamente la sequenza di input di streambuf, il peer remoto deve richiamare esplicitamente consume() dopo ogni operazione di scrittura riuscita.

---

Come indicato nella documentazione, di successo read_until() operazioni possono contenere dati aggiuntivi oltre il delimitatore, tra cui delimitatori aggiuntivi. Ad esempio, se si scrive "a@b@" su un socket, l'operazione read_until() utilizzando '@' come delimitatore può leggere e memorizzare "a@b@" nella sequenza di input dello streambuf. Tuttavia, l'operazione indicherà che la quantità di byte trasferiti è compresa fino al primo delimitatore. Pertanto, bytes_transferred sarebbe 2 e streambuf.size() sarebbe 4. Dopo che sono stati consumati i byte 2, la sequenza di input dello streambuf conterrà "b@" e una chiamata successiva a read_until() verrà restituita immediatamente, poiché lo streambuf contiene già il delimitatore.

Ecco un esempio completo demonstratingstreambuf utilizzo per la lettura e la scrittura, e come la sequenza di ingresso è consumato:

#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; 
    } 
} 

uscita:

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@