3

我使用zmqpp建立了 PUB/SUB 连接，现在我想使用仅标头的 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. 每个部分都将使用 msgpack打包/解包。

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

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 次。

Q1：我是否正确打包/解包数据？

Q2：这是使用 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());
4

1 回答 1

4

我认为调用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。

于 2015-08-22T18:49:08.833 回答