Commit f96b992e authored by Jakob Otto's avatar Jakob Otto

Refactor code to span/byte

parent 9ccb5dc6
......@@ -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 ------------------------------------------
......
......@@ -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 {
......@@ -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));
......@@ -128,15 +130,15 @@ public:
void configure_read(net::receive_policy::config cfg);
void write_packet(span<char> buf);
void write_packet(span<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);
......
......@@ -69,7 +69,7 @@ void scribe::configure_read(net::receive_policy::config cfg) {
prepare_next_read();
}
void scribe::write_packet(span<char> buf) {
void scribe::write_packet(const span<byte> buf) {
write_buf_.insert(write_buf_.end(), buf.begin(), buf.end());
}
......
......@@ -24,13 +24,14 @@
#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"
using namespace caf;
using namespace caf::net;
......@@ -58,10 +59,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 +71,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,7 +82,9 @@ public:
template <class Manager>
error init(Manager& manager) {
write_buf_.insert(write_buf_.end(), hello_test.begin(), hello_test.end());
write_buf_.insert(write_buf_.end(),
reinterpret_cast<const byte*>(hello_test.begin()),
reinterpret_cast<const byte*>(hello_test.end()));
CAF_CHECK(manager.mask_add(operation::read_write));
return none;
}
......@@ -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,9 +151,9 @@ 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()),
......@@ -166,15 +169,19 @@ CAF_TEST(send and receive) {
hello_manager.size()),
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_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*>(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); });
......
......@@ -20,11 +20,8 @@
#include "caf/policy/scribe.hpp"
#include "caf/test/dsl.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/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -32,6 +29,9 @@
#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/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
......@@ -57,7 +57,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 +76,7 @@ public:
}
template <class Parent>
void handle_data(Parent&, span<char> data) {
void handle_data(Parent&, span<byte> data) {
rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
}
......@@ -102,17 +102,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,9 +120,9 @@ 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()),
......@@ -140,12 +140,14 @@ CAF_TEST(receive) {
hello_manager.size()),
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); });
......
......@@ -18,21 +18,21 @@
#define CAF_SUITE string_application
#include "caf/net/endpoint_manager.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/policy/scribe.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
......@@ -40,10 +40,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 +70,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 +81,23 @@ 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<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 str_ptr = str.data();
auto byte_ptr = reinterpret_cast<const byte*>(str_ptr);
buf_->insert(buf_->end(), byte_ptr, byte_ptr + str.size());
}
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 +105,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,13 +116,13 @@ public:
private:
actor_system& sys_;
std::shared_ptr<std::vector<char>> buf_;
std::shared_ptr<std::vector<byte>> buf_;
};
class stream_string_application : public string_application {
public:
stream_string_application(actor_system& sys,
std::shared_ptr<std::vector<char>> buf)
std::shared_ptr<std::vector<byte>> buf)
: string_application(sys, std::move(buf)), await_payload_(false) {
// nop
}
......@@ -137,7 +135,7 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<char> data) {
void handle_data(Parent& parent, span<byte> data) {
if (await_payload_) {
handle_packet(parent, header_, data);
await_payload_ = false;
......@@ -146,7 +144,7 @@ public:
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
if (header_.payload == 0)
handle_packet(parent, header_, span<char>{});
handle_packet(parent, header_, span<byte>{});
else
parent.configure_read(net::receive_policy::exactly(header_.payload));
await_payload_ = true;
......@@ -185,9 +183,9 @@ 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()),
......@@ -214,7 +212,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