Commit 617cd5ea authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #13

Refactor read and write functions to span
parents 232033c4 827962ee
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <memory> #include <memory>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/byte.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp" #include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
...@@ -42,7 +43,7 @@ public: ...@@ -42,7 +43,7 @@ public:
using super = socket_manager; using super = socket_manager;
/// Represents either an error or a serialized payload. /// 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. /// A function type for serializing message payloads.
using serialize_fun_type = maybe_buffer (*)(actor_system&, using serialize_fun_type = maybe_buffer (*)(actor_system&,
...@@ -90,9 +91,9 @@ public: ...@@ -90,9 +91,9 @@ public:
mailbox_element_ptr msg; mailbox_element_ptr msg;
/// Serialized representation of of `msg->content()`. /// 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 { struct message_policy {
...@@ -134,7 +135,7 @@ public: ...@@ -134,7 +135,7 @@ public:
void resolve(std::string path, actor listener); void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint. /// 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 ------------------------------------------ // -- pure virtual member functions ------------------------------------------
......
...@@ -52,7 +52,7 @@ expected<std::pair<pipe_socket, pipe_socket>> make_pipe(); ...@@ -52,7 +52,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 +60,7 @@ variant<size_t, sec> write(pipe_socket x, const void* buf, size_t buf_size); ...@@ -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. /// @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_;
}; };
......
...@@ -60,7 +60,7 @@ error nodelay(stream_socket x, bool new_value); ...@@ -60,7 +60,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 +69,7 @@ variant<size_t, sec> read(stream_socket x, void* buf, size_t buf_size); ...@@ -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. /// @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.
......
...@@ -18,15 +18,16 @@ ...@@ -18,15 +18,16 @@
#pragma once #pragma once
#include <caf/error.hpp> #include "caf/byte.hpp"
#include <caf/fwd.hpp> #include "caf/error.hpp"
#include <caf/logger.hpp> #include "caf/fwd.hpp"
#include <caf/net/endpoint_manager.hpp> #include "caf/logger.hpp"
#include <caf/net/receive_policy.hpp> #include "caf/net/endpoint_manager.hpp"
#include <caf/net/stream_socket.hpp> #include "caf/net/receive_policy.hpp"
#include <caf/sec.hpp> #include "caf/net/stream_socket.hpp"
#include <caf/span.hpp> #include "caf/sec.hpp"
#include <caf/variant.hpp> #include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
...@@ -49,10 +50,10 @@ public: ...@@ -49,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));
...@@ -75,6 +76,7 @@ public: ...@@ -75,6 +76,7 @@ public:
// Try to write leftover data. // Try to write leftover data.
write_some(parent); write_some(parent);
// Get new data from 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; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
parent.application().write_message(*this, std::move(msg)); parent.application().write_message(*this, std::move(msg));
...@@ -88,9 +90,9 @@ public: ...@@ -88,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_;
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_, 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.
...@@ -128,15 +130,15 @@ public: ...@@ -128,15 +130,15 @@ public:
void configure_read(net::receive_policy::config cfg); void configure_read(net::receive_policy::config cfg);
void write_packet(span<char> buf); void write_packet(span<const byte> buf);
private: private:
net::stream_socket handle_; net::stream_socket handle_;
std::vector<char> read_buf_; std::vector<byte> read_buf_;
std::vector<char> write_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 read_threshold_;
size_t collected_; size_t collected_;
size_t max_; size_t max_;
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp" #include "caf/intrusive/inbox_result.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -36,7 +37,7 @@ endpoint_manager::event::event(atom_value type, uint64_t id) ...@@ -36,7 +37,7 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
} }
endpoint_manager::message::message(mailbox_element_ptr msg, endpoint_manager::message::message(mailbox_element_ptr msg,
std::vector<char> payload) std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) { : msg(std::move(msg)), payload(std::move(payload)) {
// nop // nop
} }
...@@ -86,7 +87,7 @@ void endpoint_manager::resolve(std::string path, actor listener) { ...@@ -86,7 +87,7 @@ void endpoint_manager::resolve(std::string path, actor listener) {
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, 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)); auto ptr = new message(std::move(msg), std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader) if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write); mask_add(operation::write);
......
...@@ -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,7 @@ void multiplexer::update(const socket_manager_ptr& mgr) { ...@@ -117,7 +119,7 @@ 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, 1)));
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,8 @@ pollset_updater::~pollset_updater() { ...@@ -38,7 +39,8 @@ 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() - 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 {
...@@ -69,7 +71,7 @@ void scribe::configure_read(net::receive_policy::config cfg) { ...@@ -69,7 +71,7 @@ void scribe::configure_read(net::receive_policy::config cfg) {
prepare_next_read(); 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()); write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
} }
......
...@@ -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);
} }
......
...@@ -24,13 +24,15 @@ ...@@ -24,13 +24,15 @@
#include "host_fixture.hpp" #include "host_fixture.hpp"
#include "caf/binary_serializer.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#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/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -58,10 +60,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -58,10 +60,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application { class dummy_application {
public: public:
static expected<std::vector<char>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<char> result; std::vector<byte> result;
binary_serializer sink{sys, result}; serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err;
return result; return result;
...@@ -70,7 +72,7 @@ public: ...@@ -70,7 +72,7 @@ public:
class dummy_transport { class dummy_transport {
public: 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) { : handle_(handle), data_(data), read_buf_(1024) {
// nop // nop
} }
...@@ -81,14 +83,15 @@ public: ...@@ -81,14 +83,15 @@ public:
template <class Manager> template <class Manager>
error init(Manager& 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)); CAF_CHECK(manager.mask_add(operation::read_write));
return none; return none;
} }
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);
...@@ -103,7 +106,7 @@ public: ...@@ -103,7 +106,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_, 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;
...@@ -136,11 +139,11 @@ public: ...@@ -136,11 +139,11 @@ public:
private: private:
stream_socket handle_; 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 } // namespace
...@@ -148,12 +151,12 @@ private: ...@@ -148,12 +151,12 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) { 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); 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()); 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,
...@@ -162,19 +165,21 @@ CAF_TEST(send and receive) { ...@@ -162,19 +165,21 @@ 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, as_bytes(make_span(hello_manager))),
hello_manager.size()),
hello_manager.size()); hello_manager.size());
run(); run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager); CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(buf->data()),
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()), buf->size()),
hello_test.size()); hello_manager);
CAF_CHECK_EQUAL(string_view(read_buf.data(), hello_test.size()), hello_test); 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) { CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024); std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>(); 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);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
...@@ -184,8 +189,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -184,8 +189,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(
...@@ -196,7 +200,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -196,7 +200,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(), 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,22 @@ public: ...@@ -84,21 +87,22 @@ public:
} }
void send(string_view x) { 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 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; 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 +114,9 @@ private: ...@@ -110,9 +114,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,15 @@ using namespace caf::net; ...@@ -32,16 +35,15 @@ 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, make_span(send_buf)), send_buf.size());
send_buf.size()); CAF_CHECK_EQUAL(read(rd_sock, make_span(receive_buf)), send_buf.size());
CAF_CHECK_EQUAL(read(rd_sock, receive_buf.data(), receive_buf.size()),
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()));
} }
......
...@@ -24,7 +24,8 @@ ...@@ -24,7 +24,8 @@
#include "host_fixture.hpp" #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/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
...@@ -32,6 +33,8 @@ ...@@ -32,6 +33,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/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -57,7 +60,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -57,7 +60,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application { class dummy_application {
public: 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)){ : rec_buf_(std::move(rec_buf)){
// nop // nop
}; };
...@@ -76,7 +79,7 @@ public: ...@@ -76,7 +79,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_data(Parent&, span<char> 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());
} }
...@@ -102,17 +105,17 @@ public: ...@@ -102,17 +105,17 @@ public:
// nop // nop
} }
static expected<std::vector<char>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<char> result; std::vector<byte> result;
binary_serializer sink{sys, result}; serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err;
return result; return result;
} }
private: private:
std::shared_ptr<std::vector<char>> rec_buf_; std::shared_ptr<std::vector<byte>> rec_buf_;
}; };
} // namespace } // namespace
...@@ -120,12 +123,12 @@ private: ...@@ -120,12 +123,12 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
std::vector<char> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); 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()); 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,16 +139,17 @@ CAF_TEST(receive) { ...@@ -136,16 +139,17 @@ 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, as_bytes(make_span(hello_manager))),
hello_manager.size()),
hello_manager.size()); hello_manager.size());
run(); 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) { CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024); std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>(); 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);
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
...@@ -164,7 +168,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -164,7 +168,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, 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, 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,21 +18,25 @@ ...@@ -18,21 +18,25 @@
#define CAF_SUITE string_application #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 "caf/test/dsl.hpp"
#include "host_fixture.hpp" #include "host_fixture.hpp"
#include <vector>
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#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/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -40,10 +44,6 @@ using namespace caf::policy; ...@@ -40,10 +44,6 @@ using namespace caf::policy;
namespace { namespace {
string_view hello_manager{"hello manager!"};
string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { fixture() {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
...@@ -74,7 +74,7 @@ class string_application { ...@@ -74,7 +74,7 @@ class string_application {
public: public:
using header_type = string_application_header; 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)) { : sys_(sys), buf_(std::move(buf)) {
// nop // nop
} }
...@@ -85,21 +85,22 @@ public: ...@@ -85,21 +85,22 @@ public:
} }
template <class Parent> template <class Parent>
void handle_packet(Parent&, header_type&, span<char> payload) { void handle_packet(Parent&, header_type&, span<const byte> payload) {
binary_deserializer source{sys_, payload.data(), payload.size()}; binary_deserializer source{sys_, payload};
message msg; message msg;
if (auto err = msg.load(source)) if (auto err = msg.load(source))
CAF_FAIL("unable to deserialize message: " << err); CAF_FAIL("unable to deserialize message: " << err);
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);
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> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) { 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())}; header_type header{static_cast<uint32_t>(msg->payload.size())};
buf.resize(sizeof(header_type)); buf.resize(sizeof(header_type));
memcpy(buf.data(), &header, buf.size()); memcpy(buf.data(), &header, buf.size());
...@@ -107,10 +108,10 @@ public: ...@@ -107,10 +108,10 @@ public:
parent.write_packet(buf); 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) { const type_erased_tuple& x) {
std::vector<char> result; std::vector<byte> result;
binary_serializer sink{sys, result}; serializer_impl<std::vector<byte>> sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err;
return result; return result;
...@@ -118,7 +119,7 @@ public: ...@@ -118,7 +119,7 @@ public:
private: private:
actor_system& sys_; actor_system& sys_;
std::shared_ptr<std::vector<char>> buf_; std::shared_ptr<std::vector<byte>> buf_;
}; };
template <class Base, class Subtype> template <class Base, class Subtype>
...@@ -127,7 +128,7 @@ public: ...@@ -127,7 +128,7 @@ public:
using header_type = typename Base::header_type; using header_type = typename Base::header_type;
stream_string_application(actor_system& sys, 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) { : Base(sys, std::move(buf)), await_payload_(false) {
// nop // nop
} }
...@@ -140,7 +141,7 @@ public: ...@@ -140,7 +141,7 @@ public:
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<char> data) { void handle_data(Parent& parent, span<const byte> data) {
if (await_payload_) { if (await_payload_) {
Base::handle_packet(parent, header_, data); Base::handle_packet(parent, header_, data);
await_payload_ = false; await_payload_ = false;
...@@ -149,7 +150,7 @@ public: ...@@ -149,7 +150,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)
Base::handle_packet(parent, header_, span<char>{}); Base::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;
...@@ -190,12 +191,12 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -190,12 +191,12 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
using application_type = extend<string_application>::with< using application_type = extend<string_application>::with<
stream_string_application>; stream_string_application>;
std::vector<char> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); 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()); 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},
...@@ -219,7 +220,9 @@ CAF_TEST(receive) { ...@@ -219,7 +220,9 @@ CAF_TEST(receive) {
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();
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() 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