Unverified Commit 38e6ccdf authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #27

Fix udp datagram socket
parents 984e2dc5 e0183f4c
......@@ -27,6 +27,7 @@ namespace caf {
namespace detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst) {
memset(&dst, 0, sizeof(sockaddr_storage));
if (src.address().embeds_v4()) {
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&dst);
sockaddr4->sin_family = AF_INET;
......
......@@ -100,9 +100,10 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
<< CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes)
<< " bytes");
ip_endpoint ep;
// TODO: how to properly pass error further?
if (auto err = detail::convert(addr, ep))
return sec::runtime_error;
if (auto err = detail::convert(addr, ep)) {
CAF_ASSERT(err.category() == atom("system"));
return static_cast<sec>(err.code());
}
return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
} else {
return get<sec>(ret);
......
......@@ -37,22 +37,21 @@ namespace {
struct fixture : host_fixture {
fixture() : host_fixture() {
memset(&sockaddr6_src, 0, sizeof(sockaddr_in6));
memset(&sockaddr6_dst, 0, sizeof(sockaddr_in6));
sockaddr6_src.sin6_family = AF_INET6;
sockaddr6_src.sin6_port = htons(23);
sockaddr6_src.sin6_addr = in6addr_loopback;
memset(&sockaddr4_src, 0, sizeof(sockaddr_in));
memset(&sockaddr4_dst, 0, sizeof(sockaddr_in));
sockaddr4_src.sin_family = AF_INET;
sockaddr4_src.sin_port = htons(23);
sockaddr4_src.sin_addr.s_addr = INADDR_LOOPBACK;
memset(&sockaddr4_src, 0, sizeof(sockaddr_storage));
memset(&sockaddr6_src, 0, sizeof(sockaddr_storage));
auto sockaddr6_ptr = reinterpret_cast<sockaddr_in6*>(&sockaddr6_src);
sockaddr6_ptr->sin6_family = AF_INET6;
sockaddr6_ptr->sin6_port = htons(23);
sockaddr6_ptr->sin6_addr = in6addr_loopback;
auto sockaddr4_ptr = reinterpret_cast<sockaddr_in*>(&sockaddr4_src);
sockaddr4_ptr->sin_family = AF_INET;
sockaddr4_ptr->sin_port = htons(23);
sockaddr4_ptr->sin_addr.s_addr = INADDR_LOOPBACK;
}
sockaddr_in6 sockaddr6_src;
sockaddr_in6 sockaddr6_dst;
sockaddr_in sockaddr4_src;
sockaddr_in sockaddr4_dst;
sockaddr_storage sockaddr6_src;
sockaddr_storage sockaddr4_src;
sockaddr_storage dst;
ip_endpoint ep_src;
ip_endpoint ep_dst;
};
......@@ -62,51 +61,43 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(convert_ip_endpoint_tests, fixture)
CAF_TEST(sockaddr_in6 roundtrip) {
ip_endpoint ep;
ip_endpoint tmp;
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr6_src),
ep),
none);
CAF_CHECK_EQUAL(convert(sockaddr6_src, tmp), none);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr6_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr6_src, &sockaddr6_dst, sizeof(sockaddr_in6)),
0);
convert(tmp, dst);
CAF_CHECK_EQUAL(memcmp(&sockaddr6_src, &dst, sizeof(sockaddr_storage)), 0);
}
CAF_TEST(ipv6_endpoint roundtrip) {
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
sockaddr_storage tmp = {};
if (auto err = detail::parse("[::1]:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
convert(ep_src, addr);
convert(ep_src, tmp);
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(convert(tmp, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST(sockaddr_in4 roundtrip) {
ip_endpoint ep;
ip_endpoint tmp;
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr4_src),
ep),
none);
CAF_CHECK_EQUAL(convert(sockaddr4_src, tmp), none);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr4_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr4_src, &sockaddr4_dst, sizeof(sockaddr_in)),
0);
convert(tmp, dst);
CAF_CHECK_EQUAL(memcmp(&sockaddr4_src, &dst, sizeof(sockaddr_storage)), 0);
}
CAF_TEST(ipv4_endpoint roundtrip) {
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
sockaddr_storage tmp = {};
if (auto err = detail::parse("127.0.0.1:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
convert(ep_src, addr);
convert(ep_src, tmp);
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(convert(tmp, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST_FIXTURE_SCOPE_END();
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -30,69 +30,38 @@
#include "caf/ip_endpoint.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
using namespace caf;
using namespace caf::net;
using namespace caf::net::ip;
namespace {
constexpr string_view hello_test = "Hello test!";
struct fixture : host_fixture {
fixture()
: host_fixture(),
v4_local{make_ipv4_address(127, 0, 0, 1)},
v6_local{{0}, {0x1}},
// TODO: use local_addresses() when merged
addrs{net::ip::resolve("localhost")} {
}
bool contains(ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
}
void test_send_receive(ip_address addr) {
std::vector<byte> buf(1024);
ip_endpoint ep(addr, 0);
fixture() : host_fixture() {
addresses = local_addresses("localhost");
CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0);
auto send_pair = unbox(make_udp_datagram_socket(ep));
auto send_socket = send_pair.first;
send_socket = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep));
auto receive_socket = receive_pair.first;
receive_socket = receive_pair.first;
ep.port(ntohs(receive_pair.second));
CAF_MESSAGE("sending data to: " << to_string(ep));
auto send_guard = make_socket_guard(send_socket);
auto receive_guard = make_socket_guard(receive_socket);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("nonblocking returned an error" << err);
auto test_read_res = read(receive_socket, make_span(buf));
if (auto err = get_if<sec>(&test_read_res))
CAF_CHECK_EQUAL(*err, sec::unavailable_or_would_block);
else
CAF_FAIL("read should have failed");
auto write_ret = write(send_socket, as_bytes(make_span(hello_test)), ep);
if (auto num_bytes = get_if<size_t>(&write_ret))
CAF_CHECK_EQUAL(*num_bytes, hello_test.size());
else
CAF_FAIL("write returned an error: " << sys.render(get<sec>(write_ret)));
}
auto 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 {
CAF_FAIL("write returned an error: " << sys.render(get<sec>(read_ret)));
}
string_view buf_view{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(buf_view, hello_test);
~fixture() {
close(send_socket);
close(receive_socket);
}
std::vector<ip_address> addresses;
actor_system_config cfg;
actor_system sys{cfg};
ip_address v4_local;
ip_address v6_local;
std::vector<ip_address> addrs;
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket receive_socket;
};
} // namespace
......@@ -100,15 +69,29 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) {
// TODO: check which versions exist and test existing versions accordingly
// -> local_addresses()
if (contains(v4_local)) {
test_send_receive(v4_local);
} else if (contains(v6_local)) {
test_send_receive(v6_local);
} else {
CAF_FAIL("could not resolve 'localhost'");
}
std::vector<byte> buf(1024);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err);
CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)),
sec::unavailable_or_would_block);
CAF_MESSAGE("sending data to " << to_string(ep));
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep),
hello_test.size());
int receive_attempts = 0;
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()};
CAF_CHECK_EQUAL(received, hello_test);
}
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