Commit 43e84135 authored by Dominik Charousset's avatar Dominik Charousset

Add abstraction for stream-based sockets

parent 4ab6c8b2
......@@ -12,6 +12,7 @@ set(LIBCAF_NET_SRCS
src/pipe_socket.cpp
src/socket.cpp
src/socket_manager.cpp
src/stream_socket.cpp
)
add_custom_target(libcaf_net)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/config.hpp"
namespace caf {
namespace net {
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socket_size_type = int;
#else // CAF_WINDOWS
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
......@@ -43,14 +43,6 @@ struct network_socket : abstract_socket<network_socket> {
}
};
/// Enables or disables keepalive on `x`.
/// @relates network_socket
error keepalive(network_socket x, bool new_value);
/// Enables or disables Nagle's algorithm on `x`.
/// @relates network_socket
error tcp_nodelay(network_socket x, bool new_value);
/// Enables or disables `SIGPIPE` events from `x`.
/// @relates network_socket
error allow_sigpipe(network_socket x, bool new_value);
......@@ -96,28 +88,5 @@ void shutdown_write(network_socket x);
/// @relates network_socket
void shutdown(network_socket x);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
variant<size_t, std::errc> write(network_socket x, const void* buf,
size_t buf_size);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, 0 if the connection was
/// closed and an error code otherwise.
/// @relates pipe_socket
variant<size_t, std::errc> read(network_socket x, void* buf, size_t buf_size);
/// Creates two connected sockets to mimic network communication (usually for
/// testing purposes).
/// @relates network_socket
expected<std::pair<network_socket, network_socket>> make_network_socket_pair();
} // 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/fwd.hpp"
#include "caf/net/network_socket.hpp"
namespace caf {
namespace net {
/// A connection-oriented network communication endpoint for bidirectional byte
/// streams.
struct stream_socket : abstract_socket<stream_socket> {
using super = abstract_socket<stream_socket>;
using super::super;
constexpr operator socket() const noexcept {
return socket{id};
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
};
/// Creates two connected sockets to mimic network communication (usually for
/// testing purposes).
/// @relates stream_socket
expected<std::pair<stream_socket, stream_socket>> make_stream_socket_pair();
/// Enables or disables keepalive on `x`.
/// @relates network_socket
error keepalive(stream_socket x, bool new_value);
/// Enables or disables Nagle's algorithm on `x`.
/// @relates stream_socket
error nodelay(stream_socket x, bool new_value);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, 0 if the connection was
/// closed and an error code otherwise.
/// @relates pipe_socket
variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
variant<size_t, std::errc> write(stream_socket x, const void* buf,
size_t buf_size);
} // namespace net
} // namespace caf
......@@ -23,6 +23,7 @@
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/io/network/protocol.hpp"
......@@ -59,31 +60,10 @@ namespace net {
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
# define CAF_HAS_NOSIGPIPE_SOCKET_FLAG
const int no_sigpipe_io_flag = 0;
#elif defined(CAF_WINDOWS)
const int no_sigpipe_io_flag = 0;
#else
// Use flags to recv/send on Linux/Android but no socket option.
const int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socket_size_type = int;
error keepalive(network_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
char value = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, SOL_SOCKET, SO_KEEPALIVE, &value,
static_cast<int>(sizeof(value))));
return none;
}
error allow_sigpipe(network_socket x, bool) {
if (x == invalid_socket)
return make_error(sec::network_syscall_failed, "setsockopt",
......@@ -101,20 +81,6 @@ error allow_udp_connreset(network_socket x, bool new_value) {
}
#else // CAF_WINDOWS
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
error keepalive(network_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
int value = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, SOL_SOCKET, SO_KEEPALIVE, &value,
static_cast<unsigned>(sizeof(value))));
return none;
}
error allow_sigpipe(network_socket x, bool new_value) {
# ifdef CAF_HAS_NOSIGPIPE_SOCKET_FLAG
......@@ -159,16 +125,6 @@ error send_buffer_size(network_socket x, size_t capacity) {
return none;
}
error tcp_nodelay(network_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socket_size_type>(sizeof(flag))));
return none;
}
expected<std::string> local_addr(network_socket x) {
sockaddr_storage st;
socket_size_type st_len = sizeof(st);
......@@ -237,119 +193,5 @@ void shutdown(network_socket x) {
::shutdown(x.id, 2);
}
variant<size_t, std::errc> write(network_socket x, const void* buf,
size_t buf_size) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf), buf_size,
no_sigpipe_io_flag);
if (res < 0)
return static_cast<std::errc>(last_socket_error());
return static_cast<size_t>(res);
}
variant<size_t, std::errc> read(network_socket x, void* buf, size_t buf_size) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf), buf_size,
no_sigpipe_io_flag);
if (res < 0)
return static_cast<std::errc>(last_socket_error());
return static_cast<size_t>(res);
}
#ifdef CAF_WINDOWS
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected<std::pair<network_socket, network_socket>> make_network_socket_pair() {
auto addrlen = static_cast<int>(sizeof(sockaddr_in));
socket_id socks[2] = {invalid_socket_id, invalid_socket_id};
CAF_NET_SYSCALL("socket", listener, ==, invalid_socket_id,
::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
union {
sockaddr_in inaddr;
sockaddr addr;
} a;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
// makes sure all sockets are closed in case of an error
auto guard = detail::make_scope_guard([&] {
auto e = WSAGetLastError();
close(socket{listener});
close(socket{socks[0]});
close(socket{socks[1]});
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CAF_NET_SYSCALL("bind", tmp2, !=, 0,
bind(listener, &a.addr, static_cast<int>(sizeof(a.inaddr))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset(&a, 0, sizeof(a));
CAF_NET_SYSCALL("getsockname", tmp3, !=, 0,
getsockname(listener, &a.addr, &addrlen));
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
CAF_NET_SYSCALL("listen", tmp5, !=, 0, listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CAF_NET_SYSCALL("WSASocketW", read_fd, ==, invalid_socket_id,
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CAF_NET_SYSCALL("connect", tmp6, !=, 0,
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CAF_NET_SYSCALL("accept", write_fd, ==, invalid_socket_id,
accept(listener, nullptr, nullptr));
close(socket{listener});
guard.disable();
return std::make_pair(read_fd, write_fd);
}
#else // CAF_WINDOWS
expected<std::pair<network_socket, network_socket>> make_network_socket_pair() {
int sockets[2];
CAF_NET_SYSCALL("socketpair", spair_res, !=, 0,
socketpair(AF_UNIX, SOCK_STREAM, 0, sockets));
return std::make_pair(network_socket{sockets[0]}, network_socket{sockets[1]});
}
#endif // CAF_WINDOWS
} // namespace net
} // namespace caf
......@@ -27,7 +27,7 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
......@@ -39,7 +39,7 @@ namespace net {
expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
// Windows has no support for unidirectional pipes. Emulate pipes by using a
// pair of regular TCP sockets with read/write channels closed appropriately.
if (auto result = make_network_socket_pair()) {
if (auto result = make_stream_socket_pair()) {
shutdown_write(result->first);
shutdown_read(result->second);
return std::make_pair(socket_cast<pipe_socket>(result->first),
......@@ -52,12 +52,12 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
variant<size_t, std::errc> write(pipe_socket x, const void* buf,
size_t buf_size) {
// On Windows, a pipe consists of two stream sockets.
return write(network_socket{x.id}, buf, buf_size);
return write(socket_cast<stream_socket>(x), buf, buf_size);
}
variant<size_t, std::errc> read(pipe_socket x, void* buf, size_t buf_size) {
// On Windows, a pipe consists of two stream sockets.
return read(network_socket{x.id}, buf, buf_size);
return read(socket_cast<stream_socket>(x), buf, buf_size);
}
#else // CAF_WINDOWS
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/stream_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
#ifdef CAF_WINDOWS
constexpr int no_sigpipe_io_flag = 0;
/**************************************************************************\
* Based on work of others; *
* original header: *
* *
* Copyright 2007, 2010 by Nathan C. Myers <ncm@cantrip.org> *
* Redistribution and use in source and binary forms, with or without *
* modification, are permitted provided that the following conditions *
* are met: *
* *
* Redistributions of source code must retain the above copyright notice, *
* this list of conditions and the following disclaimer. *
* *
* Redistributions in binary form must reproduce the above copyright *
* notice, this list of conditions and the following disclaimer in the *
* documentation and/or other materials provided with the distribution. *
* *
* The name of the author must not be used to endorse or promote products *
* derived from this software without specific prior written permission. *
* *
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS *
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT *
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR *
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT *
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, *
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, *
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY *
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT *
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE *
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
expected<std::pair<stream_socket, stream_socket>> make_stream_socket_pair() {
auto addrlen = static_cast<int>(sizeof(sockaddr_in));
socket_id socks[2] = {invalid_socket_id, invalid_socket_id};
CAF_NET_SYSCALL("socket", listener, ==, invalid_socket_id,
::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
union {
sockaddr_in inaddr;
sockaddr addr;
} a;
memset(&a, 0, sizeof(a));
a.inaddr.sin_family = AF_INET;
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_port = 0;
// makes sure all sockets are closed in case of an error
auto guard = detail::make_scope_guard([&] {
auto e = WSAGetLastError();
close(socket{listener});
close(socket{socks[0]});
close(socket{socks[1]});
WSASetLastError(e);
});
// bind listener to a local port
int reuse = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<char*>(&reuse),
static_cast<int>(sizeof(reuse))));
CAF_NET_SYSCALL("bind", tmp2, !=, 0,
bind(listener, &a.addr, static_cast<int>(sizeof(a.inaddr))));
// Read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx).
memset(&a, 0, sizeof(a));
CAF_NET_SYSCALL("getsockname", tmp3, !=, 0,
getsockname(listener, &a.addr, &addrlen));
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
CAF_NET_SYSCALL("listen", tmp5, !=, 0, listen(listener, 1));
// create read-only end of the pipe
DWORD flags = 0;
CAF_NET_SYSCALL("WSASocketW", read_fd, ==, invalid_socket_id,
WSASocketW(AF_INET, SOCK_STREAM, 0, nullptr, 0, flags));
CAF_NET_SYSCALL("connect", tmp6, !=, 0,
connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe
CAF_NET_SYSCALL("accept", write_fd, ==, invalid_socket_id,
accept(listener, nullptr, nullptr));
close(socket{listener});
guard.disable();
return std::make_pair(read_fd, write_fd);
}
error keepalive(stream_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
char value = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, SOL_SOCKET, SO_KEEPALIVE, &value,
static_cast<int>(sizeof(value))));
return none;
}
#else // CAF_WINDOWS
# if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
constexpr int no_sigpipe_io_flag = 0;
# else
constexpr int no_sigpipe_io_flag = MSG_NOSIGNAL;
# endif
expected<std::pair<stream_socket, stream_socket>> make_stream_socket_pair() {
int sockets[2];
CAF_NET_SYSCALL("socketpair", spair_res, !=, 0,
socketpair(AF_UNIX, SOCK_STREAM, 0, sockets));
return std::make_pair(stream_socket{sockets[0]}, stream_socket{sockets[1]});
}
error keepalive(stream_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
int value = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, SOL_SOCKET, SO_KEEPALIVE, &value,
static_cast<unsigned>(sizeof(value))));
return none;
}
#endif // CAF_WINDOWS
error nodelay(stream_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socket_size_type>(sizeof(flag))));
return none;
}
variant<size_t, std::errc> read(stream_socket x, void* buf, size_t buf_size) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf), buf_size,
no_sigpipe_io_flag);
if (res < 0)
return static_cast<std::errc>(last_socket_error());
return static_cast<size_t>(res);
}
variant<size_t, std::errc> write(stream_socket x, const void* buf,
size_t buf_size) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf), buf_size,
no_sigpipe_io_flag);
if (res < 0)
return static_cast<std::errc>(last_socket_error());
return static_cast<size_t>(res);
}
} // namespace net
} // namespace caf
......@@ -31,9 +31,6 @@ CAF_TEST_FIXTURE_SCOPE(network_socket_tests, host_fixture)
CAF_TEST(invalid socket) {
network_socket x;
CAF_CHECK_EQUAL(keepalive(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(tcp_nodelay(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(allow_sigpipe(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(allow_udp_connreset(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(send_buffer_size(x), sec::network_syscall_failed);
CAF_CHECK_EQUAL(local_port(x), sec::network_syscall_failed);
......@@ -42,35 +39,4 @@ CAF_TEST(invalid socket) {
CAF_CHECK_EQUAL(remote_addr(x), sec::network_syscall_failed);
}
CAF_TEST(connected socket pair) {
std::vector<char> wr_buf{1, 2, 4, 8, 16, 32, 64};
std::vector<char> rd_buf(124);
CAF_MESSAGE("create sockets and configure nonblocking I/O");
auto x = unbox(make_network_socket_pair());
CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none);
CAF_CHECK_EQUAL(nonblocking(x.second, true), caf::none);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(x.first, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("transfer data from second to first socket");
CAF_CHECK_EQUAL(write(x.second, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), size_t{0});
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 stream_socket
#include "caf/net/stream_socket.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, host_fixture)
CAF_TEST(invalid socket) {
stream_socket x;
CAF_CHECK_EQUAL(keepalive(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(nodelay(x, true), sec::network_syscall_failed);
CAF_CHECK_EQUAL(allow_sigpipe(x, true), sec::network_syscall_failed);
}
CAF_TEST(connected socket pair) {
std::vector<char> wr_buf{1, 2, 4, 8, 16, 32, 64};
std::vector<char> rd_buf(124);
CAF_MESSAGE("create sockets and configure nonblocking I/O");
auto x = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none);
CAF_CHECK_EQUAL(nonblocking(x.second, true), caf::none);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
std::errc::operation_would_block);
CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(x.first, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("transfer data from second to first socket");
CAF_CHECK_EQUAL(write(x.second, wr_buf.data(), wr_buf.size()), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0);
CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), size_t{0});
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
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