Commit ac6ed000 authored by Jakob Otto's avatar Jakob Otto

Refactor std::vector<byte>

parent 2e9fe41b
......@@ -55,8 +55,6 @@ class application {
public:
// -- member types -----------------------------------------------------------
using buffer_type = std::vector<byte>;
using byte_span = span<const byte>;
using hub_type = detail::worker_hub<worker>;
......@@ -175,7 +173,7 @@ private:
byte_span received);
/// Writes the handshake payload to `buf_`.
error generate_handshake(buffer_type& buf);
error generate_handshake(byte_buffer& buf);
// -- member variables -------------------------------------------------------
......
......@@ -78,7 +78,7 @@ std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
void to_bytes(header x, std::vector<byte>& buf);
void to_bytes(header x, byte_buffer& buf);
/// @relates header
template <class Inspector>
......
......@@ -20,7 +20,6 @@
#include <atomic>
#include <cstdint>
#include <vector>
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
......@@ -48,8 +47,6 @@ public:
using scheduler_type = scheduler::abstract_coordinator;
using buffer_type = std::vector<byte>;
using hub_type = detail::worker_hub<worker>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -72,9 +69,8 @@ private:
// -- constants and assertions -----------------------------------------------
/// Stores how many bytes the "first half" of this object requires.
static constexpr size_t pointer_members_size = sizeof(hub_type*)
+ sizeof(message_queue*)
+ sizeof(proxy_registry*)
static constexpr size_t pointer_members_size
= sizeof(hub_type*) + sizeof(message_queue*) + sizeof(proxy_registry*)
+ sizeof(actor_system*);
static_assert(CAF_CACHE_LINE_SIZE > pointer_members_size,
......@@ -108,7 +104,7 @@ private:
header hdr_;
/// Contains whatever this worker deserializes next.
buffer_type payload_;
byte_buffer payload_;
};
} // namespace caf::net::basp
......@@ -34,18 +34,18 @@
namespace caf::net {
template <class Factory>
using datagram_transport_base = transport_base<
datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>, udp_datagram_socket,
Factory, ip_endpoint>;
using datagram_transport_base
= transport_base<datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>,
udp_datagram_socket, Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size = std::numeric_limits<
uint16_t>::max();
static constexpr size_t max_datagram_size
= std::numeric_limits<uint16_t>::max();
// -- member types -----------------------------------------------------------
......@@ -57,8 +57,6 @@ public:
using super = datagram_transport_base<factory_type>;
using buffer_type = typename super::buffer_type;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
......@@ -127,7 +125,7 @@ public:
return none;
}
void write_packet(id_type id, span<buffer_type*> buffers) override {
void write_packet(id_type id, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
......@@ -143,7 +141,7 @@ public:
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<buffer_type*> bufs) : id(id) {
packet(id_type id, span<byte_buffer*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
......@@ -151,8 +149,8 @@ public:
}
}
std::vector<std::vector<byte>*> get_buffer_ptrs() {
std::vector<std::vector<byte>*> ptrs;
std::vector<byte_buffer*> get_buffer_ptrs() {
std::vector<byte_buffer*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
......
......@@ -18,8 +18,6 @@
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
......@@ -30,15 +28,13 @@ namespace caf::net {
/// Implements an interface for packet writing in application-layers.
class CAF_NET_EXPORT packet_writer {
public:
using buffer_type = std::vector<byte>;
virtual ~packet_writer();
/// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0;
virtual byte_buffer next_header_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual buffer_type next_payload_buffer() = 0;
virtual byte_buffer next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
......@@ -46,14 +42,14 @@ public:
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...};
byte_buffer* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<buffer_type*> buffers) = 0;
virtual void write_impl(span<byte_buffer*> buffers) = 0;
};
} // namespace caf::net
......@@ -20,8 +20,6 @@
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
namespace caf::net {
/// Implements the interface for transport and application policies and
......@@ -56,11 +54,11 @@ public:
return parent_.manager();
}
buffer_type next_header_buffer() override {
byte_buffer next_header_buffer() override {
return transport().next_header_buffer();
}
buffer_type next_payload_buffer() override {
byte_buffer next_payload_buffer() override {
return transport().next_payload_buffer();
}
......@@ -76,7 +74,7 @@ public:
}
protected:
void write_impl(span<buffer_type*> buffers) override {
void write_impl(span<byte_buffer*> buffers) override {
parent_.write_packet(object_.id(), buffers);
}
......
......@@ -19,7 +19,6 @@
#pragma once
#include <deque>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
......@@ -35,9 +34,9 @@
namespace caf::net {
template <class Application>
using stream_transport_base = transport_base<
stream_transport<Application>, transport_worker<Application>, stream_socket,
Application, unit_t>;
using stream_transport_base
= transport_base<stream_transport<Application>, transport_worker<Application>,
stream_socket, Application, unit_t>;
/// Implements a stream_transport that manages a stream socket.
template <class Application>
......@@ -53,9 +52,7 @@ public:
using id_type = typename super::id_type;
using buffer_type = typename super::buffer_type;
using write_queue_type = std::deque<std::pair<bool, buffer_type>>;
using write_queue_type = std::deque<std::pair<bool, byte_buffer>>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -85,9 +82,8 @@ public:
<< CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) {
if (auto err = this->next_layer_.handle_data(*this,
make_span(
this->read_buf_))) {
if (auto err = this->next_layer_.handle_data(
*this, make_span(this->read_buf_))) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
return false;
}
......@@ -121,7 +117,8 @@ public:
if (is_header) {
if (this->header_bufs_.size() < this->header_bufs_.capacity())
this->header_bufs_.emplace_back(std::move(buf));
} else if (this->payload_bufs_.size() < this->payload_bufs_.capacity()) {
} else if (this->payload_bufs_.size()
< this->payload_bufs_.capacity()) {
this->payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
......@@ -166,7 +163,7 @@ public:
return false;
}
void write_packet(id_type, span<buffer_type*> buffers) override {
void write_packet(id_type, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (this->write_queue_.empty())
......
......@@ -51,9 +51,7 @@ public:
using id_type = IdType;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
using buffer_cache_type = std::vector<byte_buffer>;
// -- constructors, destructors, and assignment operators --------------------
......@@ -191,26 +189,26 @@ public:
/// transport.
/// @param id The id of the destination endpoint.
/// @param buffers Pointers to the buffers that make up the packet content.
virtual void write_packet(id_type id, span<buffer_type*> buffers) = 0;
virtual void write_packet(id_type id, span<byte_buffer*> buffers) = 0;
// -- buffer management ------------------------------------------------------
/// Returns the next cached header buffer or creates a new one if no buffers
/// are cached.
buffer_type next_header_buffer() {
byte_buffer next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
/// Returns the next cached payload buffer or creates a new one if no buffers
/// are cached.
buffer_type next_payload_buffer() {
byte_buffer next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private:
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
static byte_buffer next_buffer_impl(buffer_cache_type cache) {
if (cache.empty())
return {};
auto buf = std::move(cache.back());
......@@ -225,7 +223,7 @@ protected:
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
byte_buffer read_buf_;
endpoint_manager* manager_;
......
......@@ -38,8 +38,9 @@ struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
/// specific port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected<std::pair<udp_datagram_socket, uint16_t>> CAF_NET_EXPORT
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr = false);
expected<std::pair<udp_datagram_socket, uint16_t>>
CAF_NET_EXPORT make_udp_datagram_socket(ip_endpoint ep,
bool reuse_addr = false);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
......@@ -66,7 +67,7 @@ variant<std::pair<size_t, ip_endpoint>, sec>
/// @relates udp_datagram_socket
/// @pre `bufs.size() < 10`
variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
span<std::vector<byte>*> bufs,
span<byte_buffer*> bufs,
ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
......
......@@ -379,7 +379,7 @@ error application::handle_down_message(packet_writer&, header received_hdr,
return none;
}
error application::generate_handshake(std::vector<byte>& buf) {
error application::generate_handshake(byte_buffer& buf) {
binary_serializer sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
......
......@@ -64,7 +64,7 @@ std::array<byte, header_size> to_bytes(header x) {
return result;
}
void to_bytes(header x, std::vector<byte>& buf) {
void to_bytes(header x, byte_buffer& buf) {
buf.resize(header_size);
to_bytes_impl(x, buf.data());
}
......
......@@ -126,11 +126,11 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
#ifdef CAF_WINDOWS
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](std::vector<byte>* buf) {
auto convert = [](byte_buffer* buf) {
return WSABUF{static_cast<ULONG>(buf->size()),
reinterpret_cast<CHAR*>(buf->data())};
};
......@@ -155,10 +155,10 @@ variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
#else // CAF_WINDOWS
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
auto convert = [](std::vector<byte>* buf) {
auto convert = [](byte_buffer* buf) {
return iovec{buf->data(), buf->size()};
};
sockaddr_storage addr = {};
......
......@@ -46,8 +46,6 @@ struct fixture : test_coordinator_fixture<>,
proxy_registry::backend,
basp::application::test_tag,
public packet_writer {
using buffer_type = std::vector<byte>;
fixture() : proxies(sys, *this), app(proxies) {
REQUIRE_OK(app.init(*this));
uri mars_uri;
......@@ -56,8 +54,8 @@ struct fixture : test_coordinator_fixture<>,
}
template <class... Ts>
buffer_type to_buf(const Ts&... xs) {
buffer_type buf;
byte_buffer to_buf(const Ts&... xs) {
byte_buffer buf;
binary_serializer sink{system(), buf};
REQUIRE_OK(sink(xs...));
return buf;
......@@ -111,11 +109,11 @@ struct fixture : test_coordinator_fixture<>,
CAF_FAIL("unexpected function call");
}
buffer_type next_payload_buffer() override {
byte_buffer next_payload_buffer() override {
return {};
}
buffer_type next_header_buffer() override {
byte_buffer next_header_buffer() override {
return {};
}
......@@ -136,14 +134,14 @@ struct fixture : test_coordinator_fixture<>,
}
protected:
void write_impl(span<buffer_type*> buffers) override {
void write_impl(span<byte_buffer*> buffers) override {
for (auto buf : buffers)
output.insert(output.end(), buf->begin(), buf->end());
}
buffer_type input;
byte_buffer input;
buffer_type output;
byte_buffer output;
node_id mars;
......
......@@ -42,16 +42,14 @@ using namespace caf::net::ip;
namespace {
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
constexpr string_view hello_manager = "hello manager!";
class dummy_application_factory;
struct fixture : test_coordinator_fixture<>, host_fixture {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
fixture() : shared_buf(std::make_shared<buffer_type>(1024)) {
fixture() : shared_buf(std::make_shared<byte_buffer>(1024)) {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << err);
......@@ -79,7 +77,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
return mpx->poll_once(false);
}
error read_from_socket(udp_datagram_socket sock, buffer_type& buf) {
error read_from_socket(udp_datagram_socket sock, byte_buffer& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
......@@ -97,19 +95,15 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
multiplexer_ptr mpx;
buffer_ptr shared_buf;
byte_buffer_ptr shared_buf;
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket recv_socket;
};
class dummy_application {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
public:
explicit dummy_application(buffer_ptr rec_buf)
explicit dummy_application(byte_buffer_ptr rec_buf)
: rec_buf_(std::move(rec_buf)){
// nop
};
......@@ -169,16 +163,15 @@ public:
}
private:
buffer_ptr rec_buf_;
byte_buffer_ptr rec_buf_;
};
class dummy_application_factory {
using buffer_ptr = std::shared_ptr<std::vector<byte>>;
public:
using application_type = dummy_application;
explicit dummy_application_factory(buffer_ptr buf) : buf_(std::move(buf)) {
explicit dummy_application_factory(byte_buffer_ptr buf)
: buf_(std::move(buf)) {
// nop
}
......@@ -187,7 +180,7 @@ public:
}
private:
buffer_ptr buf_;
byte_buffer_ptr buf_;
};
} // namespace
......@@ -218,7 +211,7 @@ CAF_TEST(receive) {
CAF_TEST(resolve and proxy communication) {
using transport_type = datagram_transport<dummy_application_factory>;
buffer_type recv_buf(1024);
byte_buffer recv_buf(1024);
auto uri = unbox(make_uri("test:/id/42"));
auto mgr = make_endpoint_manager(
mpx, sys,
......
......@@ -40,6 +40,8 @@ using namespace caf::net;
namespace {
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
string_view hello_manager{"hello manager!"};
string_view hello_test{"hello test!"};
......@@ -69,7 +71,7 @@ class dummy_transport {
public:
using application_type = dummy_application;
dummy_transport(stream_socket handle, std::shared_ptr<std::vector<byte>> data)
dummy_transport(stream_socket handle, byte_buffer_ptr data)
: handle_(handle), data_(data), read_buf_(1024) {
// nop
}
......@@ -146,11 +148,11 @@ public:
private:
stream_socket handle_;
std::shared_ptr<std::vector<byte>> data_;
byte_buffer_ptr data_;
std::vector<byte> read_buf_;
byte_buffer read_buf_;
std::vector<byte> buf_;
byte_buffer buf_;
};
} // namespace
......@@ -158,8 +160,8 @@ private:
CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) {
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
byte_buffer read_buf(1024);
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
......@@ -184,8 +186,8 @@ CAF_TEST(send and receive) {
}
CAF_TEST(resolve and proxy communication) {
std::vector<byte> read_buf(1024);
auto buf = std::make_shared<std::vector<byte>>();
byte_buffer read_buf(1024);
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
......
......@@ -30,7 +30,7 @@ using namespace caf::net;
CAF_TEST(serialization) {
basp::header x{basp::message_type::handshake, 42, 4};
std::vector<byte> buf;
byte_buffer buf;
{
binary_serializer sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none);
......
......@@ -113,9 +113,9 @@ private:
size_t rd_buf_pos_;
std::vector<byte> wr_buf_;
byte_buffer wr_buf_;
std::vector<byte> rd_buf_;
byte_buffer rd_buf_;
};
using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
......
......@@ -107,7 +107,7 @@ CAF_TEST(deliver serialized message) {
CAF_REQUIRE_NOT_EQUAL(hub.peek(), nullptr);
auto w = hub.pop();
CAF_MESSAGE("create a fake message + BASP header");
std::vector<byte> payload;
byte_buffer payload;
std::vector<strong_actor_ptr> stages;
binary_serializer sink{sys, payload};
if (auto err = sink(node_id{}, self->id(), testee.id(), stages,
......
......@@ -23,8 +23,6 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/byte.hpp"
using namespace caf;
......@@ -33,9 +31,9 @@ using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(pipe_socket_tests, host_fixture)
CAF_TEST(send and receive) {
std::vector<byte> send_buf{byte(1), byte(2), byte(3), byte(4),
byte_buffer send_buf{byte(1), byte(2), byte(3), byte(4),
byte(5), byte(6), byte(7), byte(8)};
std::vector<byte> receive_buf;
byte_buffer receive_buf;
receive_buf.resize(100);
pipe_socket rd_sock;
pipe_socket wr_sock;
......
......@@ -46,8 +46,6 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
using transport_type = stream_transport<basp::application>;
size_t fetch_size(variant<size_t, sec> x) {
......@@ -80,8 +78,8 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
}
template <class... Ts>
buffer_type to_buf(const Ts&... xs) {
buffer_type buf;
byte_buffer to_buf(const Ts&... xs) {
byte_buffer buf;
binary_serializer sink{system(), buf};
REQUIRE_OK(sink(xs...));
return buf;
......@@ -108,7 +106,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
}
void consume_handshake() {
buffer_type buf(basp::header_size);
byte_buffer buf(basp::header_size);
if (fetch_size(read(sock, buf)) != basp::header_size)
CAF_FAIL("unable to read " << basp::header_size << " bytes");
auto hdr = basp::header::from_bytes(buf);
......@@ -159,7 +157,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
#define RECEIVE(msg_type, op_data, ...) \
do { \
CAF_MESSAGE("receive " << msg_type); \
buffer_type buf(basp::header_size); \
byte_buffer buf(basp::header_size); \
if (fetch_size(read(sock, buf)) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \
......
......@@ -66,7 +66,7 @@ struct fixture : host_fixture {
stream_socket first;
stream_socket second;
std::vector<byte> rd_buf;
byte_buffer rd_buf;
};
} // namespace
......@@ -79,7 +79,7 @@ CAF_TEST(read on empty sockets) {
}
CAF_TEST(transfer data from first to second socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
byte_buffer wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(first, wr_buf), wr_buf.size());
CAF_CHECK_EQUAL(read(second, rd_buf), wr_buf.size());
......@@ -88,7 +88,7 @@ CAF_TEST(transfer data from first to second socket) {
}
CAF_TEST(transfer data from second to first socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
byte_buffer wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_CHECK_EQUAL(write(second, wr_buf), wr_buf.size());
CAF_CHECK_EQUAL(read(first, rd_buf), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
......@@ -101,9 +101,9 @@ CAF_TEST(shut down first socket and observe shutdown on the second one) {
}
CAF_TEST(transfer data using multiple buffers) {
std::vector<byte> wr_buf_1{1_b, 2_b, 4_b};
std::vector<byte> wr_buf_2{8_b, 16_b, 32_b, 64_b};
std::vector<byte> full_buf;
byte_buffer wr_buf_1{1_b, 2_b, 4_b};
byte_buffer wr_buf_2{8_b, 16_b, 32_b, 64_b};
byte_buffer full_buf;
full_buf.insert(full_buf.end(), wr_buf_1.begin(), wr_buf_1.end());
full_buf.insert(full_buf.end(), wr_buf_2.begin(), wr_buf_2.end());
CAF_CHECK_EQUAL(write(second, {make_span(wr_buf_1), make_span(wr_buf_2)}),
......
......@@ -44,11 +44,9 @@ namespace {
constexpr string_view hello_manager = "hello manager!";
struct fixture : test_coordinator_fixture<>, host_fixture {
using buffer_type = std::vector<byte>;
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
using buffer_ptr = std::shared_ptr<buffer_type>;
fixture() : recv_buf(1024), shared_buf{std::make_shared<buffer_type>()} {
fixture() : recv_buf(1024), shared_buf{std::make_shared<byte_buffer>()} {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << err);
......@@ -66,19 +64,17 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
multiplexer_ptr mpx;
buffer_type recv_buf;
byte_buffer recv_buf;
socket_guard<stream_socket> send_socket_guard;
socket_guard<stream_socket> recv_socket_guard;
buffer_ptr shared_buf;
byte_buffer_ptr shared_buf;
};
class dummy_application {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
public:
dummy_application(buffer_ptr rec_buf)
dummy_application(byte_buffer_ptr rec_buf)
: rec_buf_(std::move(rec_buf)){
// nop
};
......@@ -140,7 +136,7 @@ public:
}
private:
buffer_ptr rec_buf_;
byte_buffer_ptr rec_buf_;
};
} // namespace
......
......@@ -43,7 +43,7 @@ using namespace caf::policy;
namespace {
using buffer_type = std::vector<byte>;
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
......@@ -75,7 +75,7 @@ class string_application {
public:
using header_type = string_application_header;
string_application(std::shared_ptr<std::vector<byte>> buf)
string_application(byte_buffer_ptr buf)
: buf_(std::move(buf)) {
// nop
}
......@@ -117,7 +117,7 @@ public:
}
private:
std::shared_ptr<std::vector<byte>> buf_;
byte_buffer_ptr buf_;
};
template <class Base, class Subtype>
......@@ -125,7 +125,7 @@ class stream_string_application : public Base {
public:
using header_type = typename Base::header_type;
stream_string_application(std::shared_ptr<std::vector<byte>> buf)
stream_string_application(byte_buffer_ptr buf)
: Base(std::move(buf)), await_payload_(false) {
// nop
}
......@@ -204,9 +204,9 @@ CAF_TEST(receive) {
using application_type
= extend<string_application>::with<stream_string_application>;
using transport_type = stream_transport<application_type>;
std::vector<byte> read_buf(1024);
byte_buffer read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
......
......@@ -50,7 +50,7 @@ struct application_result {
};
struct transport_result {
std::vector<byte> packet_buffer;
byte_buffer packet_buffer;
ip_endpoint ep;
};
......
......@@ -34,7 +34,7 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
constexpr string_view hello_test = "hello_test";
......@@ -50,7 +50,7 @@ struct dummy_actor : public monitorable_actor {
class dummy_application {
public:
dummy_application(std::shared_ptr<buffer_type> rec_buf, uint8_t id)
dummy_application(byte_buffer_ptr rec_buf, uint8_t id)
: rec_buf_(std::move(rec_buf)),
id_(id){
// nop
......@@ -68,7 +68,7 @@ public:
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
rec_buf_->push_back(static_cast<byte>(id_));
auto data = ptr->msg->content().get_as<std::vector<byte>>(0);
auto data = ptr->msg->content().get_as<byte_buffer>(0);
parent.write_packet(data);
}
......@@ -93,7 +93,7 @@ public:
}
private:
std::shared_ptr<buffer_type> rec_buf_;
byte_buffer_ptr rec_buf_;
uint8_t id_;
};
......@@ -101,7 +101,7 @@ struct dummy_application_factory {
public:
using application_type = dummy_application;
dummy_application_factory(std::shared_ptr<buffer_type> buf)
dummy_application_factory(byte_buffer_ptr buf)
: buf_(std::move(buf)), application_cnt_(0) {
// nop
}
......@@ -111,7 +111,7 @@ public:
}
private:
std::shared_ptr<buffer_type> buf_;
byte_buffer_ptr buf_;
uint8_t application_cnt_;
};
......@@ -122,13 +122,13 @@ struct dummy_transport {
using application_type = dummy_application;
dummy_transport(actor_system& sys, std::shared_ptr<buffer_type> buf)
dummy_transport(actor_system& sys, byte_buffer_ptr buf)
: sys_(sys), buf_(std::move(buf)) {
// nop
}
template <class IdType>
void write_packet(IdType, span<buffer_type*> buffers) {
void write_packet(IdType, span<byte_buffer*> buffers) {
for (auto buf : buffers)
buf_->insert(buf_->end(), buf->begin(), buf->end());
}
......@@ -141,17 +141,17 @@ struct dummy_transport {
return *this;
}
buffer_type next_header_buffer() {
byte_buffer next_header_buffer() {
return {};
}
buffer_type next_payload_buffer() {
byte_buffer next_payload_buffer() {
return {};
}
private:
actor_system& sys_;
std::shared_ptr<buffer_type> buf_;
byte_buffer_ptr buf_;
};
struct testdata {
......@@ -188,7 +188,7 @@ struct fixture : host_fixture {
= transport_worker_dispatcher<dummy_application_factory, ip_endpoint>;
fixture()
: buf{std::make_shared<buffer_type>()},
: buf{std::make_shared<byte_buffer>()},
dispatcher{dummy_application_factory{buf}},
dummy{sys, buf} {
add_new_workers();
......@@ -235,7 +235,7 @@ struct fixture : host_fixture {
actor_system_config cfg{};
actor_system sys{cfg};
std::shared_ptr<buffer_type> buf;
byte_buffer_ptr buf;
dispatcher_type dispatcher;
dummy_transport dummy;
......
......@@ -62,10 +62,10 @@ struct fixture : host_fixture {
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket receive_socket;
std::vector<byte> buf;
byte_buffer buf;
};
error read_from_socket(udp_datagram_socket sock, std::vector<byte>& buf) {
error read_from_socket(udp_datagram_socket sock, byte_buffer& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
......@@ -115,17 +115,17 @@ CAF_TEST(read / write using span<byte>) {
CAF_CHECK_EQUAL(received, hello_test);
}
CAF_TEST(read / write using span<std::vector<byte>*>) {
CAF_TEST(read / write using span<byte_buffer*>) {
// generate header and payload in separate buffers
header hdr{hello_test.size()};
std::vector<byte> hdr_buf;
byte_buffer hdr_buf;
binary_serializer sink(sys, hdr_buf);
if (auto err = sink(hdr))
CAF_FAIL("serializing payload failed" << err);
auto bytes = as_bytes(make_span(hello_test));
std::vector<byte> payload_buf(bytes.begin(), bytes.end());
byte_buffer payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size();
std::vector<std::vector<byte>*> bufs{&hdr_buf, &payload_buf};
std::vector<byte_buffer*> bufs{&hdr_buf, &payload_buf};
CAF_CHECK_EQUAL(write(send_socket, bufs, ep), packet_size);
// receive both as one single packet.
buf.resize(packet_size);
......
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