Commit 696b6a2e authored by Jakob Otto's avatar Jakob Otto

Refactor code to use new binary_serializer

parent b156634c
...@@ -46,7 +46,6 @@ ...@@ -46,7 +46,6 @@
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
......
...@@ -87,7 +87,7 @@ public: ...@@ -87,7 +87,7 @@ public:
CAF_LOG_DEBUG("received " << num_bytes << " bytes"); CAF_LOG_DEBUG("received " << num_bytes << " bytes");
auto ep = res->second; auto ep = res->second;
this->read_buf_.resize(num_bytes); this->read_buf_.resize(num_bytes);
this->next_layer_.handle_data(*this, make_span(this->read_buf_), this->next_layer_.handle_data(*this, as_bytes(make_span(this->read_buf_)),
std::move(ep)); std::move(ep));
prepare_next_read(); prepare_next_read();
} else { } else {
......
...@@ -84,7 +84,9 @@ public: ...@@ -84,7 +84,9 @@ public:
<< CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes)); << CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
this->collected_ += *num_bytes; this->collected_ += *num_bytes;
if (this->collected_ >= this->read_threshold_) { if (this->collected_ >= this->read_threshold_) {
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_)) { if (auto err = this->next_layer_.handle_data(*this,
as_bytes(make_span(
this->read_buf_)))) {
CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err)); CAF_LOG_ERROR("handle_data failed: " << CAF_ARG(err));
return false; return false;
} }
...@@ -184,7 +186,7 @@ private: ...@@ -184,7 +186,7 @@ private:
CAF_ASSERT(!buf.empty()); CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_; auto data = buf.data() + written_;
auto len = buf.size() - written_; auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len)); auto write_ret = write(this->handle(), as_bytes(make_span(data, len)));
if (auto num_bytes = get_if<size_t>(&write_ret)) { if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
written_ += *num_bytes; written_ += *num_bytes;
......
...@@ -62,7 +62,7 @@ public: ...@@ -62,7 +62,7 @@ public:
} }
template <class Parent> template <class Parent>
error handle_data(Parent& parent, span<byte> data, id_type id) { error handle_data(Parent& parent, span<const byte> data, id_type id) {
if (auto worker = find_worker(id)) if (auto worker = find_worker(id))
return worker->handle_data(parent, data); return worker->handle_data(parent, data);
// TODO: Where to get node_id from here? // TODO: Where to get node_id from here?
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
...@@ -37,7 +38,6 @@ ...@@ -37,7 +38,6 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/type_erased_tuple.hpp" #include "caf/type_erased_tuple.hpp"
...@@ -54,7 +54,7 @@ error application::write_message( ...@@ -54,7 +54,7 @@ error application::write_message(
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_payload_buffer(); auto payload_prefix = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload_prefix}; binary_serializer 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;
if (dst == nullptr) { if (dst == nullptr) {
...@@ -84,7 +84,7 @@ void application::resolve(packet_writer& writer, string_view path, ...@@ -84,7 +84,7 @@ 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_payload_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; binary_serializer 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));
return; return;
...@@ -108,7 +108,7 @@ void application::new_proxy(packet_writer& writer, actor_id id) { ...@@ -108,7 +108,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_payload_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload}; binary_serializer 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");
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
...@@ -122,9 +122,9 @@ void application::local_actor_down(packet_writer& writer, actor_id id, ...@@ -122,9 +122,9 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
expected<std::vector<byte>> application::serialize(actor_system& sys, expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
...@@ -307,7 +307,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -307,7 +307,7 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
} }
// TODO: figure out how to obtain messaging interface. // TODO: figure out how to obtain messaging interface.
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (auto err = sink(aid, ifs))
return err; return err;
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
...@@ -363,7 +363,7 @@ error application::handle_monitor_message(packet_writer& writer, ...@@ -363,7 +363,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_payload_buffer(); auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(reason)) if (auto err = sink(reason))
return err; return err;
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
...@@ -389,7 +389,7 @@ error application::handle_down_message(packet_writer&, header received_hdr, ...@@ -389,7 +389,7 @@ error application::handle_down_message(packet_writer&, header received_hdr,
} }
error application::generate_handshake(std::vector<byte>& buf) { error application::generate_handshake(std::vector<byte>& buf) {
serializer_impl<buffer_type> sink{&executor_, buf}; binary_serializer sink{&executor_, buf};
return sink(system().node(), return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers", get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers)); defaults::middleman::app_identifiers));
......
...@@ -274,7 +274,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -274,7 +274,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
{ // Lifetime scope of guard. { // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) if (write_handle_ != invalid_socket)
res = write(write_handle_, make_span(buf)); res = write(write_handle_, as_bytes(make_span(buf)));
else else
res = sec::socket_invalid; res = sec::socket_invalid;
} }
......
...@@ -19,16 +19,15 @@ ...@@ -19,16 +19,15 @@
#include "caf/net/pipe_socket.hpp" #include "caf/net/pipe_socket.hpp"
#include <cstdio> #include <cstdio>
#include <cstdlib>
#include <utility> #include <utility>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/make_message.hpp"
#include "caf/message.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include <cstring>
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
......
...@@ -58,7 +58,7 @@ struct fixture : test_coordinator_fixture<>, ...@@ -58,7 +58,7 @@ struct fixture : test_coordinator_fixture<>,
template <class... Ts> template <class... Ts>
buffer_type to_buf(const Ts&... xs) { buffer_type to_buf(const Ts&... xs) {
buffer_type buf; buffer_type buf;
serializer_impl<buffer_type> sink{system(), buf}; binary_serializer sink{system(), buf};
REQUIRE_OK(sink(xs...)); REQUIRE_OK(sink(xs...));
return buf; return buf;
} }
...@@ -75,16 +75,16 @@ struct fixture : test_coordinator_fixture<>, ...@@ -75,16 +75,16 @@ struct fixture : test_coordinator_fixture<>,
set_input(basp::header{basp::message_type::handshake, set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
basp::version}); basp::version});
REQUIRE_OK(app.handle_data(*this, input)); REQUIRE_OK(app.handle_data(*this, as_bytes(make_span(input))));
CAF_CHECK_EQUAL(app.state(), CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_payload); basp::connection_state::await_handshake_payload);
REQUIRE_OK(app.handle_data(*this, payload)); REQUIRE_OK(app.handle_data(*this, as_bytes(make_span(payload))));
} }
void consume_handshake() { void consume_handshake() {
if (output.size() < basp::header_size) if (output.size() < basp::header_size)
CAF_FAIL("BASP application did not write a handshake header"); CAF_FAIL("BASP application did not write a handshake header");
auto hdr = basp::header::from_bytes(output); auto hdr = basp::header::from_bytes(as_bytes(make_span(output)));
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0 if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version) || hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header"); CAF_FAIL("invalid handshake header");
...@@ -158,10 +158,10 @@ protected: ...@@ -158,10 +158,10 @@ protected:
do { \ do { \
auto payload = to_buf(__VA_ARGS__); \ auto payload = to_buf(__VA_ARGS__); \
set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \ set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
if (auto err = app.handle_data(*this, input)) \ if (auto err = app.handle_data(*this, as_bytes(make_span(input)))) \
CAF_FAIL("application-under-test failed to process header: " \ CAF_FAIL("application-under-test failed to process header: " \
<< sys.render(err)); \ << sys.render(err)); \
if (auto err = app.handle_data(*this, payload)) \ if (auto err = app.handle_data(*this, as_bytes(make_span(payload)))) \
CAF_FAIL("application-under-test failed to process payload: " \ CAF_FAIL("application-under-test failed to process payload: " \
<< sys.render(err)); \ << sys.render(err)); \
} while (false) } while (false)
...@@ -185,19 +185,22 @@ CAF_TEST_FIXTURE_SCOPE(application_tests, fixture) ...@@ -185,19 +185,22 @@ CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(missing handshake) { CAF_TEST(missing handshake) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::heartbeat, 0, 0}); set_input(basp::header{basp::message_type::heartbeat, 0, 0});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::missing_handshake); CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(input))),
basp::ec::missing_handshake);
} }
CAF_TEST(version mismatch) { CAF_TEST(version mismatch) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::handshake, 0, 0}); set_input(basp::header{basp::message_type::handshake, 0, 0});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::version_mismatch); CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(input))),
basp::ec::version_mismatch);
} }
CAF_TEST(missing payload in handshake) { CAF_TEST(missing payload in handshake) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_header);
set_input(basp::header{basp::message_type::handshake, 0, basp::version}); set_input(basp::header{basp::message_type::handshake, 0, basp::version});
CAF_CHECK_EQUAL(app.handle_data(*this, input), basp::ec::missing_payload); CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(input))),
basp::ec::missing_payload);
} }
CAF_TEST(invalid handshake) { CAF_TEST(invalid handshake) {
...@@ -207,9 +210,10 @@ CAF_TEST(invalid handshake) { ...@@ -207,9 +210,10 @@ CAF_TEST(invalid handshake) {
auto payload = to_buf(no_nid, no_ids); auto payload = to_buf(no_nid, no_ids);
set_input(basp::header{basp::message_type::handshake, set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version}); static_cast<uint32_t>(payload.size()), basp::version});
REQUIRE_OK(app.handle_data(*this, input)); REQUIRE_OK(app.handle_data(*this, as_bytes(make_span(input))));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload);
CAF_CHECK_EQUAL(app.handle_data(*this, payload), basp::ec::invalid_handshake); CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(payload))),
basp::ec::invalid_handshake);
} }
CAF_TEST(app identifier mismatch) { CAF_TEST(app identifier mismatch) {
...@@ -218,9 +222,9 @@ CAF_TEST(app identifier mismatch) { ...@@ -218,9 +222,9 @@ CAF_TEST(app identifier mismatch) {
auto payload = to_buf(mars, wrong_ids); auto payload = to_buf(mars, wrong_ids);
set_input(basp::header{basp::message_type::handshake, set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version}); static_cast<uint32_t>(payload.size()), basp::version});
REQUIRE_OK(app.handle_data(*this, input)); REQUIRE_OK(app.handle_data(*this, as_bytes(make_span(input))));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_handshake_payload);
CAF_CHECK_EQUAL(app.handle_data(*this, payload), CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(payload))),
basp::ec::app_identifiers_mismatch); basp::ec::app_identifiers_mismatch);
} }
...@@ -233,8 +237,8 @@ CAF_TEST(repeated handshake) { ...@@ -233,8 +237,8 @@ CAF_TEST(repeated handshake) {
auto payload = to_buf(no_nid, no_ids); auto payload = to_buf(no_nid, no_ids);
set_input(basp::header{basp::message_type::handshake, set_input(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version}); static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app.handle_data(*this, input), none); CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(input))), none);
CAF_CHECK_EQUAL(app.handle_data(*this, payload), CAF_CHECK_EQUAL(app.handle_data(*this, as_bytes(make_span(payload))),
basp::ec::unexpected_handshake); basp::ec::unexpected_handshake);
} }
...@@ -333,7 +337,7 @@ CAF_TEST(heartbeat message) { ...@@ -333,7 +337,7 @@ CAF_TEST(heartbeat message) {
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
auto bytes = to_bytes(basp::header{basp::message_type::heartbeat, 0, 0}); auto bytes = to_bytes(basp::header{basp::message_type::heartbeat, 0, 0});
set_input(bytes); set_input(bytes);
REQUIRE_OK(app.handle_data(*this, input)); REQUIRE_OK(app.handle_data(*this, as_bytes(make_span(input))));
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header); CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
} }
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
...@@ -33,7 +34,6 @@ ...@@ -33,7 +34,6 @@
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/udp_datagram_socket.hpp" #include "caf/net/udp_datagram_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
...@@ -169,9 +169,9 @@ public: ...@@ -169,9 +169,9 @@ public:
static expected<buffer_type> serialize(actor_system& sys, static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
buffer_type result; buffer_type result;
serializer_impl<buffer_type> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
...@@ -30,7 +31,6 @@ ...@@ -30,7 +31,6 @@
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
...@@ -64,9 +64,9 @@ public: ...@@ -64,9 +64,9 @@ public:
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
}; };
...@@ -109,7 +109,7 @@ public: ...@@ -109,7 +109,7 @@ public:
auto& payload = x->payload; auto& payload = x->payload;
buf_.insert(buf_.end(), payload.begin(), payload.end()); buf_.insert(buf_.end(), payload.begin(), payload.end());
} }
auto res = write(handle_, make_span(buf_)); auto res = write(handle_, as_bytes(make_span(buf_)));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
buf_.erase(buf_.begin(), buf_.begin() + *num_bytes); buf_.erase(buf_.begin(), buf_.begin() + *num_bytes);
return buf_.size() > 0; return buf_.size() > 0;
......
...@@ -23,7 +23,7 @@ ...@@ -23,7 +23,7 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/serializer_impl.hpp" #include "caf/binary_serializer.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -32,7 +32,7 @@ CAF_TEST(serialization) { ...@@ -32,7 +32,7 @@ CAF_TEST(serialization) {
basp::header x{basp::message_type::handshake, 42, 4}; basp::header x{basp::message_type::handshake, 42, 4};
std::vector<byte> buf; std::vector<byte> buf;
{ {
serializer_impl<std::vector<byte>> sink{nullptr, buf}; binary_serializer sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none); CAF_CHECK_EQUAL(sink(x), none);
} }
CAF_CHECK_EQUAL(buf.size(), basp::header_size); CAF_CHECK_EQUAL(buf.size(), basp::header_size);
...@@ -45,7 +45,7 @@ CAF_TEST(serialization) { ...@@ -45,7 +45,7 @@ CAF_TEST(serialization) {
CAF_CHECK_EQUAL(source(y), none); CAF_CHECK_EQUAL(source(y), none);
} }
CAF_CHECK_EQUAL(x, y); CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf); auto z = basp::header::from_bytes(as_bytes(make_span(buf)));
CAF_CHECK_EQUAL(x, z); CAF_CHECK_EQUAL(x, z);
CAF_CHECK_EQUAL(y, z); CAF_CHECK_EQUAL(y, z);
} }
......
...@@ -72,7 +72,7 @@ public: ...@@ -72,7 +72,7 @@ public:
bool handle_write_event() override { bool handle_write_event() override {
if (wr_buf_.size() == 0) if (wr_buf_.size() == 0)
return false; return false;
auto res = write(handle(), make_span(wr_buf_)); auto res = write(handle(), as_bytes(make_span(wr_buf_)));
if (auto num_bytes = get_if<size_t>(&res)) { if (auto num_bytes = get_if<size_t>(&res)) {
CAF_ASSERT(*num_bytes > 0); CAF_ASSERT(*num_bytes > 0);
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes); wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + *num_bytes);
......
...@@ -25,10 +25,10 @@ ...@@ -25,10 +25,10 @@
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/basp/message_queue.hpp" #include "caf/net/basp/message_queue.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/serializer_impl.hpp"
using namespace caf; using namespace caf;
...@@ -109,7 +109,7 @@ CAF_TEST(deliver serialized message) { ...@@ -109,7 +109,7 @@ CAF_TEST(deliver serialized message) {
CAF_MESSAGE("create a fake message + BASP header"); CAF_MESSAGE("create a fake message + BASP header");
std::vector<byte> payload; std::vector<byte> payload;
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
serializer_impl<std::vector<byte>> sink{sys, payload}; binary_serializer sink{sys, payload};
if (auto err = sink(node_id{}, self->id(), testee.id(), stages, if (auto err = sink(node_id{}, self->id(), testee.id(), stages,
make_message(ok_atom::value))) make_message(ok_atom::value)))
CAF_FAIL("unable to serialize message: " << sys.render(err)); CAF_FAIL("unable to serialize message: " << sys.render(err));
...@@ -117,7 +117,7 @@ CAF_TEST(deliver serialized message) { ...@@ -117,7 +117,7 @@ CAF_TEST(deliver serialized message) {
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
make_message_id().integer_value()}; make_message_id().integer_value()};
CAF_MESSAGE("launch worker"); CAF_MESSAGE("launch worker");
w->launch(last_hop, hdr, payload); w->launch(last_hop, hdr, as_bytes(make_span(payload)));
sched.run_once(); sched.run_once();
expect((ok_atom), from(_).to(testee)); expect((ok_atom), from(_).to(testee));
} }
......
...@@ -41,7 +41,8 @@ CAF_TEST(send and receive) { ...@@ -41,7 +41,8 @@ CAF_TEST(send and receive) {
pipe_socket rd_sock; pipe_socket rd_sock;
pipe_socket wr_sock; pipe_socket wr_sock;
std::tie(rd_sock, wr_sock) = unbox(make_pipe()); std::tie(rd_sock, wr_sock) = unbox(make_pipe());
CAF_CHECK_EQUAL(write(wr_sock, make_span(send_buf)), send_buf.size()); CAF_CHECK_EQUAL(write(wr_sock, as_bytes(make_span(send_buf))),
send_buf.size());
CAF_CHECK_EQUAL(read(rd_sock, make_span(receive_buf)), send_buf.size()); CAF_CHECK_EQUAL(read(rd_sock, make_span(receive_buf)), send_buf.size());
CAF_CHECK(std::equal(send_buf.begin(), send_buf.end(), receive_buf.begin())); CAF_CHECK(std::equal(send_buf.begin(), send_buf.end(), receive_buf.begin()));
} }
......
...@@ -82,7 +82,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -82,7 +82,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
template <class... Ts> template <class... Ts>
buffer_type to_buf(const Ts&... xs) { buffer_type to_buf(const Ts&... xs) {
buffer_type buf; buffer_type buf;
serializer_impl<buffer_type> sink{system(), buf}; binary_serializer sink{system(), buf};
REQUIRE_OK(sink(xs...)); REQUIRE_OK(sink(xs...));
return buf; return buf;
} }
...@@ -90,7 +90,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -90,7 +90,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
template <class... Ts> template <class... Ts>
void mock(const Ts&... xs) { void mock(const Ts&... xs) {
auto buf = to_buf(xs...); auto buf = to_buf(xs...);
if (fetch_size(write(sock, make_span(buf))) != buf.size()) if (fetch_size(write(sock, as_bytes(make_span(buf)))) != buf.size())
CAF_FAIL("unable to write " << buf.size() << " bytes"); CAF_FAIL("unable to write " << buf.size() << " bytes");
run(); run();
} }
...@@ -103,7 +103,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -103,7 +103,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
static_cast<uint32_t>(payload.size()), basp::version}); static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app->state(), CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_payload); basp::connection_state::await_handshake_payload);
write(sock, make_span(payload)); write(sock, as_bytes(make_span(payload)));
run(); run();
} }
...@@ -111,7 +111,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -111,7 +111,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
buffer_type buf(basp::header_size); buffer_type buf(basp::header_size);
if (fetch_size(read(sock, make_span(buf))) != basp::header_size) if (fetch_size(read(sock, make_span(buf))) != basp::header_size)
CAF_FAIL("unable to read " << basp::header_size << " bytes"); CAF_FAIL("unable to read " << basp::header_size << " bytes");
auto hdr = basp::header::from_bytes(buf); auto hdr = basp::header::from_bytes(as_bytes(make_span(buf)));
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0 if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version) || hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header"); CAF_FAIL("invalid handshake header");
...@@ -149,7 +149,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -149,7 +149,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
std::tuple<unit_t>>::value) { \ std::tuple<unit_t>>::value) { \
auto payload = to_buf(__VA_ARGS__); \ auto payload = to_buf(__VA_ARGS__); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \ mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
write(sock, make_span(payload)); \ write(sock, as_bytes(make_span(payload))); \
} else { \ } else { \
mock(basp::header{kind, 0, op}); \ mock(basp::header{kind, 0, op}); \
} \ } \
...@@ -162,7 +162,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -162,7 +162,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
buffer_type buf(basp::header_size); \ buffer_type buf(basp::header_size); \
if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \ if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \ CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \ auto hdr = basp::header::from_bytes(as_bytes(make_span(buf))); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \ CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \ CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \ if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
......
...@@ -83,7 +83,7 @@ CAF_TEST(read on empty sockets) { ...@@ -83,7 +83,7 @@ CAF_TEST(read on empty sockets) {
CAF_TEST(transfer data from first to second socket) { 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}; std::vector<byte> 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_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(first, make_span(wr_buf)), wr_buf.size()); CAF_CHECK_EQUAL(write(first, as_bytes(make_span(wr_buf))), wr_buf.size());
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), wr_buf.size()); CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin())); CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), byte(0)); rd_buf.assign(rd_buf.size(), byte(0));
...@@ -91,7 +91,7 @@ CAF_TEST(transfer data from first to second socket) { ...@@ -91,7 +91,7 @@ CAF_TEST(transfer data from first to second socket) {
CAF_TEST(transfer data from second to first 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}; std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_CHECK_EQUAL(write(second, make_span(wr_buf)), wr_buf.size()); CAF_CHECK_EQUAL(write(second, as_bytes(make_span(wr_buf))), wr_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), wr_buf.size()); CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin())); CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
} }
...@@ -108,7 +108,8 @@ CAF_TEST(transfer data using multiple buffers) { ...@@ -108,7 +108,8 @@ CAF_TEST(transfer data using multiple buffers) {
std::vector<byte> full_buf; std::vector<byte> full_buf;
full_buf.insert(full_buf.end(), wr_buf_1.begin(), wr_buf_1.end()); 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()); 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)}), CAF_CHECK_EQUAL(write(second, {as_bytes(make_span(wr_buf_1)),
as_bytes(make_span(wr_buf_2))}),
full_buf.size()); full_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), full_buf.size()); CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), full_buf.size());
CAF_CHECK(std::equal(full_buf.begin(), full_buf.end(), rd_buf.begin())); CAF_CHECK(std::equal(full_buf.begin(), full_buf.end(), rd_buf.begin()));
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
...@@ -34,7 +35,6 @@ ...@@ -34,7 +35,6 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
...@@ -140,9 +140,9 @@ public: ...@@ -140,9 +140,9 @@ public:
static expected<buffer_type> serialize(actor_system& sys, static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
buffer_type result; buffer_type result;
serializer_impl<buffer_type> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include <vector> #include <vector>
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
...@@ -34,7 +35,6 @@ ...@@ -34,7 +35,6 @@
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
...@@ -105,18 +105,19 @@ public: ...@@ -105,18 +105,19 @@ public:
if (ptr->msg == nullptr) if (ptr->msg == nullptr)
return; return;
auto header_buf = parent.next_header_buffer(); auto header_buf = parent.next_header_buffer();
serializer_impl<buffer_type> sink{sys_, header_buf}; binary_serializer sink{sys_, header_buf};
header_type header{static_cast<uint32_t>(ptr->payload.size())}; header_type header{static_cast<uint32_t>(ptr->payload.size())};
sink(header); if (auto err = sink(header))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(header_buf, ptr->payload); parent.write_packet(header_buf, ptr->payload);
} }
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
......
...@@ -23,13 +23,13 @@ ...@@ -23,13 +23,13 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
...@@ -105,9 +105,9 @@ public: ...@@ -105,9 +105,9 @@ public:
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; std::vector<byte> result;
serializer_impl<std::vector<byte>> sink{sys, result}; binary_serializer sink{sys, result};
if (auto err = message::save(sink, x)) if (auto err = message::save(sink, x))
return err; return err.value();
return result; return result;
} }
......
...@@ -247,7 +247,7 @@ struct fixture : host_fixture { ...@@ -247,7 +247,7 @@ struct fixture : host_fixture {
}; };
#define CHECK_HANDLE_DATA(testcase) \ #define CHECK_HANDLE_DATA(testcase) \
dispatcher.handle_data(dummy, span<byte>{}, testcase.ep); \ dispatcher.handle_data(dummy, span<const byte>{}, testcase.ep); \
CAF_CHECK_EQUAL(buf->size(), 1u); \ CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \ CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear(); buf->clear();
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/net/test/host_fixture.hpp" #include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp" #include "caf/ip_address.hpp"
...@@ -119,7 +120,7 @@ CAF_TEST(read / write using span<std::vector<byte>*>) { ...@@ -119,7 +120,7 @@ CAF_TEST(read / write using span<std::vector<byte>*>) {
// generate header and payload in separate buffers // generate header and payload in separate buffers
header hdr{hello_test.size()}; header hdr{hello_test.size()};
std::vector<byte> hdr_buf; std::vector<byte> hdr_buf;
serializer_impl<std::vector<byte>> sink(sys, hdr_buf); binary_serializer sink(sys, hdr_buf);
if (auto err = sink(hdr)) if (auto err = sink(hdr))
CAF_FAIL("serializing payload failed" << sys.render(err)); CAF_FAIL("serializing payload failed" << sys.render(err));
auto bytes = as_bytes(make_span(hello_test)); auto bytes = as_bytes(make_span(hello_test));
...@@ -133,7 +134,8 @@ CAF_TEST(read / write using span<std::vector<byte>*>) { ...@@ -133,7 +134,8 @@ CAF_TEST(read / write using span<std::vector<byte>*>) {
CAF_CHECK_EQUAL(buf.size(), packet_size); CAF_CHECK_EQUAL(buf.size(), packet_size);
binary_deserializer source(nullptr, buf); binary_deserializer source(nullptr, buf);
header recv_hdr; header recv_hdr;
source(recv_hdr); if (auto err = source(recv_hdr))
CAF_FAIL("serializing failed: " << err);
CAF_CHECK_EQUAL(hdr.payload_size, recv_hdr.payload_size); CAF_CHECK_EQUAL(hdr.payload_size, recv_hdr.payload_size);
string_view received{reinterpret_cast<const char*>(buf.data()) string_view received{reinterpret_cast<const char*>(buf.data())
+ sizeof(header), + sizeof(header),
......
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