c++ - 使用 MsgPack 通过 ZeroMQ (zmqpp) 发送数据给出 'msgpack::v1::insufficient_bytes' 错误

Question

我使用 zmqpp 建立了一个 PUB/SUB 连接，现在我想使用仅 header 的 C++11 版本的 msgpack-c 将数据从发布者发送到订阅者。

发布者必须发送 2 个 int64_t 数字 -- header_1 和 header_2 -- 后跟一个 std::vector<T> -- data --，其中 T 由 (header_1, header_2) 组合确定。

因为关于如何组合 msgpack 和 zmqpp 的例子并不多，我想出的想法是使用 zmqpp::message::add/add_raw 发送一个由 3 部分组成的消息。每个部分都是使用 msgpack 的 packed/unpacked。

发布者按如下方式打包单个数据部分:

zmqpp::message msg;
int64_t header_1 = 1234567;
msgpack::sbuffer buffer;
msgpack::pack(buffer, header_1);
msg.add(buffer.data(), buffer.size());

然后接收方像这样解压它:

zmqpp::message msg;
subscriberSock.receive(msg);

int64_t header_1;
msgpack::unpacked unpackedData;
// crash !
msgpack::unpack(unpackedData,
                static_cast<const char*>(msg.raw_data(0)),
                msg.size(0));
unpackedData.get().convert(&header_1);

当我运行代码时，订阅者端出现以下错误:

terminate called after throwing an instance of 'msgpack::v1::insufficient_bytes'
  what():  insufficient bytes
Aborted

另外，zmqpp 似乎生成了一个由 5 部分组成的消息，即使我只调用了 add() 3 次。

问题 1:我是否正确打包/解包数据？

问题 2:这是使用 zmqpp 发送 msgpack 缓冲区的正确方法吗？

以下是代码的重要部分:

发布者

zmqpp::socket publisherSock;
/* connection setup stuff ...*/

// forever send data to the subscribers
while(true)
{
    zmqpp::message msg;

    // meta info about the data
    int64_t header_1 = 1234567;
    int64_t header_2 = 89;
    // sample data
    std::vector<double> data;
    data.push_back(1.2);
    data.push_back(3.4);
    data.push_back(5.6);


    {
        msgpack::sbuffer buffer;
        msgpack::pack(buffer, header_1);
        msg.add(buffer.data(), buffer.size());
        cout << "header_1:" << header_1 << endl;  // header_1:1234567
    }

    {
        msgpack::sbuffer buffer;
        msgpack::pack(buffer, header_2);
        msg.add(buffer.data(), buffer.size());
        cout << "header_2:" << header_2 << endl;  // header_2:89
    }

    {
        msgpack::sbuffer buffer;
        msgpack::pack(buffer, data);
        msg.add_raw(buffer.data(), buffer.size());
        std::cout << "data: " << data << std::endl;  // data:[1.2 3.4 5.6]
    }

    std::cout << msg.parts() << " parts" << std::endl;  // prints "5 parts"... why ?
    publisherSock.send(msg);

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}

订阅者

zmqpp::socket subscriberSock;
/* connection setup stuff ...*/

zmqpp::message msg;
subscriberSock.receive(msg);

int64_t header_1;
int64_t header_2;
std::vector<double> data;

std::cout << msg.parts() << " parts" << std::endl;  // prints "5 parts"
{
    // header 1
    {
        msgpack::unpacked unpackedData;
        // crash !
        msgpack::unpack(unpackedData,
                        static_cast<const char*>(msg.raw_data(0)),
                        msg.size(0));
        unpackedData.get().convert(&header_1);
        cout << "header_1:" << header_1 << endl;
    }
    // header 2
    {
        msgpack::unpacked unpackedData;
        msgpack::unpack(unpackedData,
                        static_cast<const char*>(msg.raw_data(1)),
                        msg.size(1));
        unpackedData.get().convert(&header_2);
        cout << "header_2:" << header_2 << endl;
    }
    // data
    {
        msgpack::unpacked unpacked_data;
        msgpack::unpack(unpacked_data,
                        static_cast<const char*>(msg.raw_data(2)),
                        msg.size(2));
        unpacked_data.get().convert(&data);
        std::cout << "data:" << data << std::endl;
    }

}

编辑: 问题已解决:正如@Jens 所指出的，打包/发送数据的正确方法是使用 zmqpp::message::add_raw()

zmqpp::message msg;
int64_t header_1 = 1234567;
msgpack::sbuffer buffer;
msgpack::pack(buffer, header_1);
msg.add_raw(buffer.data(), buffer.size());

Answer 1

我认为调用 msg.add(buffer.data(), buffer.size()不要添加 buffer.size() 的数组字节，但调用 message::add(Type const& part, Args &&...args) , 哪个

1. msg << buffer.data() ，它可能调用 message::operator<<(bool)因为指针转换为 bool
2. add(buffer.size())然后调用 msg << buffer.size() , 它增加了一个 size_t值作为下一部分。

查看 zmqpp::message 类，使用 message::add_raw 应该可以解决问题。

PS:这一切都没有任何保证，因为我从未使用过zmqpp或msgpack。