c++ - 使用带有 boost::asio 异步操作的自定义 streambuf

Question

我正在使用 boost::asio 编写服务器应用程序。
我正在从用户那里读取已知数量的数据并将其写入具有以下 api 的数据接收器:

class Sink {
    ...
    void write(const char *data, size_t size);
}

数据很大，可以多次调用write(..)来处理一个流。
在我的代码中，我想调用:

boost::asio::async_read(socket, sink_buffer,
    boost::asio::transfer_exactly(size), ...);

是否可以使用自定义 std::streambuf 或 boost::asio::basic_streambuf 包装 Sink 以便它可以处理写入数据它的一部分？

Answer 1

对于作为async_read() 的buffers 参数传递的对象，buffers 参数可以是:

• 需要是满足 MutableBufferSequence 的类型要求。
• 是一个 boost::asio::basic_streambuf 对象。

因此，可以编写一个在读取发生时与 Sink 对象交互的自定义​​类。但是，boost::asio::basic_streambuf 似乎并未设计为用作基类。

如果 Sink::write 只是对底层内存的抽象，请考虑使用类似于 basic_streambuf::prepare() 的方法，其中成员函数返回给定大小的缓冲区句柄。底层内存实现仍将在 mutable_buffer 之后保持抽象。 .例如:

boost::asio::async_read( socket, sink.buffer( size ), ... );

如果Sink::write有业务逻辑，比如根据某些字节的值进行逻辑分支，那么可能需要传递一个中间缓冲区给async_read()。然后，将在 async_read() 的 handler 中使用中间缓冲区调用 Sink::write()。例如:

void handle_read_into_sink( boost::system::error_code error,
                            std::size_t bytes_transferred,
                            boost::asio::ip::tcp::socket& socket,
                            Sink& sink,
                            char* buffer,
                            std::size_t buffer_size,
                            std::size_t bytes_remaining,
                            boost::function< void() > on_finish )
{
  sink.write( buffer, buffer_size );

  bytes_remaining -= bytes_transferred;
  // If there are more bytes remaining, then continue reading.
  if ( bytes_remaining )
  {
    read_into_sink( socket, sink, buffer, buffer_size,
                    bytes_remaining, on_finish );
  }
  // Otherwise, all data has been read.
  else
  {
    on_finish();
  }  
}

void read_into_sink( boost::asio::ip::tcp::socket& socket,
                     Sink& sink,
                     char* buffer,
                     std::size_t buffer_size,
                     std::size_t bytes_remaining,
                     boost::function< void() > on_finish )
{
  boost::asio::async_read(
    socket, boost::asio::buffer( buffer , buffer_size ),
    boost::asio::transfer_exactly( buffer_size ),
    boost::bind( handle_read_into_sink,
                 boost::asio::placeholders::error,
                 boost::asio::placeholders::bytes_transferred,
                 boost::ref( socket ),
                 boost::ref( sink ),
                 buffer,
                 buffer_size,
                 bytes_remaining,
                 on_finish ) );
}

并开始异步读取循环:

read_into_sink( socket, sink, small_buffer, sizeof_small_buffer, 
                total_stream_size, read_handler_callback );

确保根据您所需的逻辑检查和处理错误。