Unverified Commit e1060543 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #69

Refactor byte_buffer
parents 2e9fe41b 761eed57
......@@ -19,7 +19,6 @@
#pragma once
#include <string>
#include <vector>
#include "caf/attach_stream_source.hpp"
#include "caf/behavior.hpp"
......
......@@ -21,7 +21,6 @@
#include <cstdint>
#include <limits>
#include <string>
#include <vector>
#include "caf/behavior.hpp"
#include "caf/event_based_actor.hpp"
......
......@@ -33,6 +33,7 @@
#include "caf/defaults.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
......@@ -55,8 +56,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 +174,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,11 +20,12 @@
#include <atomic>
#include <cstdint>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/net/basp/remote_message_handler.hpp"
......@@ -48,8 +49,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 +71,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 +106,7 @@ private:
header hdr_;
/// Contains whatever this worker deserializes next.
buffer_type payload_;
byte_buffer payload_;
};
} // namespace caf::net::basp
......@@ -21,6 +21,7 @@
#include <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
......@@ -34,18 +35,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 +58,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 +126,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 +142,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 +150,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/logger.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/socket.hpp"
......@@ -77,10 +75,9 @@ public:
CAF_LOG_DEBUG("unable to get multiplexer from parent");
return false;
}
auto child = make_endpoint_manager(mpx, parent.system(),
stream_transport<
application_type>{*x,
factory_.make()});
auto child = make_endpoint_manager(
mpx, parent.system(),
stream_transport<application_type>{*x, factory_.make()});
if (auto err = child->init())
return false;
return true;
......
......@@ -24,7 +24,6 @@
#include "caf/actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
......
......@@ -18,10 +18,8 @@
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.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,7 +20,8 @@
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/span.hpp"
namespace caf::net {
......@@ -56,11 +57,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 +77,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,8 +19,8 @@
#pragma once
#include <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
......@@ -35,9 +35,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 +53,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 +83,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 +118,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 +164,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())
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/byte_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/overload.hpp"
#include "caf/fwd.hpp"
......@@ -51,9 +52,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 +190,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 +224,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`.
......
......@@ -24,7 +24,7 @@
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp"
......@@ -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",
......
......@@ -18,7 +18,6 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
......
......@@ -21,6 +21,7 @@
#include <cstring>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
......@@ -64,7 +65,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());
}
......
......@@ -50,10 +50,6 @@
# define HOST_NAME_MAX 255
#endif
using std::pair;
using std::string;
using std::vector;
namespace caf::net::ip {
namespace {
......@@ -91,9 +87,8 @@ void for_each_adapter(F f, bool is_link_local = false) {
CAF_LOG_ERROR("failed to get adapter addresses buffer length");
return;
}
auto adapters = adapters_ptr{reinterpret_cast<IP_ADAPTER_ADDRESSES*>(
::malloc(len)),
free};
auto adapters = adapters_ptr{
reinterpret_cast<IP_ADAPTER_ADDRESSES*>(::malloc(len)), free};
if (!adapters) {
CAF_LOG_ERROR("malloc failed");
return;
......
......@@ -19,6 +19,7 @@
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
......@@ -126,11 +127,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 +156,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 = {};
......
......@@ -24,7 +24,7 @@
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
......@@ -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;
......
......@@ -25,6 +25,7 @@
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -42,16 +43,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 +78,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 +96,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 +164,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 +181,7 @@ public:
}
private:
buffer_ptr buf_;
byte_buffer_ptr buf_;
};
} // namespace
......@@ -218,7 +212,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,
......
......@@ -25,7 +25,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -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); });
......
......@@ -24,13 +24,14 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
using namespace caf;
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);
......
......@@ -25,9 +25,9 @@
#include <new>
#include <tuple>
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/span.hpp"
......@@ -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>;
......
......@@ -26,6 +26,7 @@
#include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp"
......@@ -107,7 +108,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,9 +23,8 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
using namespace caf;
using namespace caf::net;
......@@ -33,9 +32,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;
......
......@@ -25,7 +25,7 @@
#include <vector>
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
......@@ -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); \
......
......@@ -24,6 +24,7 @@
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/span.hpp"
using namespace caf;
......@@ -66,7 +67,7 @@ struct fixture : host_fixture {
stream_socket first;
stream_socket second;
std::vector<byte> rd_buf;
byte_buffer rd_buf;
};
} // namespace
......@@ -79,7 +80,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 +89,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 +102,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)}),
......
......@@ -26,6 +26,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -44,11 +45,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 +65,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 +137,7 @@ public:
}
private:
buffer_ptr rec_buf_;
byte_buffer_ptr rec_buf_;
};
} // namespace
......
......@@ -23,11 +23,10 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
......@@ -43,7 +42,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,8 +74,7 @@ class string_application {
public:
using header_type = string_application_header;
string_application(std::shared_ptr<std::vector<byte>> buf)
: buf_(std::move(buf)) {
string_application(byte_buffer_ptr buf) : buf_(std::move(buf)) {
// nop
}
......@@ -117,7 +115,7 @@ public:
}
private:
std::shared_ptr<std::vector<byte>> buf_;
byte_buffer_ptr buf_;
};
template <class Base, class Subtype>
......@@ -125,7 +123,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 +202,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),
......
......@@ -25,6 +25,7 @@
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
......@@ -50,7 +51,7 @@ struct application_result {
};
struct transport_result {
std::vector<byte> packet_buffer;
byte_buffer packet_buffer;
ip_endpoint ep;
};
......
......@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
......@@ -34,7 +35,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 +51,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 +69,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 +94,7 @@ public:
}
private:
std::shared_ptr<buffer_type> rec_buf_;
byte_buffer_ptr rec_buf_;
uint8_t id_;
};
......@@ -101,7 +102,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 +112,7 @@ public:
}
private:
std::shared_ptr<buffer_type> buf_;
byte_buffer_ptr buf_;
uint8_t application_cnt_;
};
......@@ -122,13 +123,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 +142,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 +189,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 +236,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;
......
......@@ -24,6 +24,7 @@
#include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
......@@ -62,10 +63,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 +116,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