Unverified Commit 7f2798df authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #13

Refactor read and write functions to span
parents a0192a41 28415141
......@@ -23,6 +23,7 @@
#include <memory>
#include "caf/actor.hpp"
#include "caf/byte.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
......@@ -42,7 +43,7 @@ public:
using super = socket_manager;
/// Represents either an error or a serialized payload.
using maybe_buffer = expected<std::vector<char>>;
using maybe_buffer = expected<std::vector<byte>>;
/// A function type for serializing message payloads.
using serialize_fun_type = maybe_buffer (*)(actor_system&,
......@@ -90,9 +91,9 @@ public:
mailbox_element_ptr msg;
/// Serialized representation of of `msg->content()`.
std::vector<char> payload;
std::vector<byte> payload;
message(mailbox_element_ptr msg, std::vector<char> payload);
message(mailbox_element_ptr msg, std::vector<byte> payload);
};
struct message_policy {
......@@ -134,7 +135,7 @@ public:
void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<char> payload);
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload);
// -- pure virtual member functions ------------------------------------------
......
......@@ -52,7 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe();
/// @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, 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`.
/// @param x Connected endpoint.
......@@ -60,7 +60,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.
/// @returns The number of received bytes on success, otherwise an error code.
/// @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
/// error code or a non-zero positive integer.
......
......@@ -21,6 +21,7 @@
#include <array>
#include <cstdint>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_manager.hpp"
......@@ -55,7 +56,7 @@ public:
void handle_error(sec code) override;
private:
std::array<char, sizeof(intptr_t)> buf_;
std::array<byte, sizeof(intptr_t)> buf_;
size_t buf_size_;
};
......
......@@ -60,7 +60,7 @@ error nodelay(stream_socket x, bool new_value);
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates pipe_socket
/// @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.
/// @param x Connected endpoint.
......@@ -69,7 +69,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.
/// @relates pipe_socket
/// @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
/// error code or a non-zero positive integer.
......
......@@ -18,15 +18,16 @@
#pragma once
#include <caf/error.hpp>
#include <caf/fwd.hpp>
#include <caf/logger.hpp>
#include <caf/net/endpoint_manager.hpp>
#include <caf/net/receive_policy.hpp>
#include <caf/net/stream_socket.hpp>
#include <caf/sec.hpp>
#include <caf/span.hpp>
#include <caf/variant.hpp>
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace policy {
......@@ -49,10 +50,10 @@ public:
template <class 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_;
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.
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
......@@ -75,6 +76,7 @@ public:
// Try to write leftover data.
write_some(parent);
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg));
......@@ -88,9 +90,9 @@ public:
if (write_buf_.empty())
return false;
auto len = write_buf_.size() - written_;
void* buf = write_buf_.data() + written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = net::write(handle_, buf, len);
auto ret = net::write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
......@@ -128,15 +130,15 @@ public:
void configure_read(net::receive_policy::config cfg);
void write_packet(span<char> buf);
void write_packet(span<const byte> buf);
private:
net::stream_socket handle_;
std::vector<char> read_buf_;
std::vector<char> write_buf_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
size_t max_consecutive_reads_;
size_t max_consecutive_reads_; // TODO use this field!
size_t read_threshold_;
size_t collected_;
size_t max_;
......
......@@ -18,6 +18,7 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
......@@ -36,7 +37,7 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
}
endpoint_manager::message::message(mailbox_element_ptr msg,
std::vector<char> payload)
std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) {
// nop
}
......@@ -86,7 +87,7 @@ void endpoint_manager::resolve(std::string path, actor listener) {
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
std::vector<char> payload) {
std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
......
......@@ -20,6 +20,7 @@
#include <algorithm>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
......@@ -28,6 +29,7 @@
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
#ifndef CAF_WINDOWS
......@@ -117,7 +119,7 @@ void multiplexer::update(const socket_manager_ptr& mgr) {
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, &value, sizeof(intptr_t));
res = write(write_handle_, as_bytes(make_span(&value, 1)));
else
res = sec::socket_invalid;
}
......
......@@ -22,13 +22,16 @@
#include <cstdlib>
#include <utility>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
......@@ -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.
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.
return read(socket_cast<stream_socket>(x), buf, buf_size);
return read(socket_cast<stream_socket>(x), buf);
}
#else // CAF_WINDOWS
......@@ -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]});
}
variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size) {
auto res = ::write(x.id, buf, buf_size);
variant<size_t, sec> write(pipe_socket x, span<const byte> buf) {
auto res = ::write(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res);
}
variant<size_t, sec> read(pipe_socket x, void* buf, size_t buf_size) {
auto res = ::read(x.id, buf, buf_size);
variant<size_t, sec> read(pipe_socket x, span<byte> buf) {
auto res = ::read(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size());
return check_pipe_socket_io_res(res);
}
......
......@@ -20,6 +20,7 @@
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
......@@ -38,7 +39,8 @@ pollset_updater::~pollset_updater() {
bool pollset_updater::handle_read_event() {
for (;;) {
auto res = read(handle(), buf_.data(), buf_.size() - buf_size_);
auto res = read(handle(), make_span(buf_.data() + buf_size_,
buf_.size() - buf_size_));
if (auto num_bytes = get_if<size_t>(&res)) {
buf_size_ += *num_bytes;
if (buf_.size() == buf_size_) {
......
......@@ -20,7 +20,9 @@
#include <system_error>
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/span.hpp"
namespace caf {
namespace policy {
......@@ -69,7 +71,7 @@ void scribe::configure_read(net::receive_policy::config cfg) {
prepare_next_read();
}
void scribe::write_packet(span<char> buf) {
void scribe::write_packet(span<const byte> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
}
......
......@@ -18,11 +18,13 @@
#include "caf/net/stream_socket.hpp"
#include "caf/byte.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/span.hpp"
#include "caf/variant.hpp"
namespace caf {
......@@ -161,15 +163,15 @@ error nodelay(stream_socket x, bool new_value) {
return none;
}
variant<size_t, sec> 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);
variant<size_t, sec> read(stream_socket x, span<byte> buf) {
auto res = ::recv(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
}
variant<size_t, sec> 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);
variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
auto res = ::send(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), no_sigpipe_io_flag);
return check_stream_socket_io_res(res);
}
......
......@@ -24,13 +24,15 @@
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -58,10 +60,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application {
public:
static expected<std::vector<char>> serialize(actor_system& sys,
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
......@@ -70,7 +72,7 @@ public:
class dummy_transport {
public:
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<char>> data)
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data)
: handle_(handle), data_(data), read_buf_(1024) {
// nop
}
......@@ -81,14 +83,15 @@ public:
template <class Manager>
error init(Manager& manager) {
write_buf_.insert(write_buf_.end(), hello_test.begin(), hello_test.end());
auto test_bytes = as_bytes(make_span(hello_test));
write_buf_.insert(write_buf_.end(), test_bytes.begin(), test_bytes.end());
CAF_CHECK(manager.mask_add(operation::read_write));
return none;
}
template <class 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)) {
data_->insert(data_->end(), read_buf_.begin(),
read_buf_.begin() + *num_bytes);
......@@ -103,7 +106,7 @@ public:
auto& payload = x->payload;
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
}
auto res = write(handle_, write_buf_.data(), write_buf_.size());
auto res = write(handle_, make_span(write_buf_));
if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes);
return write_buf_.size() > 0;
......@@ -136,11 +139,11 @@ public:
private:
stream_socket handle_;
std::shared_ptr<std::vector<char>> data_;
std::shared_ptr<std::vector<byte>> data_;
std::vector<char> read_buf_;
std::vector<byte> read_buf_;
std::vector<char> write_buf_;
std::vector<byte> write_buf_;
};
} // namespace
......@@ -148,12 +151,12 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) {
std::vector<char> read_buf(1024);
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
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);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
......@@ -162,19 +165,21 @@ CAF_TEST(send and receive) {
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()),
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size());
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
hello_test.size());
CAF_CHECK_EQUAL(string_view(read_buf.data(), hello_test.size()), hello_test);
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()),
hello_manager);
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(read_buf.data()),
hello_test.size()),
hello_test);
}
CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>();
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
......@@ -184,8 +189,7 @@ CAF_TEST(resolve and proxy communication) {
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
hello_test.size());
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
mgr->resolve("/id/42", self);
run();
self->receive(
......@@ -196,7 +200,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
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)) {
CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res)));
return;
......
......@@ -28,8 +28,10 @@
#include <tuple>
#include <vector>
#include "caf/byte.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -58,7 +60,8 @@ public:
bool handle_read_event() override {
if (read_capacity() < 1024)
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)) {
CAF_ASSERT(*num_bytes > 0);
rd_buf_pos_ += *num_bytes;
......@@ -70,7 +73,7 @@ public:
bool handle_write_event() override {
if (wr_buf_.size() == 0)
return false;
auto res = write(handle(), wr_buf_.data(), wr_buf_.size());
auto res = write(handle(), make_span(wr_buf_));
if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
......@@ -84,21 +87,22 @@ public:
}
void send(string_view x) {
wr_buf_.insert(wr_buf_.end(), x.begin(), x.end());
auto x_bytes = as_bytes(make_span(x));
wr_buf_.insert(wr_buf_.end(), x_bytes.begin(), x_bytes.end());
}
std::string receive() {
std::string result(rd_buf_.data(), read_position_begin());
std::string result(reinterpret_cast<char*>(rd_buf_.data()), rd_buf_pos_);
rd_buf_pos_ = 0;
return result;
}
private:
char* read_position_begin() {
byte* read_position_begin() {
return rd_buf_.data() + rd_buf_pos_;
}
char* read_position_end() {
byte* read_position_end() {
return rd_buf_.data() + rd_buf_.size();
}
......@@ -110,9 +114,9 @@ private:
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>;
......
......@@ -22,9 +22,12 @@
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -32,16 +35,15 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(pipe_socket_tests, host_fixture)
CAF_TEST(send and receive) {
std::vector<char> send_buf{1, 2, 3, 4, 5, 6, 7, 8};
std::vector<char> receive_buf;
std::vector<byte> send_buf{byte(1), byte(2), byte(3), byte(4),
byte(5), byte(6), byte(7), byte(8)};
std::vector<byte> receive_buf;
receive_buf.resize(100);
pipe_socket rd_sock;
pipe_socket wr_sock;
std::tie(rd_sock, wr_sock) = unbox(make_pipe());
CAF_CHECK_EQUAL(write(wr_sock, send_buf.data(), send_buf.size()),
send_buf.size());
CAF_CHECK_EQUAL(read(rd_sock, receive_buf.data(), receive_buf.size()),
send_buf.size());
CAF_CHECK_EQUAL(write(wr_sock, make_span(send_buf)), 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()));
}
......
......@@ -24,7 +24,8 @@
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -32,6 +33,8 @@
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -57,7 +60,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application {
public:
dummy_application(std::shared_ptr<std::vector<char>> rec_buf)
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf)
: rec_buf_(std::move(rec_buf)){
// nop
};
......@@ -76,7 +79,7 @@ public:
}
template <class Parent>
void handle_data(Parent&, span<char> data) {
void handle_data(Parent&, span<const byte> data) {
rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
}
......@@ -102,17 +105,17 @@ public:
// nop
}
static expected<std::vector<char>> serialize(actor_system& sys,
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
std::shared_ptr<std::vector<char>> rec_buf_;
std::shared_ptr<std::vector<byte>> rec_buf_;
};
} // namespace
......@@ -120,12 +123,12 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
std::vector<char> read_buf(1024);
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
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);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
CAF_MESSAGE("configure scribe_policy");
......@@ -136,16 +139,17 @@ CAF_TEST(receive) {
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_MESSAGE("sending data to scribe_policy");
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()),
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size());
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager);
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()),
hello_manager);
}
CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>();
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
......@@ -164,7 +168,7 @@ CAF_TEST(resolve and proxy communication) {
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
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))
CAF_FAIL("read() returned an error: " << sys.render(get<sec>(read_res)));
read_buf.resize(get<size_t>(read_res));
......
......@@ -24,6 +24,9 @@
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -37,8 +40,9 @@ CAF_TEST(invalid socket) {
}
CAF_TEST(connected socket pair) {
std::vector<char> wr_buf{1, 2, 4, 8, 16, 32, 64};
std::vector<char> rd_buf(124);
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);
......@@ -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.second)), 0u);
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);
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);
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_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(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_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_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_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");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, rd_buf.data(), rd_buf.size()),
sec::socket_disconnected);
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), sec::socket_disconnected);
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
......
......@@ -18,21 +18,25 @@
#define CAF_SUITE string_application
#include "caf/net/endpoint_manager.hpp"
#include <caf/policy/scribe.hpp>
#include "caf/policy/scribe.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
......@@ -40,10 +44,6 @@ using namespace caf::policy;
namespace {
string_view hello_manager{"hello manager!"};
string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
......@@ -74,7 +74,7 @@ class string_application {
public:
using header_type = string_application_header;
string_application(actor_system& sys, std::shared_ptr<std::vector<char>> buf)
string_application(actor_system& sys, std::shared_ptr<std::vector<byte>> buf)
: sys_(sys), buf_(std::move(buf)) {
// nop
}
......@@ -85,21 +85,22 @@ public:
}
template <class Parent>
void handle_packet(Parent&, header_type&, span<char> payload) {
binary_deserializer source{sys_, payload.data(), payload.size()};
void handle_packet(Parent&, header_type&, span<const byte> payload) {
binary_deserializer source{sys_, payload};
message msg;
if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err);
if (!msg.match_elements<std::string>())
CAF_FAIL("unexpected message: " << msg);
auto& str = msg.get_as<std::string>(0);
buf_->insert(buf_->end(), str.begin(), str.end());
auto bytes = as_bytes(make_span(str));
buf_->insert(buf_->end(), bytes.begin(), bytes.end());
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::vector<char> buf;
std::vector<byte> buf;
header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type));
memcpy(buf.data(), &header, buf.size());
......@@ -107,10 +108,10 @@ public:
parent.write_packet(buf);
}
static expected<std::vector<char>> serialize(actor_system& sys,
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
......@@ -118,7 +119,7 @@ public:
private:
actor_system& sys_;
std::shared_ptr<std::vector<char>> buf_;
std::shared_ptr<std::vector<byte>> buf_;
};
template <class Base, class Subtype>
......@@ -127,7 +128,7 @@ public:
using header_type = typename Base::header_type;
stream_string_application(actor_system& sys,
std::shared_ptr<std::vector<char>> buf)
std::shared_ptr<std::vector<byte>> buf)
: Base(sys, std::move(buf)), await_payload_(false) {
// nop
}
......@@ -140,7 +141,7 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<char> data) {
void handle_data(Parent& parent, span<const byte> data) {
if (await_payload_) {
Base::handle_packet(parent, header_, data);
await_payload_ = false;
......@@ -149,7 +150,7 @@ public:
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
if (header_.payload == 0)
Base::handle_packet(parent, header_, span<char>{});
Base::handle_packet(parent, header_, span<const byte>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
await_payload_ = true;
......@@ -190,12 +191,12 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) {
using application_type = extend<string_application>::with<
stream_string_application>;
std::vector<char> read_buf(1024);
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<char>>();
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
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);
CAF_MESSAGE("adding both endpoint managers");
auto mgr1 = make_endpoint_manager(mpx, sys, policy::scribe{sockets.first},
......@@ -219,7 +220,9 @@ CAF_TEST(receive) {
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), "hello proxy!");
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
buf->size()),
"hello proxy!");
}
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