我使用 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());
最佳答案
我认为调用 msg.add(buffer.data(), buffer.size()
不要添加 buffer.size()
的数组字节,但调用 message::add(Type const& part, Args &&...args)
, 哪个
-
msg << buffer.data()
,它可能调用message::operator<<(bool)
因为指针转换为 bool -
add(buffer.size())
然后调用msg << buffer.size()
, 它增加了一个size_t
值作为下一部分。
查看 zmqpp::message 类,使用 message::add_raw 应该可以解决问题。
PS:这一切都没有任何保证,因为我从未使用过zmqpp或msgpack。
关于c++ - 使用 MsgPack 通过 ZeroMQ (zmqpp) 发送数据给出 'msgpack::v1::insufficient_bytes' 错误,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32158225/