Commit 138dc295 authored by Dominik Charousset's avatar Dominik Charousset

Add write overload for multiple buffers

parent c08a983d
......@@ -56,21 +56,30 @@ 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, an error code otherwise.
/// @relates pipe_socket
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> read(stream_socket x, span<byte> buf);
/// 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
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(stream_socket x, span<const byte> buf);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param bufs Points to the message to send, scattered across up to 10
/// buffers.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// @relates stream_socket
......
......@@ -27,6 +27,10 @@
#include "caf/span.hpp"
#include "caf/variant.hpp"
#ifdef CAF_POSIX
# include <sys/uio.h>
#endif
namespace caf {
namespace net {
......@@ -175,6 +179,47 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
return check_stream_socket_io_res(res);
}
#ifdef CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](span<const byte> buf) {
auto data = const_cast<byte*>(buf.data());
return WSABUF{static_cast<ULONG>(buf.size()),
reinterpret_cast<CHAR*>(data)};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
DWORD bytes_sent = 0;
auto res = WSASend(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, nullptr, nullptr);
if (res != 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(bytes_sent);
}
#else // CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
iovec buf_array[10];
auto convert = [](span<const byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
auto res = writev(x.id, buf_array, static_cast<int>(bufs.size()));
return check_stream_socket_io_res(res);
}
#endif // CAF_WINDOWS
variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res) {
if (res == 0)
......
......@@ -30,7 +30,15 @@
using namespace caf;
using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, host_fixture)
namespace {
byte operator""_b(unsigned long long x) {
return static_cast<byte>(x);
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(network_socket_simple_tests, host_fixture)
CAF_TEST(invalid socket) {
stream_socket x;
......@@ -39,36 +47,71 @@ CAF_TEST(invalid socket) {
CAF_CHECK_EQUAL(allow_sigpipe(x, true), sec::network_syscall_failed);
}
CAF_TEST(connected socket pair) {
std::vector<byte> wr_buf{byte(1), byte(2), byte(4), byte(8),
byte(16), byte(32), byte(64)};
std::vector<byte> 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, make_span(rd_buf)),
CAF_TEST_FIXTURE_SCOPE_END()
namespace {
struct fixture : host_fixture {
fixture() : rd_buf(124) {
std::tie(first, second) = unbox(make_stream_socket_pair());
CAF_REQUIRE_EQUAL(nonblocking(first, true), caf::none);
CAF_REQUIRE_EQUAL(nonblocking(second, true), caf::none);
CAF_REQUIRE_NOT_EQUAL(unbox(send_buffer_size(first)), 0u);
CAF_REQUIRE_NOT_EQUAL(unbox(send_buffer_size(second)), 0u);
}
~fixture() {
close(first);
close(second);
}
stream_socket first;
stream_socket second;
std::vector<byte> rd_buf;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, fixture)
CAF_TEST(read on empty sockets) {
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)),
sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)),
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)),
sec::unavailable_or_would_block);
}
CAF_TEST(transfer data from first to second socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(x.first, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK_EQUAL(write(first, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("transfer data from second to first socket");
CAF_CHECK_EQUAL(write(x.second, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, make_span(rd_buf)), wr_buf.size());
}
CAF_TEST(transfer data from second to first socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_CHECK_EQUAL(write(second, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), sec::socket_disconnected);
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
CAF_TEST(shut down first socket and observe shutdown on the second one) {
close(first);
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), sec::socket_disconnected);
first.id = invalid_socket_id;
}
CAF_TEST(transfer data using multiple buffers) {
std::vector<byte> wr_buf_1{1_b, 2_b, 4_b};
std::vector<byte> wr_buf_2{8_b, 16_b, 32_b, 64_b};
std::vector<byte> full_buf;
full_buf.insert(full_buf.end(), wr_buf_1.begin(), wr_buf_1.end());
full_buf.insert(full_buf.end(), wr_buf_2.begin(), wr_buf_2.end());
CAF_CHECK_EQUAL(write(second, {make_span(wr_buf_1), make_span(wr_buf_2)}),
full_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), full_buf.size());
CAF_CHECK(std::equal(full_buf.begin(), full_buf.end(), rd_buf.begin()));
}
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