Commit 0479a98a authored by Joseph Noir's avatar Joseph Noir

Address feedback

parent 61411cde
...@@ -27,8 +27,8 @@ namespace caf { ...@@ -27,8 +27,8 @@ namespace caf {
namespace net { namespace net {
namespace ip { namespace ip {
/// Returns the ip addresses assigned to `host`. /// Returns all IP addresses of to `host` (if any).
std::vector<ip_address> resolve(const std::string& host); std::vector<ip_address> resolve(string_view host);
/// Returns the hostname of this device. /// Returns the hostname of this device.
std::string hostname(); std::string hostname();
......
...@@ -27,30 +27,25 @@ namespace net { ...@@ -27,30 +27,25 @@ namespace net {
template <class Socket> template <class Socket>
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(Socket fd) : fd_(fd) { explicit socket_guard(Socket sock) : sock_(sock) {
// nop // nop
} }
~socket_guard() { ~socket_guard() {
close(); if (sock_.id != invalid_socket_id) {
net::close(sock_);
sock_.id = invalid_socket_id;
}
} }
Socket release() { Socket release() {
auto fd = fd_; auto sock = sock_;
fd_ = Socket{}; sock_.id = invalid_socket_id;
return fd; return sock;
}
void close() {
if (fd_.id != net::invalid_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
net::close(fd_);
fd_ = Socket{};
}
} }
private: private:
Socket fd_; Socket sock_;
}; };
template <class Socket> template <class Socket>
......
...@@ -52,7 +52,8 @@ public: ...@@ -52,7 +52,8 @@ public:
return handle_; return handle_;
} }
/// Returns a pointer to the multiplexer running the socket manager. /// Returns a pointer to the multiplexer running the socket manager or
/// `nullptr` if the manager is no assigned to one.
multiplexer_ptr multiplexer() const { multiplexer_ptr multiplexer() const {
return parent_.lock(); return parent_.lock();
} }
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp" #include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
...@@ -28,9 +29,7 @@ ...@@ -28,9 +29,7 @@
namespace caf { namespace caf {
namespace net { namespace net {
/// Represents an open TCP endpoint. Can be implicitly converted to a /// Represents a TCP acceptr in listening mode.
/// `stream_socket` for sending and receiving, or `network_socket`
/// for inspection.
struct tcp_accept_socket : abstract_socket<tcp_accept_socket> { struct tcp_accept_socket : abstract_socket<tcp_accept_socket> {
using super = abstract_socket<tcp_accept_socket>; using super = abstract_socket<tcp_accept_socket>;
...@@ -43,32 +42,29 @@ struct tcp_accept_socket : abstract_socket<tcp_accept_socket> { ...@@ -43,32 +42,29 @@ struct tcp_accept_socket : abstract_socket<tcp_accept_socket> {
constexpr operator network_socket() const noexcept { constexpr operator network_socket() const noexcept {
return network_socket{id}; return network_socket{id};
} }
constexpr operator stream_socket() const noexcept {
return stream_socket{id};
}
}; };
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param port The port to listen on. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// @param addr Only accepts connections originating from this address. /// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket. /// @relates tcp_accept_socket
/// @relates stream_socket expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
expected<tcp_accept_socket> make_accept_socket(ip_address host, uint16_t port, bool reuse_addr = false);
bool reuse_addr);
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param port The port to listen on. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// @param addr Only accepts connections originating from this address. /// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket. /// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates stream_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> make_accept_socket(const uri::authority_type& auth, expected<tcp_accept_socket>
bool reuse_addr = false); make_tcp_accept_socket(const uri::authority_type& node,
bool reuse_addr = false);
/// Accept a connection on `x`. /// Accepts a connection on `x`.
/// @param x Listening endpoint. /// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection. /// @returns The socket that handles the accepted connection on success, an
/// @relates stream_socket /// error otherwise.
/// @relates tcp_accept_socket
expected<tcp_stream_socket> accept(tcp_accept_socket x); expected<tcp_stream_socket> accept(tcp_accept_socket x);
} // namespace net } // namespace net
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp" #include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
...@@ -27,9 +28,7 @@ ...@@ -27,9 +28,7 @@
namespace caf { namespace caf {
namespace net { namespace net {
/// Represents a TCP connection. Can be implicitly converted to a /// Represents a TCP connection.
/// `stream_socket` for sending and receiving, or `network_socket`
/// for inspection.
struct tcp_stream_socket : abstract_socket<tcp_stream_socket> { struct tcp_stream_socket : abstract_socket<tcp_stream_socket> {
using super = abstract_socket<tcp_stream_socket>; using super = abstract_socket<tcp_stream_socket>;
...@@ -48,20 +47,18 @@ struct tcp_stream_socket : abstract_socket<tcp_stream_socket> { ...@@ -48,20 +47,18 @@ struct tcp_stream_socket : abstract_socket<tcp_stream_socket> {
} }
}; };
/// Create a `stream_socket` connected to `host`:`port`. /// Creates a `tcp_stream_socket` connected to given remote node.
/// @param host The remote host to connecto to. /// @param node Host and port of the remote node
/// @param port The port on the remote host to connect to.
/// @preferred Preferred IP version.
/// @returns The connected socket or an error. /// @returns The connected socket or an error.
/// @relates stream_socket /// @relates tcp_stream_socket
expected<stream_socket> make_connected_socket(ip_address host, uint16_t port); expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node);
/// Create a `stream_socket` connected to `auth`. /// Create a `tcp_stream_socket` connected to `auth`.
/// @param authority The remote host to connecto to. /// @param node Host and port of the remote node
/// @preferred Preferred IP version.
/// @returns The connected socket or an error. /// @returns The connected socket or an error.
/// @relates stream_socket /// @relates tcp_stream_socket
expected<stream_socket> make_connected_socket(const uri::authority_type& auth); expected<tcp_stream_socket>
make_connected_tcp_stream_socket(const uri::authority_type& node);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -40,23 +40,23 @@ public: ...@@ -40,23 +40,23 @@ public:
net::tcp_accept_socket acceptor_; net::tcp_accept_socket acceptor_;
net::socket handle() { net::tcp_accept_socket handle() {
return acceptor_; return acceptor_;
} }
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
if (auto err = parent.application().init(parent)) if (auto err = parent.application().init(parent))
return std::move(err); return err;
parent.mask_add(net::operation::read); parent.mask_add(net::operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent& parent) {
auto sck = net::accept(acceptor_); auto x = net::accept(acceptor_);
if (!sck) { if (!x) {
CAF_LOG_ERROR("accept failed:" << sck.error()); CAF_LOG_ERROR("accept failed:" << parent.system().render(x.error()));
return false; return false;
} }
auto mpx = parent.multiplexer(); auto mpx = parent.multiplexer();
...@@ -64,7 +64,7 @@ public: ...@@ -64,7 +64,7 @@ public:
CAF_LOG_DEBUG("unable to get multiplexer from parent"); CAF_LOG_DEBUG("unable to get multiplexer from parent");
return false; return false;
} }
auto child = make_endpoint_manager(mpx, parent.system(), scribe{*sck}, auto child = make_endpoint_manager(mpx, parent.system(), scribe{*x},
parent.application().make()); parent.application().make());
if (auto err = child->init()) if (auto err = child->init())
return false; return false;
...@@ -80,7 +80,7 @@ public: ...@@ -80,7 +80,7 @@ public:
template <class Parent> template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) { void resolve(Parent&, const std::string& path, actor listener) {
CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(path)); CAF_LOG_ERROR("doorman called to resolve" << CAF_ARG(path));
anon_send(listener, resolve_atom::value, "doorman cannot resolve paths"); anon_send(listener, resolve_atom::value, "doormen cannot resolve paths");
} }
template <class Parent> template <class Parent>
...@@ -91,8 +91,8 @@ public: ...@@ -91,8 +91,8 @@ public:
} }
template <class Application> template <class Application>
void handle_error(Application&, sec) { void handle_error(Application&, sec err) {
close(acceptor_); CAF_LOG_ERROR("doorman encounterd error: " << err);
} }
}; };
......
...@@ -44,7 +44,7 @@ public: ...@@ -44,7 +44,7 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
if (auto err = parent.application().init(parent)) if (auto err = parent.application().init(parent))
return std::move(err); return err;
parent.mask_add(net::operation::read); parent.mask_add(net::operation::read);
return none; return none;
} }
......
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
# ifndef _WIN32_WINNT # ifndef _WIN32_WINNT
# define _WIN32_WINNT 0x0600 # define _WIN32_WINNT 0x0600
# endif # endif
# include <iostream>
# include <iphlpapi.h> # include <iphlpapi.h>
#else #else
# include <net/if.h> # include <net/if.h>
...@@ -60,15 +59,10 @@ namespace ip { ...@@ -60,15 +59,10 @@ namespace ip {
namespace { namespace {
template <class T>
void* vptr(T* ptr) {
return static_cast<void*>(ptr);
}
void* fetch_in_addr(int family, sockaddr* addr) { void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET) if (family == AF_INET)
return vptr(&reinterpret_cast<sockaddr_in*>(addr)->sin_addr); return &reinterpret_cast<sockaddr_in*>(addr)->sin_addr;
return vptr(&reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr); return &reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr;
} }
// TODO: Use getnameinfo instead? // TODO: Use getnameinfo instead?
...@@ -86,7 +80,7 @@ int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN], ...@@ -86,7 +80,7 @@ int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN],
} // namespace } // namespace
std::vector<ip_address> resolve(const std::string& host) { std::vector<ip_address> resolve(string_view host) {
addrinfo hint; addrinfo hint;
memset(&hint, 0, sizeof(hint)); memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM; hint.ai_socktype = SOCK_STREAM;
...@@ -94,7 +88,8 @@ std::vector<ip_address> resolve(const std::string& host) { ...@@ -94,7 +88,8 @@ std::vector<ip_address> resolve(const std::string& host) {
if (host.empty()) if (host.empty())
hint.ai_flags = AI_PASSIVE; hint.ai_flags = AI_PASSIVE;
addrinfo* tmp = nullptr; addrinfo* tmp = nullptr;
if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp) != 0) std::string host_str{host.begin(), host.end()};
if (getaddrinfo(host_str.c_str(), nullptr, &hint, &tmp) != 0)
return {}; return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo}; std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN]; char buffer[INET6_ADDRSTRLEN];
...@@ -103,11 +98,10 @@ std::vector<ip_address> resolve(const std::string& host) { ...@@ -103,11 +98,10 @@ std::vector<ip_address> resolve(const std::string& host) {
auto family = fetch_addr_str(true, true, buffer, i->ai_addr); auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
if (family != AF_UNSPEC) { if (family != AF_UNSPEC) {
ip_address ip; ip_address ip;
if (auto err = parse(buffer, ip)) { if (auto err = parse(buffer, ip))
CAF_LOG_ERROR("could not parse into ip address " << buffer); CAF_LOG_ERROR("could not parse IP address: " << buffer);
continue; else
} results.emplace_back(ip);
results.emplace_back(ip);
} }
} }
// TODO: Should we just prefer ipv6 or use a config option? // TODO: Should we just prefer ipv6 or use a config option?
...@@ -124,7 +118,7 @@ std::string hostname() { ...@@ -124,7 +118,7 @@ std::string hostname() {
TCHAR buf[MAX_COMPUTERNAME_LENGTH + 1]; TCHAR buf[MAX_COMPUTERNAME_LENGTH + 1];
DWORD size = MAX_COMPUTERNAME_LENGTH; DWORD size = MAX_COMPUTERNAME_LENGTH;
GetComputerName(buf, &size); GetComputerName(buf, &size);
return std::string{buf}; return buf;
} }
#else // CAF_WINDOWS #else // CAF_WINDOWS
...@@ -134,7 +128,7 @@ std::string hostname() { ...@@ -134,7 +128,7 @@ std::string hostname() {
buf[HOST_NAME_MAX] = '\0'; buf[HOST_NAME_MAX] = '\0';
gethostname(buf, HOST_NAME_MAX); gethostname(buf, HOST_NAME_MAX);
gethostbyname(buf); gethostbyname(buf);
return std::string{buf}; return buf;
} }
#endif // CAF_WINDOWS #endif // CAF_WINDOWS
......
...@@ -36,35 +36,36 @@ namespace net { ...@@ -36,35 +36,36 @@ namespace net {
namespace { namespace {
expected<void> set_inaddr_any(socket, sockaddr_in& sa) { error set_inaddr_any(socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY; sa.sin_addr.s_addr = INADDR_ANY;
return unit; return none;
} }
expected<void> set_inaddr_any(socket x, sockaddr_in6& sa) { error set_inaddr_any(socket x, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any; sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket // Also accept ipv4 connections on this socket.
int off = 0; int off = 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0, CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY, setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off), reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socket_size_type>(sizeof(off)))); static_cast<socket_size_type>(sizeof(off))));
return unit; return none;
} }
template <int Family, int SockType = SOCK_STREAM> template <int Family>
caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr, expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
bool reuse_addr, bool any) { const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family"); static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")); CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SockType; int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC #ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC; socktype |= SOCK_CLOEXEC;
#endif #endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0)); CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false); child_process_inherit(fd, false);
// sguard closes the socket in case of exception // sguard closes the socket in case of exception
auto sguard = make_socket_guard(socket{fd}); auto sguard = make_socket_guard(tcp_accept_socket{fd});
if (reuse_addr) { if (reuse_addr) {
int on = 1; int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0, CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
...@@ -78,7 +79,8 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr, ...@@ -78,7 +79,8 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
memset(&sa, 0, sizeof(sockaddr_type)); memset(&sa, 0, sizeof(sockaddr_type));
detail::family_of(sa) = Family; detail::family_of(sa) = Family;
if (any) if (any)
set_inaddr_any(fd, sa); if (auto err = set_inaddr_any(fd, sa))
return err;
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1, CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &detail::addr_of(sa))); inet_pton(Family, addr, &detail::addr_of(sa)));
detail::port_of(sa) = htons(port); detail::port_of(sa) = htons(port);
...@@ -90,42 +92,43 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr, ...@@ -90,42 +92,43 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
} // namespace } // namespace
expected<tcp_accept_socket> make_accept_socket(ip_address addr, uint16_t port, expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr) { bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << to_string(addr)); CAF_LOG_TRACE(CAF_ARG(node));
auto h = addr.embeds_v4() ? to_string(addr.embedded_v4()) : to_string(addr); auto addr = to_string(node.address());
bool any = addr.zero(); auto make_acceptor = node.address().embeds_v4()
auto p = addr.embeds_v4() ? new_tcp_acceptor_impl<AF_INET>
? new_ip_acceptor_impl<AF_INET>(port, h.c_str(), reuse_addr, any) : new_tcp_acceptor_impl<AF_INET6>;
: new_ip_acceptor_impl<AF_INET6>(port, h.c_str(), reuse_addr, any); auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr,
node.address().zero());
if (!p) { if (!p) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port) CAF_LOG_WARNING("could not open tcp socket on: " << to_string(node));
<< CAF_ARG(h)); return make_error(sec::cannot_open_port, "tcp socket creation failed",
return make_error(sec::cannot_open_port, "tcp socket creation failed", port, to_string(node));
h);
} }
auto fd = socket_cast<tcp_accept_socket>(*p); auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(fd); auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(fd.id, SOMAXCONN)); CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(fd.id)); CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release(); return sguard.release();
} }
expected<tcp_accept_socket> make_accept_socket(const uri::authority_type& auth, expected<tcp_accept_socket>
bool reuse_addr) { make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
if (auto ip = get_if<ip_address>(&auth.host)) if (auto ip = get_if<ip_address>(&node.host))
return make_accept_socket(*ip, auth.port, reuse_addr); return make_tcp_accept_socket(ip_endpoint{*ip, node.port}, reuse_addr);
auto host = get<std::string>(auth.host); auto host = get<std::string>(node.host);
auto addrs = ip::resolve(host); auto addrs = ip::resolve(host);
if (addrs.empty()) if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available", return make_error(sec::cannot_open_port, "No local interface available",
to_string(auth)); to_string(node));
for (auto& addr : addrs) { for (auto& addr : addrs) {
if (auto sock = make_accept_socket(addr, auth.port, reuse_addr)) if (auto sock = make_tcp_accept_socket(ip_endpoint{addr, node.port},
reuse_addr))
return *sock; return *sock;
} }
return make_error(sec::cannot_open_port, "tcp socket creation failed", return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(auth)); to_string(node));
} }
expected<tcp_stream_socket> accept(tcp_accept_socket x) { expected<tcp_stream_socket> accept(tcp_accept_socket x) {
......
...@@ -52,47 +52,46 @@ bool ip_connect(stream_socket fd, std::string host, uint16_t port) { ...@@ -52,47 +52,46 @@ bool ip_connect(stream_socket fd, std::string host, uint16_t port) {
} // namespace } // namespace
expected<stream_socket> make_connected_socket(ip_address host, uint16_t port) { expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
CAF_LOG_DEBUG("try to connect to:" << CAF_ARG(to_string(host)) CAF_LOG_DEBUG("try to connect to: " << to_string(node));
<< CAF_ARG(port)); auto proto = node.address().embeds_v4() ? AF_INET : AF_INET6;
auto proto = host.embeds_v4() ? AF_INET : AF_INET6;
int socktype = SOCK_STREAM; int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC #ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC; socktype |= SOCK_CLOEXEC;
#endif #endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0)); CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0));
child_process_inherit(fd, false); child_process_inherit(fd, false);
auto sguard = make_socket_guard(stream_socket{fd}); auto sguard = make_socket_guard(tcp_stream_socket{fd});
if (proto == AF_INET6) { if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(fd, to_string(host), port)) { if (ip_connect<AF_INET6>(fd, to_string(node.address()), node.port())) {
CAF_LOG_INFO("successfully connected to (IPv6):" << CAF_ARG(host) CAF_LOG_INFO("successfully connected to (IPv6):" << to_string(node));
<< CAF_ARG(port));
return sguard.release(); return sguard.release();
} }
} else if (ip_connect<AF_INET>(fd, to_string(host.embedded_v4()), port)) { } else if (ip_connect<AF_INET>(fd, to_string(node.address().embedded_v4()),
node.port())) {
return sguard.release(); return sguard.release();
} }
CAF_LOG_WARNING("could not connect to:" << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_WARNING("could not connect to: " << to_string(node));
sguard.close();
return make_error(sec::cannot_connect_to_node); return make_error(sec::cannot_connect_to_node);
} }
expected<stream_socket> make_connected_socket(const uri::authority_type& auth) { expected<tcp_stream_socket>
auto port = auth.port; make_connected_tcp_stream_socket(const uri::authority_type& node) {
auto port = node.port;
if (port == 0) if (port == 0)
return make_error(sec::cannot_connect_to_node, "port is zero"); return make_error(sec::cannot_connect_to_node, "port is zero");
std::vector<ip_address> addrs; std::vector<ip_address> addrs;
if (auto str = get_if<std::string>(&auth.host)) if (auto str = get_if<std::string>(&node.host))
addrs = ip::resolve(std::move(*str)); addrs = ip::resolve(std::move(*str));
else if (auto addr = get_if<ip_address>(&auth.host)) else if (auto addr = get_if<ip_address>(&node.host))
addrs.push_back(*addr); addrs.push_back(*addr);
if (addrs.empty()) if (addrs.empty())
return make_error(sec::cannot_connect_to_node, "empty authority"); return make_error(sec::cannot_connect_to_node, "empty nodeority");
for (auto& addr : addrs) { for (auto& addr : addrs) {
if (auto sock = make_connected_socket(addr, port)) if (auto sock = make_connected_tcp_stream_socket(ip_endpoint{addr, port}))
return *sock; return *sock;
} }
return make_error(sec::cannot_connect_to_node, to_string(auth)); return make_error(sec::cannot_connect_to_node, to_string(node));
} }
} // namespace net } // namespace net
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/net/ip.hpp" #include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -119,23 +120,22 @@ public: ...@@ -119,23 +120,22 @@ public:
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture) CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(tcp connect) { CAF_TEST(tcp connect) {
auto acceptor = unbox(make_accept_socket(auth, false)); auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor))); auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port); CAF_MESSAGE("opened acceptor on port " << port);
uri::authority_type dst; uri::authority_type dst;
dst.port = port; dst.port = port;
dst.host = std::string{"localhost"}; dst.host = std::string{"localhost"};
auto con_fd = unbox(make_connected_socket(dst)); auto conn = make_socket_guard(unbox(make_connected_tcp_stream_socket(dst)));
auto acc_fd = unbox(accept(acceptor)); auto accepted = make_socket_guard(unbox(accept(acceptor)));
CAF_MESSAGE("accepted connection"); CAF_MESSAGE("accepted connection");
close(con_fd);
close(acc_fd);
close(acceptor);
} }
CAF_TEST(doorman accept) { CAF_TEST(doorman accept) {
auto acceptor = unbox(make_accept_socket(auth, false)); auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor))); auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port); CAF_MESSAGE("opened acceptor on port " << port);
auto mgr = make_endpoint_manager(mpx, sys, policy::doorman{acceptor}, auto mgr = make_endpoint_manager(mpx, sys, policy::doorman{acceptor},
dummy_application_factory{}); dummy_application_factory{});
...@@ -146,13 +146,11 @@ CAF_TEST(doorman accept) { ...@@ -146,13 +146,11 @@ CAF_TEST(doorman accept) {
uri::authority_type dst; uri::authority_type dst;
dst.port = port; dst.port = port;
dst.host = std::string{"localhost"}; dst.host = std::string{"localhost"};
auto fd = unbox(make_connected_socket(dst)); auto conn = make_socket_guard(unbox(make_connected_tcp_stream_socket(dst)));
CAF_MESSAGE("waiting for connection"); CAF_MESSAGE("waiting for connection");
while (mpx->num_socket_managers() != before + 1) while (mpx->num_socket_managers() != before + 1)
handle_io_event(); handle_io_event();
CAF_MESSAGE("connected"); CAF_MESSAGE("connected");
close(acceptor);
close(fd);
} }
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