Commit f240c5ee authored by Jakob Otto's avatar Jakob Otto

Merge branch master

parents 69c57acf 493ec8b2
...@@ -91,6 +91,16 @@ function(pretty_yes var) ...@@ -91,6 +91,16 @@ function(pretty_yes var)
endif() endif()
endfunction(pretty_yes) endfunction(pretty_yes)
add_executable(caf-generate-enum-strings cmake/caf-generate-enum-strings.cpp)
function(enum_to_string relative_input_file relative_output_file)
set(input "${CMAKE_CURRENT_SOURCE_DIR}/${relative_input_file}")
set(output "${CMAKE_CURRENT_BINARY_DIR}/${relative_output_file}")
add_custom_command(OUTPUT "${output}"
COMMAND caf-generate-enum-strings "${input}" "${output}"
DEPENDS caf-generate-enum-strings "${input}")
endfunction()
# -- binary and library path setup --------------------------------------------- # -- binary and library path setup ---------------------------------------------
# Prohibit in-source builds. # Prohibit in-source builds.
......
#include <algorithm>
#include <cstring>
#include <fstream>
#include <iostream>
#include <string>
#include <vector>
using std::cerr;
using std::find;
using std::find_if;
using std::string;
using std::vector;
void trim(string& str) {
auto not_space = [](char c) { return isspace(c) == 0; };
str.erase(str.begin(), find_if(str.begin(), str.end(), not_space));
str.erase(find_if(str.rbegin(), str.rend(), not_space).base(), str.end());
}
template <size_t N>
bool starts_with(const string& str, const char (&prefix)[N]) {
return str.compare(0, N - 1, prefix) == 0;
}
template <size_t N>
void drop_prefix(string& str, const char (&prefix)[N]) {
if (str.compare(0, N - 1, prefix) == 0)
str.erase(str.begin(), str.begin() + (N - 1));
}
void keep_alnum(string& str) {
auto not_alnum = [](char c) { return isalnum(c) == 0 && c != '_'; };
str.erase(find_if(str.begin(), str.end(), not_alnum), str.end());
}
int main(int argc, char** argv) {
if (argc != 3) {
cerr << "wrong number of arguments.\n"
<< "usage: " << argv[0] << "input-file output-file\n";
return EXIT_FAILURE;
}
std::ifstream in{argv[1]};
if (!in) {
cerr << "unable to open input file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
vector<string> namespaces;
string enum_name;
string line;
bool is_enum_class = false;
// Locate the beginning of the enum.
for (;;) {
if (!getline(in, line)) {
cerr << "unable to locate enum in file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
trim(line);
if (starts_with(line, "enum ")) {
drop_prefix(line, "enum ");
if (starts_with(line, "class ")) {
is_enum_class = true;
drop_prefix(line, "class ");
}
trim(line);
keep_alnum(line);
enum_name = line;
break;
}
if (starts_with(line, "namespace ")) {
if (line.back() == '{')
line.pop_back();
line.erase(line.begin(), find(line.begin(), line.end(), ' '));
trim(line);
namespaces.emplace_back(line);
}
}
// Sanity checking.
if (namespaces.empty()) {
cerr << "enum found outside of a namespace\n";
return EXIT_FAILURE;
}
if (enum_name.empty()) {
cerr << "empty enum name found\n";
return EXIT_FAILURE;
}
std::ofstream out{argv[2]};
if (!out) {
cerr << "unable to open output file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
// Print file header.
out << "#include \"" << namespaces[0];
for (size_t i = 1; i < namespaces.size(); ++i)
out << '/' << namespaces[i];
out << '/' << enum_name << ".hpp\"\n\n"
<< "#include <string>\n\n"
<< "namespace " << namespaces[0] << " {\n";
for (size_t i = 1; i < namespaces.size(); ++i)
out << "namespace " << namespaces[i] << " {\n";
out << "\nstd::string to_string(" << enum_name << " x) {\n"
<< " switch(x) {\n"
<< " default:\n"
<< " return \"???\";\n";
// Read until hitting the closing '}'.
std::string case_label_prefix;
if (is_enum_class)
case_label_prefix = enum_name + "::";
for (;;) {
if (!getline(in, line)) {
cerr << "unable to read enum values\n";
return EXIT_FAILURE;
}
trim(line);
if (line.empty())
continue;
if (line[0] == '}')
break;
if (line[0] != '/') {
keep_alnum(line);
out << " case " << case_label_prefix << line << ":\n"
<< " return \"" << line << "\";\n";
}
}
// Done. Print file footer and exit.
out << " };\n"
<< "}\n\n";
for (auto i = namespaces.rbegin(); i != namespaces.rend(); ++i)
out << "} // namespace " << *i << '\n';
}
...@@ -5,24 +5,33 @@ project(caf_net C CXX) ...@@ -5,24 +5,33 @@ project(caf_net C CXX)
# e.g., for creating proper Xcode projects # e.g., for creating proper Xcode projects
file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp") file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
enum_to_string("caf/net/basp/connection_state.hpp" "basp_conn_strings.cpp")
enum_to_string("caf/net/basp/ec.hpp" "basp_ec_strings.cpp")
enum_to_string("caf/net/basp/message_type.hpp" "basp_message_type_strings.cpp")
enum_to_string("caf/net/operation.hpp" "operation_strings.cpp")
# list cpp files excluding platform-dependent files # list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS set(LIBCAF_NET_SRCS
"${CMAKE_CURRENT_BINARY_DIR}/basp_conn_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_ec_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_message_type_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/operation_strings.cpp"
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp src/application.cpp
src/connection_state.cpp
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp
src/ec.cpp src/ec.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
......
...@@ -89,7 +89,7 @@ public: ...@@ -89,7 +89,7 @@ public:
hub_->add_new_worker(*queue_, proxies_); hub_->add_new_worker(*queue_, proxies_);
// Write handshake. // Write handshake.
auto hdr = parent.next_header_buffer(); auto hdr = parent.next_header_buffer();
auto payload = parent.next_buffer(); auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload)) if (auto err = generate_handshake(payload))
return err; return err;
to_bytes(header{message_type::handshake, to_bytes(header{message_type::handshake,
......
...@@ -27,10 +27,13 @@ namespace defaults { ...@@ -27,10 +27,13 @@ namespace defaults {
namespace middleman { namespace middleman {
extern const size_t max_output_buffers; /// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers; extern const size_t max_header_buffers;
} // namespace middleman } // namespace middleman
} // namespace defaults } // namespace defaults
} // namespace caf } // namespace caf
\ No newline at end of file
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#pragma once #pragma once
#include <string>
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -45,13 +47,7 @@ constexpr operation operator~(operation x) { ...@@ -45,13 +47,7 @@ constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(x)); return static_cast<operation>(~static_cast<int>(x));
} }
constexpr const char* to_string(operation x) { std::string to_string(operation x);
return x == operation::none
? "none"
: (x == operation::read
? "read"
: (x == operation::write ? "write" : "read_write"));
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -32,18 +32,18 @@ class packet_writer { ...@@ -32,18 +32,18 @@ class packet_writer {
public: public:
using buffer_type = std::vector<byte>; using buffer_type = std::vector<byte>;
virtual ~packet_writer() = default; virtual ~packet_writer();
/// Returns the next header_buffer from transport /// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0; virtual buffer_type next_header_buffer() = 0;
/// Returns the next payload_buffer from transport /// Returns a buffer for writing payload content.
virtual buffer_type next_buffer() = 0; virtual buffer_type next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers. /// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers the buffers that make up the packet. first buffer should /// @param buffers all buffers for the packet. The first buffer is a header
/// contain the header. /// buffer, the other buffers are payload buffer.
/// @warning takes ownership of `buffers`. /// @warning this function takes ownership of `buffers`.
template <class... Ts> template <class... Ts>
void write_packet(Ts&... buffers) { void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...}; buffer_type* bufs[] = {&buffers...};
...@@ -51,8 +51,8 @@ public: ...@@ -51,8 +51,8 @@ public:
} }
protected: protected:
/// Actual write_packet implementation. /// Implementing function for `write_packet`.
/// @param buffers `span` containing all buffers of a 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<buffer_type*> buffers) = 0;
}; };
......
...@@ -62,8 +62,8 @@ public: ...@@ -62,8 +62,8 @@ public:
return transport().next_header_buffer(); return transport().next_header_buffer();
} }
buffer_type next_buffer() override { buffer_type next_payload_buffer() override {
return transport().next_buffer(); return transport().next_payload_buffer();
} }
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
......
...@@ -50,6 +50,8 @@ public: ...@@ -50,6 +50,8 @@ public:
using buffer_type = std::vector<byte>; using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application) stream_transport(stream_socket handle, application_type application)
...@@ -92,10 +94,13 @@ public: ...@@ -92,10 +94,13 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
manager_ = &parent; manager_ = &parent;
max_output_bufs_ = get_or(system().config(), "middleman.max-output-buffers", auto& cfg = system().config();
defaults::middleman::max_output_buffers); auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
max_header_bufs_ = get_or(system().config(), "middleman.max-header-buffers", defaults::middleman::max_header_buffers);
defaults::middleman::max_header_buffers); header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = worker_.init(*this)) if (auto err = worker_.init(*this))
return err; return err;
return none; return none;
...@@ -208,65 +213,62 @@ public: ...@@ -208,65 +213,62 @@ public:
} }
void write_packet(unit_t, span<buffer_type*> buffers) { void write_packet(unit_t, span<buffer_type*> buffers) {
// Sanity check
CAF_ASSERT(!buffers.empty()); CAF_ASSERT(!buffers.empty());
auto it = buffers.begin();
if (write_queue_.empty()) if (write_queue_.empty())
manager().register_writing(); manager().register_writing();
// move header by itself to keep things sorted. // By convention, the first buffer is a header buffer. Every other buffer is
write_queue_.emplace_back(true, std::move(**it++)); // a payload buffer.
// payload buffers. just write them auto i = buffers.begin();
for (; it != buffers.end(); ++it) write_queue_.emplace_back(true, std::move(*(*i++)));
write_queue_.emplace_back(false, std::move(**it)); while (i != buffers.end())
write_queue_.emplace_back(false, std::move(*(*i++)));
} }
// -- buffer recycling ------------------------------------------------------- // -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() { buffer_type next_header_buffer() {
return next_buffer_impl(free_header_bufs_); return next_buffer_impl(header_bufs_);
} }
buffer_type next_buffer() { buffer_type next_payload_buffer() {
return next_buffer_impl(free_bufs_); return next_buffer_impl(payload_bufs_);
} }
buffer_type next_buffer_impl(std::deque<buffer_type>& container) { private:
if (container.empty()) { // -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {}; return {};
} else {
auto buf = std::move(container.front());
container.pop_front();
return buf;
} }
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
} }
private:
// -- private member functions -----------------------------------------------
bool write_some() { bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id)); CAF_LOG_TRACE(CAF_ARG(handle_.id));
// helper to sort empty buffers back into the right queues if (write_queue_.empty())
return false;
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = write_queue_.front(); auto& front = write_queue_.front();
auto& is_header = front.first; auto& is_header = front.first;
auto& buf = front.second; auto& buf = front.second;
written_ = 0; written_ = 0;
buf.clear(); buf.clear();
if (is_header && free_header_bufs_.size() < max_header_bufs_) if (is_header) {
free_header_bufs_.emplace_back(std::move(buf)); if (header_bufs_.size() < header_bufs_.capacity())
else if (free_bufs_.size() < max_output_bufs_) header_bufs_.emplace_back(std::move(buf));
free_bufs_.emplace_back(std::move(buf)); } else if (payload_bufs_.size() < payload_bufs_.capacity()) {
payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front(); write_queue_.pop_front();
}; };
// nothing to write // Write buffers from the write_queue_ for as long as possible.
if (write_queue_.empty())
return false;
do { do {
auto& buf = write_queue_.front().second; auto& buf = write_queue_.front().second;
if (buf.empty()) { CAF_ASSERT(!buf.empty());
recycle();
continue;
}
auto data = buf.data() + written_; auto data = buf.data() + written_;
auto len = buf.size() - written_; auto len = buf.size() - written_;
auto write_ret = write(handle_, make_span(data, len)); auto write_ret = write(handle_, make_span(data, len));
...@@ -276,7 +278,7 @@ private: ...@@ -276,7 +278,7 @@ private:
recycle(); recycle();
written_ = 0; written_ = 0;
} else { } else {
written_ = *num_bytes; written_ += *num_bytes;
return false; return false;
} }
} else { } else {
...@@ -295,10 +297,8 @@ private: ...@@ -295,10 +297,8 @@ private:
worker_type worker_; worker_type worker_;
stream_socket handle_; stream_socket handle_;
std::deque<buffer_type> free_header_bufs_; buffer_cache_type header_bufs_;
std::deque<buffer_type> free_bufs_; buffer_cache_type payload_bufs_;
size_t max_output_bufs_;
size_t max_header_bufs_;
buffer_type read_buf_; buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_; std::deque<std::pair<bool, buffer_type>> write_queue_;
...@@ -313,7 +313,7 @@ private: ...@@ -313,7 +313,7 @@ private:
size_t written_; size_t written_;
endpoint_manager* manager_; endpoint_manager* manager_;
}; // namespace net };
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -56,7 +56,7 @@ error application::write_message( ...@@ -56,7 +56,7 @@ error application::write_message(
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr); CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content())); CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_prefix = writer.next_buffer(); auto payload_prefix = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload_prefix}; serializer_impl<buffer_type> sink{system(), payload_prefix};
const auto& src = ptr->msg->sender; const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver; const auto& dst = ptr->receiver;
...@@ -86,7 +86,7 @@ error application::write_message( ...@@ -86,7 +86,7 @@ error application::write_message(
void application::resolve(packet_writer& writer, string_view path, void application::resolve(packet_writer& writer, string_view path,
const actor& listener) { const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener)); CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(path)) { if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err)); CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
...@@ -112,7 +112,7 @@ void application::new_proxy(packet_writer& writer, actor_id id) { ...@@ -112,7 +112,7 @@ void application::new_proxy(packet_writer& writer, actor_id id) {
void application::local_actor_down(packet_writer& writer, actor_id id, void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) { error reason) {
auto payload = writer.next_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload}; serializer_impl<buffer_type> sink{system(), payload};
if (auto err = sink(reason)) if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error"); CAF_RAISE_ERROR("unable to serialize an error");
...@@ -297,7 +297,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -297,7 +297,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size)) if (auto err = source.begin_sequence(path_size))
return err; return err;
// We expect the received to consist only of the path. // We expect the received buffer to contain the path only.
if (path_size != source.remaining()) if (path_size != source.remaining())
return ec::invalid_payload; return ec::invalid_payload;
auto remainder = source.remainder(); auto remainder = source.remainder();
...@@ -314,7 +314,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -314,7 +314,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
aid = 0; aid = 0;
} }
// TODO: figure out how to obtain messaging interface. // TODO: figure out how to obtain messaging interface.
auto payload = writer.next_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (auto err = sink(aid, ifs))
return err; return err;
...@@ -372,7 +372,7 @@ error application::handle_monitor_message(packet_writer& writer, ...@@ -372,7 +372,7 @@ error application::handle_monitor_message(packet_writer& writer,
}); });
} else { } else {
error reason = exit_reason::unknown; error reason = exit_reason::unknown;
auto payload = writer.next_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(reason)) if (auto err = sink(reason))
return err; return err;
......
...@@ -23,7 +23,8 @@ namespace defaults { ...@@ -23,7 +23,8 @@ namespace defaults {
namespace middleman { namespace middleman {
const size_t max_output_buffers = 100; const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10; const size_t max_header_buffers = 10;
} // namespace middleman } // namespace middleman
......
...@@ -26,32 +26,6 @@ namespace caf { ...@@ -26,32 +26,6 @@ namespace caf {
namespace net { namespace net {
namespace basp { namespace basp {
namespace {
string_view ec_names[] = {
"none",
"invalid_magic_number",
"unexpected_number_of_bytes",
"unexpected_payload",
"missing_payload",
"illegal_state",
"invalid_handshake",
"missing_handshake",
"unexpected_handshake",
"version_mismatch",
"unimplemented",
"app_identifiers_mismatch",
"invalid_payload",
"invalid_scheme",
};
} // namespace
std::string to_string(ec x) {
auto result = ec_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
}
error make_error(ec x) { error make_error(ec x) {
return {static_cast<uint8_t>(x), atom("basp")}; return {static_cast<uint8_t>(x), atom("basp")};
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/basp/message_type.hpp"
#include "caf/string_view.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
string_view message_type_names[] = {
"handshake", "actor_message", "resolve_request", "resolve_response",
"monitor_message", "down_message", "heartbeat",
};
} // namespace
std::string to_string(message_type x) {
auto result = message_type_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -16,28 +16,14 @@ ...@@ -16,28 +16,14 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/basp/connection_state.hpp" #include "caf/net/packet_writer.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
namespace basp {
namespace { packet_writer::~packet_writer() {
// nop
const char* connection_state_names[] = {
"await_handshake_header",
"await_handshake_payload",
"await_header",
"await_payload",
"shutdown",
};
} // namespace
std::string to_string(connection_state x) {
return connection_state_names[static_cast<uint8_t>(x)];
} }
} // namespace basp
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -111,7 +111,7 @@ struct fixture : test_coordinator_fixture<>, ...@@ -111,7 +111,7 @@ struct fixture : test_coordinator_fixture<>,
CAF_FAIL("unexpected function call"); CAF_FAIL("unexpected function call");
} }
buffer_type next_buffer() override { buffer_type next_payload_buffer() override {
return {}; return {};
} }
......
...@@ -74,8 +74,7 @@ public: ...@@ -74,8 +74,7 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto header_buf = parent.next_header_buffer(); parent.write_packet(ptr->payload);
parent.write_packet(header_buf, ptr->payload);
} }
template <class Parent> template <class Parent>
......
...@@ -138,11 +138,11 @@ public: ...@@ -138,11 +138,11 @@ public:
return *this; return *this;
} }
std::vector<byte> next_buffer() { std::vector<byte> next_header_buffer() {
return {}; return {};
} }
std::vector<byte> next_header_buffer() { std::vector<byte> next_payload_buffer() {
return {}; return {};
} }
......
...@@ -138,11 +138,11 @@ struct dummy_transport { ...@@ -138,11 +138,11 @@ struct dummy_transport {
return *this; return *this;
} }
buffer_type next_buffer() { buffer_type next_header_buffer() {
return {}; return {};
} }
buffer_type next_header_buffer() { buffer_type next_payload_buffer() {
return {}; return {};
} }
......
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