Commit 5b479631 authored by Jakob Otto's avatar Jakob Otto

Rework udp_datagram_socket

parent 4be5d1d9
......@@ -41,6 +41,13 @@ struct udp_datagram_socket : abstract_socket<udp_datagram_socket> {
}
};
/// Creates a `tcp_stream_socket` connected to given remote node.
/// @param node Host and port of the remote node.
/// @returns The connected socket or an error.
/// @relates tcp_stream_socket
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& node,
bool reuse_addr = false);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
error allow_connreset(udp_datagram_socket x, bool new_value);
......@@ -65,13 +72,6 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep);
/// Binds given socket to given ip_endpoint.
/// @param x the socket that should be bound.
/// @param ep the endpoint to which the socket should be bound.
/// @returns The port that was actually bound.
/// @Relates udp_datagram_socket
expected<uint16_t> bind(udp_datagram_socket x, ip_endpoint ep);
/// Converts the result from I/O operation on a ::udp_datagram_socket to either
/// an error code or a non-zero positive integer.
/// @relates udp_datagram_socket
......
......@@ -26,6 +26,7 @@
#include "caf/expected.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
namespace caf {
......@@ -54,10 +55,34 @@ error allow_connreset(udp_datagram_socket x, bool) {
#endif // CAF_WINDOWS
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& node,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node));
auto addr = to_sockaddr(node);
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(addr.ss_family, SOCK_DGRAM, 0));
udp_datagram_socket sock{fd};
auto sguard = make_socket_guard(sock);
socklen_t len = (addr.ss_family == AF_INET) ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
CAF_NET_SYSCALL("bind", err, !=, 0,
::bind(sock.id, reinterpret_cast<sockaddr*>(&addr), len));
CAF_NET_SYSCALL("getsockname", erro, !=, 0,
getsockname(sock.id, reinterpret_cast<sockaddr*>(&addr),
&len));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
auto port = addr.ss_family == AF_INET
? reinterpret_cast<sockaddr_in*>(&addr)->sin_port
: reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port;
node.port(ntohs(port));
return sguard.release();
}
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
span<byte> buf) {
sockaddr_in6 addr = {};
socklen_t len = sizeof(sockaddr_in);
sockaddr_storage addr = {};
socklen_t len = sizeof(sockaddr_storage);
auto res = ::recvfrom(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
&len);
......@@ -81,7 +106,8 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
auto addr = detail::to_sockaddr(ep);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
sizeof(sockaddr_in6));
ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6));
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret))
return *num_bytes;
......@@ -89,17 +115,6 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
return get<sec>(ret);
}
expected<uint16_t> bind(udp_datagram_socket x, ip_endpoint ep) {
auto addr = to_sockaddr(ep);
CAF_NET_SYSCALL("bind", err, !=, 0,
::bind(x.id, reinterpret_cast<sockaddr*>(&addr),
sizeof(sockaddr_in6)));
socklen_t len = sizeof(sockaddr_in6);
CAF_NET_SYSCALL("getsockname", erro, !=, 0,
getsockname(x.id, reinterpret_cast<sockaddr*>(&addr), &len));
return addr.sin6_port;
}
variant<size_t, sec>
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) {
if (res < 0) {
......
......@@ -27,64 +27,87 @@
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp"
using namespace caf;
using namespace caf::net;
namespace {
expected<udp_datagram_socket> make_socket() {
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(AF_INET6, SOCK_DGRAM, 0));
return udp_datagram_socket{fd};
}
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}},
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);
auto sender = unbox(make_udp_datagram_socket(ep));
ep.port(0);
auto receiver = unbox(make_udp_datagram_socket(ep));
CAF_MESSAGE("sending data to: " << to_string(ep));
auto guard = detail::make_scope_guard([&] {
close(socket_cast<net::socket>(sender));
close(socket_cast<net::socket>(receiver));
});
if (auto err = nonblocking(socket_cast<net::socket>(receiver), true))
CAF_FAIL("nonblocking returned an error" << err);
auto test_read_res = read(receiver, 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(sender, 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(receiver, 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);
}
actor_system_config cfg;
actor_system sys{cfg};
ip_address v4_local;
ip_address v6_local;
std::vector<ip_address> addrs;
};
constexpr string_view hello_test = "Hello test!";
} // namespace
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) {
std::vector<byte> buf(1024);
ip_endpoint ep;
if (auto err = detail::parse("[::1]:0", ep))
CAF_FAIL("unable to parse input: " << err);
auto sender = unbox(make_socket());
auto receiver = unbox(make_socket());
auto guard = detail::make_scope_guard([&] {
close(socket_cast<net::socket>(sender));
close(socket_cast<net::socket>(receiver));
});
auto port = unbox(bind(receiver, ep));
CAF_MESSAGE("socket bound to port " << port);
ep.port(htons(port));
if (nonblocking(socket_cast<net::socket>(receiver), true))
CAF_FAIL("nonblocking failed");
auto test_read_res = read(receiver, 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(sender, 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(receiver, 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);
CAF_TEST(send and receive IPv4) {
if (!contains(v4_local)) {
CAF_MESSAGE("this host does not have av IPv4 address");
} else {
test_send_receive(v4_local);
}
}
CAF_TEST(send and receive IPv6) {
if (!contains(v6_local)) {
CAF_MESSAGE("this host does not have av IPv6 address");
} else {
CAF_FAIL("write returned an error: " << sys.render(get<sec>(read_ret)));
test_send_receive(v6_local);
}
string_view buf_view{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(buf_view, 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