generic-library/cppzmq
1
0
Fork 0
mirror of https://github.com/zeromq/cppzmq.git synced 2025-10-16 18:56:56 +02:00
Code Issues Projects Releases Wiki Activity

Problem: send/recv functions lack type-safety

Browse Source 
Solution: Add functions taking buffers and enum class flags
This commit is contained in:
Gudmundur Adalsteinsson
2019-04-22 21:15:51 +00:00
parent 35ba5bea3b
commit 3d4be814e8
8 changed files with 866 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/CMakeLists.txt
View File

@@ -17,6 +17,7 @@ find_package(Threads)
add_executable(
unit_tests
buffer.cpp
message.cpp
context.cpp
socket.cpp

22
tests/active_poller.cpp
View File

@@ -157,7 +157,7 @@ TEST_CASE("poll basic", "[active_poller]")
{
server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::active_poller_t active_poller;
bool message_received = false;
@@ -184,7 +184,7 @@ TEST_CASE("client server", "[active_poller]")
zmq::active_poller_t::handler_t handler = [&](short e) {
if (0 != (e & ZMQ_POLLIN)) {
zmq::message_t zmq_msg;
CHECK_NOTHROW(s.server.recv(&zmq_msg)); // get message
CHECK_NOTHROW(s.server.recv(zmq_msg)); // get message
std::string recv_msg(zmq_msg.data<char>(), zmq_msg.size());
CHECK(send_msg == recv_msg);
} else if (0 != (e & ~ZMQ_POLLOUT)) {
@@ -197,7 +197,7 @@ TEST_CASE("client server", "[active_poller]")
CHECK_NOTHROW(active_poller.add(s.server, ZMQ_POLLIN, handler));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{send_msg}));
CHECK_NOTHROW(s.client.send(zmq::message_t{send_msg}, zmq::send_flags::none));
CHECK(1 == active_poller.wait(std::chrono::milliseconds{-1}));
CHECK(events == ZMQ_POLLIN);
@@ -236,7 +236,7 @@ TEST_CASE("remove invalid socket throws", "[active_poller]")
TEST_CASE("wait on added empty handler", "[active_poller]")
{
server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::active_poller_t active_poller;
zmq::active_poller_t::handler_t handler;
CHECK_NOTHROW(active_poller.add(s.server, ZMQ_POLLIN, handler));
@@ -291,7 +291,7 @@ TEST_CASE("poll client server", "[active_poller]")
CHECK_NOTHROW(active_poller.add(s.server, ZMQ_POLLIN, s.handler));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify events
CHECK_NOTHROW(active_poller.wait(std::chrono::milliseconds{500}));
@@ -316,7 +316,7 @@ TEST_CASE("wait one return", "[active_poller]")
active_poller.add(s.server, ZMQ_POLLIN, [&count](short) { ++count; }));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify events
CHECK(1 == active_poller.wait(std::chrono::milliseconds{500}));
@@ -326,7 +326,7 @@ TEST_CASE("wait one return", "[active_poller]")
TEST_CASE("wait on move constructed active_poller", "[active_poller]")
{
server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::active_poller_t a;
zmq::active_poller_t::handler_t handler;
CHECK_NOTHROW(a.add(s.server, ZMQ_POLLIN, handler));
@@ -340,7 +340,7 @@ TEST_CASE("wait on move constructed active_poller", "[active_poller]")
TEST_CASE("wait on move assigned active_poller", "[active_poller]")
{
server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::active_poller_t a;
zmq::active_poller_t::handler_t handler;
CHECK_NOTHROW(a.add(s.server, ZMQ_POLLIN, handler));
@@ -361,14 +361,14 @@ TEST_CASE("received on move constructed active_poller", "[active_poller]")
zmq::active_poller_t a;
CHECK_NOTHROW(a.add(s.server, ZMQ_POLLIN, [&count](short) { ++count; }));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify it is received
CHECK(1 == a.wait(std::chrono::milliseconds{500}));
CHECK(1u == count);
// Move construct active_poller b
zmq::active_poller_t b{std::move(a)};
// client sends message again
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify it is received
CHECK(1 == b.wait(std::chrono::milliseconds{500}));
CHECK(2u == count);
@@ -399,7 +399,7 @@ TEST_CASE("remove from handler", "[active_poller]")
CHECK(ITER_NO == active_poller.size());
// Clients send messages
for (auto &s : setup_list) {
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
}
// Wait for all servers to receive a message

253
tests/buffer.cpp Normal file
View File

@@ -0,0 +1,253 @@
#include <catch.hpp>
#include <zmq.hpp>
#ifdef ZMQ_CPP17
static_assert(std::is_nothrow_swappable_v<zmq::const_buffer>);
static_assert(std::is_nothrow_swappable_v<zmq::mutable_buffer>);
static_assert(std::is_trivially_copyable_v<zmq::const_buffer>);
static_assert(std::is_trivially_copyable_v<zmq::mutable_buffer>);
#endif
#ifdef ZMQ_CPP11
using BT = int16_t;
TEST_CASE("buffer default ctor", "[buffer]")
{
zmq::mutable_buffer mb;
zmq::const_buffer cb;
CHECK(mb.size() == 0);
CHECK(mb.data() == nullptr);
CHECK(cb.size() == 0);
CHECK(cb.data() == nullptr);
}
TEST_CASE("buffer data ctor", "[buffer]")
{
std::vector<BT> v(10);
zmq::const_buffer cb(v.data(), v.size() * sizeof(BT));
CHECK(cb.size() == v.size() * sizeof(BT));
CHECK(cb.data() == v.data());
zmq::mutable_buffer mb(v.data(), v.size() * sizeof(BT));
CHECK(mb.size() == v.size() * sizeof(BT));
CHECK(mb.data() == v.data());
zmq::const_buffer from_mut = mb;
CHECK(mb.size() == from_mut.size());
CHECK(mb.data() == from_mut.data());
const auto cmb = mb;
static_assert(std::is_same<decltype(cmb.data()), void*>::value, "");
}
TEST_CASE("const_buffer operator+", "[buffer]")
{
std::vector<BT> v(10);
zmq::const_buffer cb(v.data(), v.size() * sizeof(BT));
const size_t shift = 4;
auto shifted = cb + shift;
CHECK(shifted.size() == v.size() * sizeof(BT) - shift);
CHECK(shifted.data() == v.data() + shift / sizeof(BT));
auto shifted2 = shift + cb;
CHECK(shifted.size() == shifted2.size());
CHECK(shifted.data() == shifted2.data());
auto cbinp = cb;
cbinp += shift;
CHECK(shifted.size() == cbinp.size());
CHECK(shifted.data() == cbinp.data());
}
TEST_CASE("mutable_buffer operator+", "[buffer]")
{
std::vector<BT> v(10);
zmq::mutable_buffer mb(v.data(), v.size() * sizeof(BT));
const size_t shift = 4;
auto shifted = mb + shift;
CHECK(shifted.size() == v.size() * sizeof(BT) - shift);
CHECK(shifted.data() == v.data() + shift / sizeof(BT));
auto shifted2 = shift + mb;
CHECK(shifted.size() == shifted2.size());
CHECK(shifted.data() == shifted2.data());
auto mbinp = mb;
mbinp += shift;
CHECK(shifted.size() == mbinp.size());
CHECK(shifted.data() == mbinp.data());
}
TEST_CASE("mutable_buffer creation basic", "[buffer]")
{
std::vector<BT> v(10);
zmq::mutable_buffer mb(v.data(), v.size() * sizeof(BT));
zmq::mutable_buffer mb2 = zmq::buffer(v.data(), v.size() * sizeof(BT));
CHECK(mb.data() == mb2.data());
CHECK(mb.size() == mb2.size());
zmq::mutable_buffer mb3 = zmq::buffer(mb);
CHECK(mb.data() == mb3.data());
CHECK(mb.size() == mb3.size());
zmq::mutable_buffer mb4 = zmq::buffer(mb, 10 * v.size() * sizeof(BT));
CHECK(mb.data() == mb4.data());
CHECK(mb.size() == mb4.size());
zmq::mutable_buffer mb5 = zmq::buffer(mb, 4);
CHECK(mb.data() == mb5.data());
CHECK(4 == mb5.size());
}
TEST_CASE("const_buffer creation basic", "[buffer]")
{
const std::vector<BT> v(10);
zmq::const_buffer cb(v.data(), v.size() * sizeof(BT));
zmq::const_buffer cb2 = zmq::buffer(v.data(), v.size() * sizeof(BT));
CHECK(cb.data() == cb2.data());
CHECK(cb.size() == cb2.size());
zmq::const_buffer cb3 = zmq::buffer(cb);
CHECK(cb.data() == cb3.data());
CHECK(cb.size() == cb3.size());
zmq::const_buffer cb4 = zmq::buffer(cb, 10 * v.size() * sizeof(BT));
CHECK(cb.data() == cb4.data());
CHECK(cb.size() == cb4.size());
zmq::const_buffer cb5 = zmq::buffer(cb, 4);
CHECK(cb.data() == cb5.data());
CHECK(4 == cb5.size());
}
TEST_CASE("mutable_buffer creation C array", "[buffer]")
{
BT d[10] = {};
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == 10 * sizeof(BT));
CHECK(b.data() == static_cast<BT*>(d));
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == static_cast<BT*>(d));
}
TEST_CASE("const_buffer creation C array", "[buffer]")
{
const BT d[10] = {};
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == 10 * sizeof(BT));
CHECK(b.data() == static_cast<const BT*>(d));
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == static_cast<const BT*>(d));
}
TEST_CASE("mutable_buffer creation array", "[buffer]")
{
std::array<BT, 10> d = {};
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(BT));
CHECK(b.data() == d.data());
zmq::mutable_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
TEST_CASE("const_buffer creation array", "[buffer]")
{
const std::array<BT, 10> d = {};
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(BT));
CHECK(b.data() == d.data());
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
TEST_CASE("const_buffer creation array 2", "[buffer]")
{
std::array<const BT, 10> d = {{}};
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(BT));
CHECK(b.data() == d.data());
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
TEST_CASE("mutable_buffer creation vector", "[buffer]")
{
std::vector<BT> d(10);
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(BT));
CHECK(b.data() == d.data());
zmq::mutable_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
d.clear();
b = zmq::buffer(d);
CHECK(b.size() == 0);
CHECK(b.data() == nullptr);
}
TEST_CASE("const_buffer creation vector", "[buffer]")
{
std::vector<BT> d(10);
zmq::const_buffer b = zmq::buffer(static_cast<const std::vector<BT>&>(d));
CHECK(b.size() == d.size() * sizeof(BT));
CHECK(b.data() == d.data());
zmq::const_buffer b2 = zmq::buffer(static_cast<const std::vector<BT>&>(d), 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
d.clear();
b = zmq::buffer(static_cast<const std::vector<BT>&>(d));
CHECK(b.size() == 0);
CHECK(b.data() == nullptr);
}
TEST_CASE("const_buffer creation string", "[buffer]")
{
const std::wstring d(10, L'a');
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(wchar_t));
CHECK(b.data() == d.data());
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
TEST_CASE("mutable_buffer creation string", "[buffer]")
{
std::wstring d(10, L'a');
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(wchar_t));
CHECK(b.data() == d.data());
zmq::mutable_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
#ifdef ZMQ_CPP17
TEST_CASE("const_buffer creation string_view", "[buffer]")
{
std::wstring dstr(10, L'a');
std::wstring_view d = dstr;
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(wchar_t));
CHECK(b.data() == d.data());
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == d.data());
}
#endif
TEST_CASE("buffer of structs", "[buffer]")
{
struct some_pod
{
int64_t val;
char arr[8];
};
struct some_non_pod
{
int64_t val;
char arr[8];
std::vector<int> s; // not trivially copyable
};
static_assert(zmq::detail::is_pod_like<some_pod>::value, "");
static_assert(!zmq::detail::is_pod_like<some_non_pod>::value, "");
std::array<some_pod, 1> d;
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == d.size() * sizeof(some_pod));
CHECK(b.data() == d.data());
}
#endif

12
tests/poller.cpp
View File

@@ -142,7 +142,7 @@ TEST_CASE("poller poll basic", "[poller]")
{
common_server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::poller_t<int> poller;
std::vector<zmq_poller_event_t> events{1};
@@ -220,7 +220,7 @@ TEST_CASE("poller poll client server", "[poller]")
CHECK_NOTHROW(poller.add(s.server, ZMQ_POLLIN, s.server));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify events
std::vector<zmq_poller_event_t> events(1);
@@ -243,7 +243,7 @@ TEST_CASE("poller wait one return", "[poller]")
CHECK_NOTHROW(poller.add(s.server, ZMQ_POLLIN, nullptr));
// client sends message
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
// wait for message and verify events
std::vector<zmq_poller_event_t> events(1);
@@ -253,7 +253,7 @@ TEST_CASE("poller wait one return", "[poller]")
TEST_CASE("poller wait on move constructed", "[poller]")
{
common_server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::poller_t<> a;
CHECK_NOTHROW(a.add(s.server, ZMQ_POLLIN, nullptr));
zmq::poller_t<> b{std::move(a)};
@@ -266,7 +266,7 @@ TEST_CASE("poller wait on move constructed", "[poller]")
TEST_CASE("poller wait on move assigned", "[poller]")
{
common_server_client_setup s;
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
zmq::poller_t<> a;
CHECK_NOTHROW(a.add(s.server, ZMQ_POLLIN, nullptr));
zmq::poller_t<> b;
@@ -293,7 +293,7 @@ TEST_CASE("poller remove from handler", "[poller]")
}
// Clients send messages
for (auto &s : setup_list) {
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}));
CHECK_NOTHROW(s.client.send(zmq::message_t{"Hi"}, zmq::send_flags::none));
}
// Wait for all servers to receive a message

149
tests/socket.cpp
View File

@@ -46,6 +46,14 @@ TEST_CASE("socket swap", "[socket]")
using std::swap;
swap(socket1, socket2);
}
TEST_CASE("rass", "[socket]")
{
zmq::context_t ctx;
zmq::socket_t sock(ctx, zmq::socket_type::push);
sock.bind("inproc://test");
const std::string m = "Hello, world";
sock.send(zmq::buffer(m), zmq::send_flags::dontwait);
}
#endif
TEST_CASE("socket sends and receives const buffer", "[socket]")
@@ -55,13 +63,150 @@ TEST_CASE("socket sends and receives const buffer", "[socket]")
zmq::socket_t receiver(context, ZMQ_PAIR);
receiver.bind("inproc://test");
sender.connect("inproc://test");
CHECK(2 == sender.send("Hi", 2));
const char* str = "Hi";
#ifdef ZMQ_CPP11
CHECK(2 == sender.send(zmq::buffer(str, 2)).size);
char buf[2];
const auto res = receiver.recv(zmq::buffer(buf));
CHECK(!res.truncated());
CHECK(2 == res.size);
#else
CHECK(2 == sender.send(str, 2));
char buf[2];
CHECK(2 == receiver.recv(buf, 2));
CHECK(0 == memcmp(buf, "Hi", 2));
#endif
CHECK(0 == memcmp(buf, str, 2));
}
#ifdef ZMQ_CPP11
TEST_CASE("socket send none sndmore", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::router);
s.bind("inproc://test");
std::vector<char> buf(4);
auto res = s.send(zmq::buffer(buf), zmq::send_flags::sndmore);
CHECK(res.size == buf.size());
CHECK(res.success);
res = s.send(zmq::buffer(buf));
CHECK(res.size == buf.size());
CHECK(res.success);
}
TEST_CASE("socket send dontwait", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::push);
s.bind("inproc://test");
std::vector<char> buf(4);
auto res = s.send(zmq::buffer(buf), zmq::send_flags::dontwait);
CHECK(!res.success);
CHECK(res.size == 0);
res = s.send(zmq::buffer(buf),
zmq::send_flags::dontwait | zmq::send_flags::sndmore);
CHECK(!res.success);
CHECK(res.size == 0);
zmq::message_t msg;
auto resm = s.send(msg, zmq::send_flags::dontwait);
CHECK(!resm.success);
CHECK(resm.size == 0);
CHECK(msg.size() == 0);
}
TEST_CASE("socket send exception", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::pull);
s.bind("inproc://test");
std::vector<char> buf(4);
CHECK_THROWS_AS(s.send(zmq::buffer(buf)), const zmq::error_t &);
}
TEST_CASE("socket recv none", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::pair);
zmq::socket_t s2(context, zmq::socket_type::pair);
s2.bind("inproc://test");
s.connect("inproc://test");
std::vector<char> sbuf(4);
const auto res_send = s2.send(zmq::buffer(sbuf));
CHECK(res_send.success);
std::vector<char> buf(2);
const auto res = s.recv(zmq::buffer(buf));
CHECK(res.success);
CHECK(res.truncated());
CHECK(res.untruncated_size == sbuf.size());
CHECK(res.size == buf.size());
const auto res_send2 = s2.send(zmq::buffer(sbuf));
CHECK(res_send2.success);
std::vector<char> buf2(10);
const auto res2 = s.recv(zmq::buffer(buf2));
CHECK(res2.success);
CHECK(!res2.truncated());
CHECK(res2.untruncated_size == sbuf.size());
CHECK(res2.size == sbuf.size());
}
TEST_CASE("socket send recv message_t", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::pair);
zmq::socket_t s2(context, zmq::socket_type::pair);
s2.bind("inproc://test");
s.connect("inproc://test");
zmq::message_t smsg(size_t{10});
const auto res_send = s2.send(smsg, zmq::send_flags::none);
CHECK(res_send.success);
CHECK(res_send.size == 10);
CHECK(smsg.size() == 0);
zmq::message_t rmsg;
const auto res = s.recv(rmsg);
CHECK(res.success);
CHECK(res.size == 10);
CHECK(rmsg.size() == res.size);
}
TEST_CASE("socket recv dontwait", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::pull);
s.bind("inproc://test");
std::vector<char> buf(4);
constexpr auto flags = zmq::recv_flags::none | zmq::recv_flags::dontwait;
auto res = s.recv(zmq::buffer(buf), flags);
CHECK(!res.success);
CHECK(res.size == 0);
zmq::message_t msg;
auto resm = s.recv(msg, flags);
CHECK(!resm.success);
CHECK(resm.size == 0);
CHECK(msg.size() == 0);
}
TEST_CASE("socket recv exception", "[socket]")
{
zmq::context_t context;
zmq::socket_t s(context, zmq::socket_type::push);
s.bind("inproc://test");
std::vector<char> buf(4);
CHECK_THROWS_AS(s.recv(zmq::buffer(buf)), const zmq::error_t &);
}
TEST_CASE("socket proxy", "[socket]")
{
zmq::context_t context;