Commit 2a628275 authored by Jakob Otto's avatar Jakob Otto

Refactor socket functions

parent cb86c22d
...@@ -22,10 +22,12 @@ ...@@ -22,10 +22,12 @@
#include <system_error> #include <system_error>
#include <utility> #include <utility>
#include "caf/byte.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/abstract_socket.hpp" #include "caf/net/abstract_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/socket_id.hpp" #include "caf/net/socket_id.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -52,7 +54,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe(); ...@@ -52,7 +54,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe();
/// @param buf_size Specifies the size of the buffer in bytes. /// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size); variant<size_t, sec> write(pipe_socket x, span<const byte> buf);
/// Receives data from `x`. /// Receives data from `x`.
/// @param x Connected endpoint. /// @param x Connected endpoint.
...@@ -60,7 +62,7 @@ variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size); ...@@ -60,7 +62,7 @@ variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size);
/// @param buf_size Specifies the maximum size of the buffer in bytes. /// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, otherwise an error code. /// @returns The number of received bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size); variant<size_t, sec> read(pipe_socket x, span<byte>);
/// Converts the result from I/O operation on a ::pipe_socket to either an /// Converts the result from I/O operation on a ::pipe_socket to either an
/// error code or a non-zero positive integer. /// error code or a non-zero positive integer.
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <array> #include <array>
#include <cstdint> #include <cstdint>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp" #include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
...@@ -55,7 +56,7 @@ public: ...@@ -55,7 +56,7 @@ public:
void handle_error(sec code) override; void handle_error(sec code) override;
private: private:
std::array<char, sizeof(intptr_t)> buf_; std::array<byte, sizeof(intptr_t)> buf_;
size_t buf_size_; size_t buf_size_;
}; };
......
...@@ -18,8 +18,10 @@ ...@@ -18,8 +18,10 @@
#pragma once #pragma once
#include "caf/byte.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -60,7 +62,7 @@ error nodelay(stream_socket x, bool new_value); ...@@ -60,7 +62,7 @@ error nodelay(stream_socket x, bool new_value);
/// @returns The number of received bytes on success, an error code otherwise. /// @returns The number of received bytes on success, an error code otherwise.
/// @relates pipe_socket /// @relates pipe_socket
/// @post either the result is a `sec` or a positive (non-zero) integer /// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size); variant<size_t, sec> read(stream_socket x, span<byte> buf);
/// Transmits data from `x` to its peer. /// Transmits data from `x` to its peer.
/// @param x Connected endpoint. /// @param x Connected endpoint.
...@@ -69,7 +71,7 @@ variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size); ...@@ -69,7 +71,7 @@ variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size);
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket /// @relates pipe_socket
/// @post either the result is a `sec` or a positive (non-zero) integer /// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(stream_socket x, const void* buf, size_t buf_size); variant<size_t, sec> write(stream_socket x, span<const byte> buf);
/// Converts the result from I/O operation on a ::stream_socket to either an /// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer. /// error code or a non-zero positive integer.
......
...@@ -50,10 +50,10 @@ public: ...@@ -50,10 +50,10 @@ public:
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent& parent) {
void* buf = read_buf_.data() + collected_; auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = read(handle_, buf, len); auto ret = read(handle_, make_span(buf, len));
// Update state. // Update state.
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
...@@ -90,9 +90,9 @@ public: ...@@ -90,9 +90,9 @@ public:
if (write_buf_.empty()) if (write_buf_.empty())
return false; return false;
auto len = write_buf_.size() - written_; auto len = write_buf_.size() - written_;
const void* buf = write_buf_.data() + written_; auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, buf, len); auto ret = net::write(handle_, as_bytes(make_span(buf, len)));
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state. // Update state.
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <algorithm> #include <algorithm>
#include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
...@@ -28,6 +29,7 @@ ...@@ -28,6 +29,7 @@
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
#ifndef CAF_WINDOWS #ifndef CAF_WINDOWS
...@@ -117,7 +119,8 @@ void multiplexer::update(const socket_manager_ptr& mgr) { ...@@ -117,7 +119,8 @@ void multiplexer::update(const socket_manager_ptr& mgr) {
{ // Lifetime scope of guard. { // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) if (write_handle_ != invalid_socket)
res = write(write_handle_, &value, sizeof(intptr_t)); res = write(write_handle_,
as_bytes(make_span(&value, sizeof(intptr_t))));
else else
res = sec::socket_invalid; res = sec::socket_invalid;
} }
......
...@@ -22,13 +22,16 @@ ...@@ -22,13 +22,16 @@
#include <cstdlib> #include <cstdlib>
#include <utility> #include <utility>
#include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
namespace caf { namespace caf {
...@@ -49,14 +52,14 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -49,14 +52,14 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
} }
} }
variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size) { variant<size_t, sec> write(pipe_socket x, span<const byte> buf) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return write(socket_cast<stream_socket>(x), buf, buf_size); return write(socket_cast<stream_socket>(x), buf);
} }
variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(pipe_socket x, span<byte> buf) {
// On Windows, a pipe consists of two stream sockets. // On Windows, a pipe consists of two stream sockets.
return read(socket_cast<stream_socket>(x), buf, buf_size); return read(socket_cast<stream_socket>(x), buf);
} }
#else // CAF_WINDOWS #else // CAF_WINDOWS
...@@ -80,13 +83,15 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() { ...@@ -80,13 +83,15 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe() {
return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]}); return std::make_pair(pipe_socket{pipefds[0]}, pipe_socket{pipefds[1]});
} }
variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size) { variant<size_t, sec> write(pipe_socket x, span<const byte> buf) {
auto res = ::write(x.id, buf, buf_size); auto res = ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res); return check_pipe_socket_io_res(res);
} }
variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(pipe_socket x, span<byte> buf) {
auto res = ::read(x.id, buf, buf_size); auto res = ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res); return check_pipe_socket_io_res(res);
} }
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
namespace caf { namespace caf {
...@@ -38,7 +39,7 @@ pollset_updater::~pollset_updater() { ...@@ -38,7 +39,7 @@ pollset_updater::~pollset_updater() {
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
for (;;) { for (;;) {
auto res = read(handle(), buf_.data(), buf_.size() - buf_size_); auto res = read(handle(), make_span(buf_.data(), buf_.size() - buf_size_));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
buf_size_ += *num_bytes; buf_size_ += *num_bytes;
if (buf_.size() == buf_size_) { if (buf_.size() == buf_size_) {
......
...@@ -20,7 +20,9 @@ ...@@ -20,7 +20,9 @@
#include <system_error> #include <system_error>
#include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/span.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
......
...@@ -18,11 +18,13 @@ ...@@ -18,11 +18,13 @@
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/byte.hpp"
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
namespace caf { namespace caf {
...@@ -161,15 +163,15 @@ error nodelay(stream_socket x, bool new_value) { ...@@ -161,15 +163,15 @@ error nodelay(stream_socket x, bool new_value) {
return none; return none;
} }
variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size) { variant<size_t, sec> read(stream_socket x, span<byte> buf) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf), buf_size, auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
no_sigpipe_io_flag); buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res); return check_stream_socket_io_res(res);
} }
variant<size_t, sec> write(stream_socket x, const void* buf, size_t buf_size) { variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf), buf_size, auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
no_sigpipe_io_flag); buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res); return check_stream_socket_io_res(res);
} }
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -91,7 +92,7 @@ public: ...@@ -91,7 +92,7 @@ public:
template <class Manager> template <class Manager>
bool handle_read_event(Manager&) { bool handle_read_event(Manager&) {
auto res = read(handle_, read_buf_.data(), read_buf_.size()); auto res = read(handle_, make_span(read_buf_));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
data_->insert(data_->end(), read_buf_.begin(), data_->insert(data_->end(), read_buf_.begin(),
read_buf_.begin() + *num_bytes); read_buf_.begin() + *num_bytes);
...@@ -106,7 +107,7 @@ public: ...@@ -106,7 +107,7 @@ public:
auto& payload = x->payload; auto& payload = x->payload;
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
auto res = write(handle_, write_buf_.data(), write_buf_.size()); auto res = write(handle_, as_bytes(make_span(write_buf_)));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes); write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes);
return write_buf_.size() > 0; return write_buf_.size() > 0;
...@@ -156,7 +157,7 @@ CAF_TEST(send and receive) { ...@@ -156,7 +157,7 @@ CAF_TEST(send and receive) {
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
...@@ -165,15 +166,15 @@ CAF_TEST(send and receive) { ...@@ -165,15 +166,15 @@ CAF_TEST(send and receive) {
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(), CAF_CHECK_EQUAL(write(sockets.second,
hello_manager.size()), as_bytes(make_span(hello_manager.data(),
hello_manager.size()))),
hello_manager.size()); hello_manager.size());
run(); run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()), buf->size()),
hello_manager); hello_manager);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
hello_test.size());
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(read_buf.data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(read_buf.data()),
hello_test.size()), hello_test.size()),
hello_test); hello_test);
...@@ -191,8 +192,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -191,8 +192,7 @@ CAF_TEST(resolve and proxy communication) {
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates(); mpx->handle_updates();
run(); run();
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
hello_test.size());
mgr->resolve("/id/42", self); mgr->resolve("/id/42", self);
run(); run();
self->receive( self->receive(
...@@ -203,7 +203,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -203,7 +203,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >> after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(sockets.second, read_buf.data(), read_buf.size()); auto read_res = read(sockets.second, make_span(read_buf));
if (!holds_alternative<size_t>(read_res)) { if (!holds_alternative<size_t>(read_res)) {
CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res))); CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res)));
return; return;
......
...@@ -28,8 +28,10 @@ ...@@ -28,8 +28,10 @@
#include <tuple> #include <tuple>
#include <vector> #include <vector>
#include "caf/byte.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -58,7 +60,8 @@ public: ...@@ -58,7 +60,8 @@ public:
bool handle_read_event() override { bool handle_read_event() override {
if (read_capacity() < 1024) if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048); rd_buf_.resize(rd_buf_.size() + 2048);
auto res = read(handle(), read_position_begin(), read_capacity()); auto res = read(handle(),
make_span(read_position_begin(), read_capacity()));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0); CAF_ASSERT(*num_bytes > 0);
rd_buf_pos_ += *num_bytes; rd_buf_pos_ += *num_bytes;
...@@ -70,7 +73,7 @@ public: ...@@ -70,7 +73,7 @@ public:
bool handle_write_event() override { bool handle_write_event() override {
if (wr_buf_.size() == 0) if (wr_buf_.size() == 0)
return false; return false;
auto res = write(handle(), wr_buf_.data(), wr_buf_.size()); auto res = write(handle(), as_bytes(make_span(wr_buf_)));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0); CAF_ASSERT(*num_bytes > 0);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes); wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
...@@ -84,21 +87,23 @@ public: ...@@ -84,21 +87,23 @@ public:
} }
void send(string_view x) { void send(string_view x) {
wr_buf_.insert(wr_buf_.end(), x.begin(), x.end()); wr_buf_.insert(wr_buf_.end(), reinterpret_cast<const byte*>(x.begin()),
reinterpret_cast<const byte*>(x.end()));
} }
std::string receive() { std::string receive() {
std::string result(rd_buf_.data(), read_position_begin()); std::string result(reinterpret_cast<char*>(rd_buf_.data()),
reinterpret_cast<char*>(read_position_begin()));
rd_buf_pos_ = 0; rd_buf_pos_ = 0;
return result; return result;
} }
private: private:
char* read_position_begin() { byte* read_position_begin() {
return rd_buf_.data() + rd_buf_pos_; return rd_buf_.data() + rd_buf_pos_;
} }
char* read_position_end() { byte* read_position_end() {
return rd_buf_.data() + rd_buf_.size(); return rd_buf_.data() + rd_buf_.size();
} }
...@@ -110,9 +115,9 @@ private: ...@@ -110,9 +115,9 @@ private:
size_t rd_buf_pos_; size_t rd_buf_pos_;
std::vector<char> wr_buf_; std::vector<byte> wr_buf_;
std::vector<char> rd_buf_; std::vector<byte> rd_buf_;
}; };
using dummy_manager_ptr = intrusive_ptr<dummy_manager>; using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
......
...@@ -22,9 +22,12 @@ ...@@ -22,9 +22,12 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector> #include <vector>
#include "host_fixture.hpp" #include "caf/byte.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -32,16 +35,16 @@ using namespace caf::net; ...@@ -32,16 +35,16 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(pipe_socket_tests, host_fixture) CAF_TEST_FIXTURE_SCOPE(pipe_socket_tests, host_fixture)
CAF_TEST(send and receive) { CAF_TEST(send and receive) {
std::vector<char> send_buf{1, 2, 3, 4, 5, 6, 7, 8}; std::vector<byte> send_buf{byte(1), byte(2), byte(3), byte(4),
std::vector<char> receive_buf; byte(5), byte(6), byte(7), byte(8)};
std::vector<byte> receive_buf;
receive_buf.resize(100); receive_buf.resize(100);
pipe_socket rd_sock; pipe_socket rd_sock;
pipe_socket wr_sock; pipe_socket wr_sock;
std::tie(rd_sock, wr_sock) = unbox(make_pipe()); std::tie(rd_sock, wr_sock) = unbox(make_pipe());
CAF_CHECK_EQUAL(write(wr_sock, send_buf.data(), send_buf.size()), CAF_CHECK_EQUAL(write(wr_sock, as_bytes(make_span(send_buf))),
send_buf.size());
CAF_CHECK_EQUAL(read(rd_sock, receive_buf.data(), receive_buf.size()),
send_buf.size()); send_buf.size());
CAF_CHECK_EQUAL(read(rd_sock, make_span(receive_buf)), send_buf.size());
CAF_CHECK(std::equal(send_buf.begin(), send_buf.end(), receive_buf.begin())); CAF_CHECK(std::equal(send_buf.begin(), send_buf.end(), receive_buf.begin()));
} }
......
...@@ -20,6 +20,10 @@ ...@@ -20,6 +20,10 @@
#include "caf/policy/scribe.hpp" #include "caf/policy/scribe.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
...@@ -30,8 +34,7 @@ ...@@ -30,8 +34,7 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/test/dsl.hpp" #include "caf/span.hpp"
#include "host_fixture.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -76,7 +79,7 @@ public: ...@@ -76,7 +79,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_data(Parent&, span<byte> data) { void handle_data(Parent&, span<const byte> data) {
rec_buf_->clear(); rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end()); rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
} }
...@@ -125,7 +128,7 @@ CAF_TEST(receive) { ...@@ -125,7 +128,7 @@ CAF_TEST(receive) {
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy"); CAF_MESSAGE("configure scribe_policy");
...@@ -136,8 +139,9 @@ CAF_TEST(receive) { ...@@ -136,8 +139,9 @@ CAF_TEST(receive) {
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_MESSAGE("sending data to scribe_policy"); CAF_MESSAGE("sending data to scribe_policy");
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(), CAF_CHECK_EQUAL(write(sockets.second,
hello_manager.size()), as_bytes(make_span(hello_manager.data(),
hello_manager.size()))),
hello_manager.size()); hello_manager.size());
run(); run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()), CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
...@@ -166,7 +170,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -166,7 +170,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >> after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); }); [&] { CAF_FAIL("manager did not respond with a proxy."); });
run(); run();
auto read_res = read(sockets.second, read_buf.data(), read_buf.size()); auto read_res = read(sockets.second, make_span(read_buf));
if (!holds_alternative<size_t>(read_res)) if (!holds_alternative<size_t>(read_res))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res))); CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
read_buf.resize(get<size_t>(read_res)); read_buf.resize(get<size_t>(read_res));
......
...@@ -24,6 +24,9 @@ ...@@ -24,6 +24,9 @@
#include "host_fixture.hpp" #include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -37,8 +40,9 @@ CAF_TEST(invalid socket) { ...@@ -37,8 +40,9 @@ CAF_TEST(invalid socket) {
} }
CAF_TEST(connected socket pair) { CAF_TEST(connected socket pair) {
std::vector<char> wr_buf{1, 2, 4, 8, 16, 32, 64}; std::vector<byte> wr_buf{byte(1), byte(2), byte(4), byte(8),
std::vector<char> rd_buf(124); byte(16), byte(32), byte(64)};
std::vector<byte> rd_buf(124);
CAF_MESSAGE("create sockets and configure nonblocking I/O"); CAF_MESSAGE("create sockets and configure nonblocking I/O");
auto x = unbox(make_stream_socket_pair()); auto x = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none); CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none);
...@@ -46,24 +50,23 @@ CAF_TEST(connected socket pair) { ...@@ -46,24 +50,23 @@ CAF_TEST(connected socket pair) {
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u); CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u); CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication"); CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), CAF_CHECK_EQUAL(read(x.first, make_span(rd_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("transfer data from first to second socket"); 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(write(x.first, as_bytes(make_span(wr_buf))), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), wr_buf.size()); CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin())); CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0); rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("transfer data from second to first socket"); 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(write(x.second, as_bytes(make_span(wr_buf))), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, rd_buf.data(), rd_buf.size()), wr_buf.size()); CAF_CHECK_EQUAL(read(x.first, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin())); CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), 0); rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("shut down first socket and observe shutdown on the second one"); CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first); close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()), CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), sec::socket_disconnected);
sec::socket_disconnected);
CAF_MESSAGE("done (cleanup)"); CAF_MESSAGE("done (cleanup)");
close(x.second); close(x.second);
} }
......
...@@ -18,6 +18,12 @@ ...@@ -18,6 +18,12 @@
#define CAF_SUITE string_application #define CAF_SUITE string_application
#include "caf/policy/scribe.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector> #include <vector>
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
...@@ -29,10 +35,8 @@ ...@@ -29,10 +35,8 @@
#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/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/policy/scribe.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/test/dsl.hpp" #include "caf/span.hpp"
#include "host_fixture.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -81,7 +85,7 @@ public: ...@@ -81,7 +85,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_packet(Parent&, header_type&, span<byte> payload) { void handle_packet(Parent&, header_type&, span<const byte> payload) {
binary_deserializer source{sys_, payload}; binary_deserializer source{sys_, payload};
message msg; message msg;
if (auto err = msg.load(source)) if (auto err = msg.load(source))
...@@ -89,8 +93,7 @@ public: ...@@ -89,8 +93,7 @@ public:
if (!msg.match_elements<std::string>()) if (!msg.match_elements<std::string>())
CAF_FAIL("unexpected message: " << msg); CAF_FAIL("unexpected message: " << msg);
auto& str = msg.get_as<std::string>(0); auto& str = msg.get_as<std::string>(0);
auto str_ptr = str.data(); auto byte_ptr = reinterpret_cast<const byte*>(str.data());
auto byte_ptr = reinterpret_cast<const byte*>(str_ptr);
buf_->insert(buf_->end(), byte_ptr, byte_ptr + str.size()); buf_->insert(buf_->end(), byte_ptr, byte_ptr + str.size());
} }
...@@ -135,7 +138,7 @@ public: ...@@ -135,7 +138,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<byte> data) { void handle_data(Parent& parent, span<const byte> data) {
if (await_payload_) { if (await_payload_) {
handle_packet(parent, header_, data); handle_packet(parent, header_, data);
await_payload_ = false; await_payload_ = false;
...@@ -144,7 +147,7 @@ public: ...@@ -144,7 +147,7 @@ public:
CAF_FAIL(""); CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type)); memcpy(&header_, data.data(), sizeof(header_type));
if (header_.payload == 0) if (header_.payload == 0)
handle_packet(parent, header_, span<byte>{}); handle_packet(parent, header_, span<const byte>{});
else else
parent.configure_read(net::receive_policy::exactly(header_.payload)); parent.configure_read(net::receive_policy::exactly(header_.payload));
await_payload_ = true; await_payload_ = true;
...@@ -188,7 +191,7 @@ CAF_TEST(receive) { ...@@ -188,7 +191,7 @@ CAF_TEST(receive) {
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)),
sec::unavailable_or_would_block); sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers"); CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first}, auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.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