Commit caf67249 authored by Jakob Otto's avatar Jakob Otto

Include review feedback

parent 554188ee
......@@ -24,9 +24,9 @@
namespace caf {
namespace detail {
sockaddr_storage to_sockaddr(const ip_endpoint& ep);
void convert(const ip_endpoint& src, sockaddr_storage& dst);
ip_endpoint to_ip_endpoint(const sockaddr_storage& addr);
error convert(const sockaddr_storage& src, ip_endpoint& dst);
} // namespace detail
} // namespace caf
......@@ -47,30 +47,29 @@ struct udp_datagram_socket : abstract_socket<udp_datagram_socket> {
/// port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& ep,
bool reuse_addr = false);
expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, 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);
/// Receives data from `x`.
/// @param x udp_datagram_socket.
/// @param buf Points to destination buffer.
/// @returns The number of received bytes and the ip_endpoint on success, an
/// error code otherwise.
/// Receives the next datagram on socket `x`.
/// @param x The UDP socket for receiving datagrams.
/// @param buf Writable output buffer.
/// @returns The number of received bytes and the sender as `ip_endpoint` on
/// success, an error code otherwise.
/// @relates udp_datagram_socket
/// @post either the result is a `sec` or a pair of positive (non-zero) integer
/// and ip_endpoint
/// @post buf was modified and the resulting integer represents the length of
/// the received datagram, even if it did not fit into the given buffer.
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
span<byte> buf);
/// Transmits data from `x` to given eendpoint.
/// @param x udp_datagram_socket.
/// @param buf Points to the message to send.
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
/// @param buf The buffer to send.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep);
......
......@@ -17,43 +17,46 @@
******************************************************************************/
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/error.hpp"
#include "caf/ipv4_endpoint.hpp"
#include "caf/ipv6_endpoint.hpp"
#include <iostream>
#include "caf/sec.hpp"
namespace caf {
namespace detail {
sockaddr_storage to_sockaddr(const ip_endpoint& ep) {
sockaddr_storage sockaddr = {};
if (ep.address().embeds_v4()) {
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&sockaddr);
void convert(const ip_endpoint& src, sockaddr_storage& dst) {
if (src.address().embeds_v4()) {
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&dst);
sockaddr4->sin_family = AF_INET;
sockaddr4->sin_port = ntohs(ep.port());
sockaddr4->sin_addr.s_addr = ep.address().embedded_v4().bits();
sockaddr4->sin_port = ntohs(src.port());
sockaddr4->sin_addr.s_addr = src.address().embedded_v4().bits();
} else {
auto sockaddr6 = reinterpret_cast<sockaddr_in6*>(&sockaddr);
auto sockaddr6 = reinterpret_cast<sockaddr_in6*>(&dst);
sockaddr6->sin6_family = AF_INET6;
sockaddr6->sin6_port = ntohs(ep.port());
memcpy(&sockaddr6->sin6_addr, ep.address().bytes().data(),
ep.address().bytes().size());
sockaddr6->sin6_port = ntohs(src.port());
memcpy(&sockaddr6->sin6_addr, src.address().bytes().data(),
src.address().bytes().size());
}
return sockaddr;
}
ip_endpoint to_ip_endpoint(const sockaddr_storage& addr) {
if (addr.ss_family == AF_INET) {
auto sockaddr4 = reinterpret_cast<const sockaddr_in*>(&addr);
error convert(const sockaddr_storage& src, ip_endpoint& dst) {
if (src.ss_family == AF_INET) {
auto sockaddr4 = reinterpret_cast<const sockaddr_in&>(src);
ipv4_address ipv4_addr;
memcpy(ipv4_addr.data().data(), &sockaddr4->sin_addr, ipv4_addr.size());
return {ipv4_addr, htons(sockaddr4->sin_port)};
} else {
auto sockaddr6 = reinterpret_cast<const sockaddr_in6*>(&addr);
memcpy(ipv4_addr.data().data(), &sockaddr4.sin_addr, ipv4_addr.size());
dst = ip_endpoint{ipv4_addr, htons(sockaddr4.sin_port)};
} else if (src.ss_family == AF_INET6) {
auto sockaddr6 = reinterpret_cast<const sockaddr_in6&>(src);
ipv6_address ipv6_addr;
memcpy(ipv6_addr.bytes().data(), &sockaddr6->sin6_addr,
memcpy(ipv6_addr.bytes().data(), &sockaddr6.sin6_addr,
ipv6_addr.bytes().size());
return {ipv6_addr, htons(sockaddr6->sin6_port)};
dst = ip_endpoint{ipv6_addr, htons(sockaddr6.sin6_port)};
} else {
return sec::invalid_argument;
}
return none;
}
} // namespace detail
......
......@@ -55,10 +55,11 @@ error allow_connreset(udp_datagram_socket x, bool) {
#endif // CAF_WINDOWS
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& ep,
bool reuse_addr) {
expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(ep));
auto addr = to_sockaddr(ep);
sockaddr_storage addr;
detail::convert(ep, addr);
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(addr.ss_family, SOCK_DGRAM, 0));
udp_datagram_socket sock{fd};
......@@ -81,8 +82,7 @@ expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint& ep,
auto port = addr.ss_family == AF_INET
? reinterpret_cast<sockaddr_in*>(&addr)->sin_port
: reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port;
ep.port(ntohs(port));
return sguard.release();
return std::make_pair(sguard.release(), port);
}
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
......@@ -94,13 +94,15 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
&len);
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret)) {
if (*num_bytes == 0)
CAF_LOG_INFO("Received empty datagram");
else if (*num_bytes > buf.size())
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes)
<< " bytes");
auto ep = detail::to_ip_endpoint(addr);
CAF_LOG_INFO_IF(*num_bytes == 0, "Received empty datagram");
CAF_LOG_WARNING_IF(*num_bytes > buf.size(),
"recvfrom cut of message, only received "
<< 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;
return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
} else {
return get<sec>(ret);
......@@ -109,7 +111,8 @@ 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) {
auto addr = detail::to_sockaddr(ep);
sockaddr_storage addr;
detail::convert(ep, addr);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
ep.address().embeds_v4() ? sizeof(sockaddr_in)
......
......@@ -22,6 +22,8 @@
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <cstring>
#include "caf/detail/socket_sys_includes.hpp"
......@@ -31,40 +33,80 @@
using namespace caf;
using namespace caf::detail;
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;
}
sockaddr_in6 sockaddr6_src;
sockaddr_in6 sockaddr6_dst;
sockaddr_in sockaddr4_src;
sockaddr_in sockaddr4_dst;
ip_endpoint ep_src;
ip_endpoint ep_dst;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(convert_ip_endpoint_tests, fixture)
CAF_TEST(sockaddr_in6 roundtrip) {
sockaddr_in6 source_addr = {};
source_addr.sin6_family = AF_INET6;
source_addr.sin6_port = htons(23);
source_addr.sin6_addr = in6addr_loopback;
auto ep = to_ip_endpoint(reinterpret_cast<sockaddr_storage&>(source_addr));
auto dest_addr = to_sockaddr(ep);
CAF_CHECK_EQUAL(memcmp(&source_addr, &dest_addr, sizeof(sockaddr_in6)), 0);
ip_endpoint ep;
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr6_src),
ep),
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);
}
CAF_TEST(ipv6_endpoint roundtrip) {
ipv6_endpoint source_ep;
if (auto err = detail::parse("[::1]:55555", source_ep))
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
if (auto err = detail::parse("[::1]:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
auto addr = to_sockaddr(source_ep);
auto dest_ep = to_ip_endpoint(addr);
CAF_CHECK_EQUAL(source_ep, dest_ep);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
convert(ep_src, addr);
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST(sockaddr_in4 roundtrip) {
sockaddr_in source_addr = {};
source_addr.sin_family = AF_INET;
source_addr.sin_port = htons(23);
source_addr.sin_addr.s_addr = INADDR_LOOPBACK;
auto ep = to_ip_endpoint(reinterpret_cast<sockaddr_storage&>(source_addr));
auto dest_addr = to_sockaddr(ep);
CAF_CHECK_EQUAL(memcmp(&source_addr, &dest_addr, sizeof(sockaddr_in)), 0);
ip_endpoint ep;
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr4_src),
ep),
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);
}
CAF_TEST(ipv4_endpoint roundtrip) {
ipv6_endpoint source_ep;
if (auto err = detail::parse("127.0.0.1:55555", source_ep))
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
if (auto err = detail::parse("127.0.0.1:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
auto addr = to_sockaddr(source_ep);
auto dest_ep = to_ip_endpoint(addr);
CAF_CHECK_EQUAL(source_ep, dest_ep);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
convert(ep_src, addr);
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST_FIXTURE_SCOPE_END();
......@@ -54,25 +54,29 @@ struct fixture : host_fixture {
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));
auto send_pair = unbox(make_udp_datagram_socket(ep));
auto send_socket = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep));
auto 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(sender);
auto receive_guard = make_socket_guard(receiver);
if (auto err = nonblocking(socket_cast<net::socket>(receiver), true))
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(receiver, make_span(buf));
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(sender, as_bytes(make_span(hello_test)), ep);
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(receiver, make_span(buf));
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);
......
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