c++ - 如何使用 boost::iostreams:filtering_istream 解压缩 boost::asio::streambuf 中的数据？

Question

我正在尝试解压缩从 boost asio 套接字接收到的 zlib 压缩数据。 (版本 1.64.0)我使用 boost::asio::streambuf 作为接收缓冲区，使用 boost::iostreams::filtering_istream 进行解压。如果我一次发送所有压缩数据，我的代码将按预期工作。但是，如果我将压缩数据作为 5 字节分块发送，代码将无法按预期工作。

我创建一个压缩数据如下:

auto str = prepare_compressed_data("Hello world");

数据是:

78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d
(Length is 19)

我设置filtering_istream解压如下:

// Prepare decompressing istream
boost::asio::streambuf sbuf;
boost::iostreams::filtering_istream is;
is.push(boost::iostreams::zlib_decompressor());
is.push(sbuf);

下面是网络收发仿真代码:

// Network (asio socket send-receive) emulation
auto str_it = str.begin();
auto rest = str.size();
while (rest != 0) {
    auto copy_size = std::min(rest, send_size); // send_size is 5

    // Emulate receive
    //   In actual code, it is `socket.read_some(mbuf)`
    //   and its return value is `copy_size`.

    auto mbuf = sbuf.prepare(copy_size);
    char* p = boost::asio::buffer_cast<char*>(mbuf);
    std::copy(str_it, str_it + copy_size, p);
    sbuf.commit(copy_size);
    hexdump(std::string(boost::asio::buffer_cast<char const*>(sbuf.data()), sbuf.size()));
    std::cout << "sbuf.size():" << sbuf.size() << std::endl;
    { // decompress
        std::cout << "<<< try decompress >>>" << std::endl;
        while (is) {
            std::cout << "  `is` has some data." << std::endl;
            char buf[buf_size + 1] = { '\0' };
            is.read(buf, buf_size);
            std::size_t read_size = is.gcount();
            std::cout << "  read_size:" << read_size << std::endl;
            std::cout << "  decompressed data: " << buf << std::endl;

            // It seems that is.read() consumed 5 bytes data in underlying sbuf
            // even if is.gcount() returned 0.
        }
        std::cout << "<<< decompress loop out >>>" << std::endl;
    }
    rest -= copy_size;
    str_it += copy_size;
}

将 str 的一部分(5 个字节分块)复制到 sbuf。第一次执行内层while循环，while (is) {，is.read(buf, buf_size)被执行，下一行is.gcount( ) 返回0，估计是数据不够解压。但是，is 似乎从 sbuf 中消耗了 5 个字节。然后复制下5个字节，不满足内层while循环的条件。

因为我错过了什么？

完整代码在这里: https://gist.github.com/redboltz/1934952be51e73d558fe8cd7e861d4da

我得到了以下输出:

compressed data: 78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
78 9c f3 48 cd 
sbuf.size():5
<<< try decompress >>>
  `is` has some data.
  read_size:0
  decompressed data: 
<<< decompress loop out >>>
c9 c9 57 28 cf 
sbuf.size():5
<<< try decompress >>>
<<< decompress loop out >>>
c9 c9 57 28 cf 2f ca 49 01 00 
sbuf.size():10
<<< try decompress >>>
<<< decompress loop out >>>
c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
sbuf.size():14
<<< try decompress >>>
<<< decompress loop out >>>

如果我一次复制所有 str，我会得到以下预期的结果。

compressed data: 78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
sbuf.size():19
<<< try decompress >>>
  `is` has some data.
  read_size:11
  decompressed data: Hello world
<<< decompress loop out >>>

Answer 1

当您读取不完整的输入时，过滤流将发出 EOF 信号，并保持信号状态。

解决它的最安全方法是暂时复制 sbuf 数据，并且只有 consume() 成功解压后的数据:

Live On Wandbox

#include <sstream>
#include <string>
#include <iostream>
#include <iomanip>
#include <algorithm>

#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filtering_streambuf.hpp>
#include <boost/iostreams/copy.hpp>
#include <boost/iostreams/filter/zlib.hpp>

#include <boost/asio.hpp>

void hexdump(std::string const& buf) {
    for (std::string::const_iterator it = buf.begin(), end = buf.end();
         it != end;
         ++it) {
        std::cout
            << std::setw(2)
            << std::hex
            << std::setfill('0')
            << (static_cast<int>(*it) & 0xff)
            << ' ';
    }
    std::cout << std::dec << std::endl;
}

std::string prepare_compressed_data(std::string const& str) {
    std::stringstream sender;
    boost::iostreams::filtering_streambuf<boost::iostreams::input> out;
    out.push(boost::iostreams::zlib_compressor());
    out.push(sender);
    sender << str << std::flush;
    std::stringstream compressed;
    boost::iostreams::copy(out, compressed);
    return compressed.str();
}

int main() {
    auto str = prepare_compressed_data("Hello world");
    std::cout << "compressed data: ";
    hexdump(str);

    // Test settings
#if 0
    // send all at conce
    std::size_t const send_size = str.size();
#else
    // send 5 byte chunks
    std::size_t const send_size = 5;
#endif
    std::size_t const buf_size = 256;

    // Prepare decompressing istream
    boost::asio::streambuf sbuf;

    // Network (asio socket send-receive) emulation
    auto str_it = str.begin();
    auto rest = str.size();
    while (rest != 0) {
        auto copy_size = std::min(rest, send_size);

        // Emulate receive
        //   In actual code, it is `socket.read_some(mbuf)`
        //   and its return value is `copy_size`.

        auto mbuf = sbuf.prepare(copy_size);
        char* p = boost::asio::buffer_cast<char*>(mbuf);
        std::copy(str_it, str_it + copy_size, p);
        sbuf.commit(copy_size);

        hexdump(std::string(boost::asio::buffer_cast<char const*>(sbuf.data()), sbuf.size()));
        std::cout << "sbuf.size():" << sbuf.size() << std::endl;
        { // decompress
            std::cout << "<<< try decompress >>>" << std::endl;
            char buf[buf_size] = {};
            std::size_t read_size = 0;

            boost::iostreams::filtering_istream is;
            is.push(boost::iostreams::zlib_decompressor());
            std::stringstream ss(std::string(boost::asio::buffer_cast<char const*>(sbuf.data()), sbuf.size()));
            is.push(ss);

            while (is.read(buf, buf_size) || (read_size = is.gcount())) {
                std::cout << "  `is` has some data." << std::endl;
                std::cout << "  read_size:" << read_size << std::endl;
                (std::cout << "  decompressed data: ").write(buf, read_size) << std::endl;

                // It seems that is.read() consumed 5 bytes data in underlying sbuf
                // even if is.gcount() returned 0.
            }
            std::cout << "<<< decompress loop out >>>" << std::endl;
        }
        rest -= copy_size;
        str_it += copy_size;
    }
}

打印

Start
compressed data: 78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
78 9c f3 48 cd 
sbuf.size():5
<<< try decompress >>>
<<< decompress loop out >>>
78 9c f3 48 cd c9 c9 57 28 cf 
sbuf.size():10
<<< try decompress >>>
<<< decompress loop out >>>
78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 
sbuf.size():15
<<< try decompress >>>
<<< decompress loop out >>>
78 9c f3 48 cd c9 c9 57 28 cf 2f ca 49 01 00 18 ab 04 3d 
sbuf.size():19
<<< try decompress >>>
  `is` has some data.
  read_size:11
  decompressed data: Hello world
<<< decompress loop out >>>

或者，您可以通过清除流标志来“破解它”，但我担心结果没有记录(即充其量不可靠)