Commit 5bff4f92 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #17

Add tcp socket types
parents a0ccc705 43bea091
......@@ -20,6 +20,8 @@ set(LIBCAF_NET_SRCS
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
src/tcp_stream_socket.cpp
src/tcp_accept_socket.cpp
)
add_custom_target(libcaf_net)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/detail/socket_sys_includes.hpp"
namespace caf {
namespace detail {
inline auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
inline auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
inline auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
inline auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
inline auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
inline auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace detail
} // namespace caf
......@@ -32,6 +32,8 @@ struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
struct tcp_stream_socket;
struct tcp_accept_socket;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
/// Represents a TCP acceptor in listening mode.
struct tcp_accept_socket : abstract_socket<tcp_accept_socket> {
using super = abstract_socket<tcp_accept_socket>;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
};
/// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// Passing port 0 lets the OS choose the port.
/// @relates tcp_accept_socket
expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr = false);
/// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host.
/// Passing port 0 lets the OS choose the port.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket
expected<tcp_accept_socket>
make_tcp_accept_socket(const uri::authority_type& node,
bool reuse_addr = false);
/// Accepts a connection on `x`.
/// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection on success, an
/// error otherwise.
/// @relates tcp_accept_socket
expected<tcp_stream_socket> accept(tcp_accept_socket x);
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/ip_endpoint.hpp"
#include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
/// Represents a TCP connection.
struct tcp_stream_socket : abstract_socket<tcp_stream_socket> {
using super = abstract_socket<tcp_stream_socket>;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
constexpr operator stream_socket() const noexcept {
return stream_socket{id};
}
};
/// 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<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node);
/// Create a `tcp_stream_socket` connected to `auth`.
/// @param node Host and port of the remote node.
/// @returns The connected socket or an error.
/// @relates tcp_stream_socket
expected<tcp_stream_socket>
make_connected_tcp_stream_socket(const uri::authority_type& node);
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/sockaddr_members.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/sec.hpp"
namespace caf {
namespace net {
namespace {
error set_inaddr_any(socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return none;
}
error set_inaddr_any(socket x, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// Also accept ipv4 connections on this socket.
int off = 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socket_size_type>(sizeof(off))));
return none;
}
template <int Family>
expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false);
// sguard closes the socket in case of exception
auto sguard = make_socket_guard(tcp_accept_socket{fd});
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
using sockaddr_type = typename std::conditional<
Family == AF_INET, sockaddr_in, sockaddr_in6>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
detail::family_of(sa) = Family;
if (any)
if (auto err = set_inaddr_any(fd, sa))
return err;
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &detail::addr_of(sa)));
detail::port_of(sa) = htons(port);
CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa))));
return sguard.release();
}
} // namespace
expected<tcp_accept_socket> make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(node));
auto addr = to_string(node.address());
auto make_acceptor = node.address().embeds_v4()
? new_tcp_acceptor_impl<AF_INET>
: new_tcp_acceptor_impl<AF_INET6>;
auto p = make_acceptor(node.port(), addr.c_str(), reuse_addr,
node.address().zero());
if (!p) {
CAF_LOG_WARNING("could not create tcp socket for: " << to_string(node));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node));
}
auto sock = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(sock);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(sock.id, SOMAXCONN));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
return sguard.release();
}
expected<tcp_accept_socket>
make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
if (auto ip = get_if<ip_address>(&node.host))
return make_tcp_accept_socket(ip_endpoint{*ip, node.port}, reuse_addr);
auto host = get<std::string>(node.host);
auto addrs = ip::resolve(host);
if (addrs.empty())
return make_error(sec::cannot_open_port, "no local interface available",
to_string(node));
for (auto& addr : addrs) {
if (auto sock = make_tcp_accept_socket(ip_endpoint{addr, node.port},
reuse_addr))
return *sock;
}
return make_error(sec::cannot_open_port, "tcp socket creation failed",
to_string(node));
}
expected<tcp_stream_socket> accept(tcp_accept_socket x) {
auto sock = ::accept(x.id, nullptr, nullptr);
if (sock == net::invalid_socket_id) {
auto err = net::last_socket_error();
if (err != std::errc::operation_would_block
&& err != std::errc::resource_unavailable_try_again) {
return caf::make_error(sec::unavailable_or_would_block);
}
return caf::make_error(sec::socket_operation_failed, "tcp accept failed");
}
return {sock};
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/sockaddr_members.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace {
template <int Family>
bool ip_connect(stream_socket fd, std::string host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
<< CAF_ARG(fd.id) << CAF_ARG(host) << CAF_ARG(port));
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
using sockaddr_type = typename std::conditional<
Family == AF_INET, sockaddr_in, sockaddr_in6>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
inet_pton(Family, host.c_str(), &detail::addr_of(sa));
detail::family_of(sa) = Family;
detail::port_of(sa) = htons(port);
using sa_ptr = const sockaddr*;
return ::connect(fd.id, reinterpret_cast<sa_ptr>(&sa), sizeof(sa)) == 0;
}
} // namespace
expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
CAF_LOG_DEBUG("tcp connect to: " << to_string(node));
auto proto = node.address().embeds_v4() ? AF_INET : AF_INET6;
int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0));
child_process_inherit(fd, false);
auto sguard = make_socket_guard(tcp_stream_socket{fd});
if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(fd, to_string(node.address()), node.port())) {
CAF_LOG_INFO("successfully connected to (IPv6):" << to_string(node));
return sguard.release();
}
} else if (ip_connect<AF_INET>(fd, to_string(node.address().embedded_v4()),
node.port())) {
CAF_LOG_INFO("successfully connected to (IPv4):" << to_string(node));
return sguard.release();
}
CAF_LOG_WARNING("could not connect to: " << to_string(node));
return make_error(sec::cannot_connect_to_node);
}
expected<tcp_stream_socket>
make_connected_tcp_stream_socket(const uri::authority_type& node) {
auto port = node.port;
if (port == 0)
return make_error(sec::cannot_connect_to_node, "port is zero");
std::vector<ip_address> addrs;
if (auto str = get_if<std::string>(&node.host))
addrs = ip::resolve(std::move(*str));
else if (auto addr = get_if<ip_address>(&node.host))
addrs.push_back(*addr);
if (addrs.empty())
return make_error(sec::cannot_connect_to_node, "empty authority");
for (auto& addr : addrs) {
if (auto sock = make_connected_tcp_stream_socket(ip_endpoint{addr, port}))
return *sock;
}
return make_error(sec::cannot_connect_to_node, to_string(node));
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE tcp_sockets
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/net/socket_guard.hpp"
using namespace caf;
using namespace caf::net;
namespace {
// TODO: switch to std::operator""s when switching to C++14
std::string operator"" _s(const char* str, size_t size) {
return std::string(str, size);
}
struct fixture : host_fixture {
fixture() {
auth.port = 0;
auth.host = "0.0.0.0"_s;
}
uri::authority_type auth;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(tcp_sockets_tests, fixture)
CAF_TEST(open tcp port) {
auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(acceptor));
CAF_CHECK_NOT_EQUAL(port, 0);
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port);
}
CAF_TEST(tcp connect) {
auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto port = unbox(local_port(acceptor));
CAF_CHECK_NOT_EQUAL(port, 0);
auto acceptor_guard = make_socket_guard(acceptor);
CAF_MESSAGE("opened acceptor on port " << port);
uri::authority_type dst;
dst.port = port;
dst.host = "localhost"_s;
CAF_MESSAGE("connecting to localhost");
auto conn = unbox(make_connected_tcp_stream_socket(dst));
auto conn_guard = make_socket_guard(conn);
CAF_CHECK_NOT_EQUAL(conn, invalid_socket);
auto accepted = unbox(accept(acceptor));
auto accepted_guard = make_socket_guard(conn);
CAF_CHECK_NOT_EQUAL(accepted, invalid_socket);
CAF_MESSAGE("connected");
}
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