Commit 16994320 authored by Jakob Otto's avatar Jakob Otto

Add another test for udp_datagram_socket

parent eae38ecc
...@@ -276,8 +276,10 @@ private: ...@@ -276,8 +276,10 @@ private:
} }
packet_queue_.pop_front(); packet_queue_.pop_front();
}; };
// TODO: Implement this with span<vector*>
// Write buffers from the write_queue_ for as long as possible. // Write buffers from the write_queue_ for as long as possible.
while (!packet_queue_.empty()) { /*while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front(); auto& packet = packet_queue_.front();
auto write_ret = write(handle_, make_span(packet.bytes), auto write_ret = write(handle_, make_span(packet.bytes),
packet.destination); packet.destination);
...@@ -295,7 +297,7 @@ private: ...@@ -295,7 +297,7 @@ private:
} }
return true; return true;
} }
} }*/
return false; return false;
} }
......
...@@ -66,7 +66,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -66,7 +66,7 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
/// @post either the result is a `sec` or a positive (non-zero) integer /// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10` /// @pre `bufs.size() < 10`
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>> bufs, variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep); ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`. /// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
......
...@@ -111,11 +111,11 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -111,11 +111,11 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
} }
} }
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>> bufs, variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep) { ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10); CAF_ASSERT(bufs.size() < 10);
auto convert = [](std::vector<byte> buf) { auto convert = [](std::vector<byte>* buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()}; return iovec{const_cast<byte*>(buf->data()), buf->size()};
}; };
sockaddr_storage addr = {}; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
......
...@@ -39,7 +39,7 @@ namespace { ...@@ -39,7 +39,7 @@ namespace {
constexpr string_view hello_test = "Hello test!"; constexpr string_view hello_test = "Hello test!";
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : host_fixture() { fixture() : host_fixture(), buf(1024) {
addresses = local_addresses("localhost"); addresses = local_addresses("localhost");
CAF_CHECK(!addresses.empty()); CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0); ep = ip_endpoint(*addresses.begin(), 0);
...@@ -61,14 +61,48 @@ struct fixture : host_fixture { ...@@ -61,14 +61,48 @@ struct fixture : host_fixture {
ip_endpoint ep; ip_endpoint ep;
udp_datagram_socket send_socket; udp_datagram_socket send_socket;
udp_datagram_socket receive_socket; udp_datagram_socket receive_socket;
std::vector<byte> buf;
};
error read_from_socket(udp_datagram_socket sock, std::vector<byte>& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(sock, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
return make_error(get<sec>(read_ret), "read failed");
}
if (++receive_attempts > 100)
return make_error(sec::runtime_error,
"too many unavailable_or_would_blocks");
} while (read_ret.index() != 0);
return none;
}
struct header {
header(size_t payload_size) : payload_size(payload_size) {
// nop
}
header() : header(0) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, header& x) {
return f(meta::type_name("header"), x.payload_size);
}
size_t payload_size;
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture) CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) { CAF_TEST(read / write using span<byte>) {
std::vector<byte> buf(1024);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true)) if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err); CAF_FAIL("setting socket to nonblocking failed: " << err);
CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)), CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)),
...@@ -76,21 +110,35 @@ CAF_TEST(send and receive) { ...@@ -76,21 +110,35 @@ CAF_TEST(send and receive) {
CAF_MESSAGE("sending data to " << to_string(ep)); CAF_MESSAGE("sending data to " << to_string(ep));
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep), CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep),
hello_test.size()); hello_test.size());
int receive_attempts = 0; CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(receive_socket, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
CAF_CHECK_EQUAL(read_res->first, hello_test.size());
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
CAF_FAIL("read returned an error: " << sys.render(get<sec>(read_ret)));
}
if (++receive_attempts > 100)
CAF_FAIL("unable to receive data from the UDP socket");
} while (read_ret.index() != 0);
string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()}; string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(received, hello_test); CAF_CHECK_EQUAL(received, hello_test);
} }
CAF_TEST(read / write using span<std::vector<byte>>) {
// generate header and payload in separate buffers
header hdr{hello_test.size()};
std::vector<byte> hdr_buf;
serializer_impl<std::vector<byte>> sink(sys, hdr_buf);
if (auto err = sink(hdr))
CAF_FAIL("serializing payload failed" << sys.render(err));
auto bytes = as_bytes(make_span(hello_test));
std::vector<byte> payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size();
std::vector<std::vector<byte>*> bufs{&hdr_buf, &payload_buf};
CAF_CHECK_EQUAL(write(send_socket, make_span(bufs), ep), packet_size);
// receive both as one single packet.
buf.resize(packet_size);
CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
CAF_CHECK_EQUAL(buf.size(), packet_size);
binary_deserializer source(nullptr, buf);
header recv_hdr;
source(recv_hdr);
CAF_CHECK_EQUAL(hdr.payload_size, recv_hdr.payload_size);
string_view received{reinterpret_cast<const char*>(buf.data())
+ sizeof(header),
buf.size() - sizeof(header)};
CAF_CHECK_EQUAL(received, hello_test);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment