Commit 117513ff authored by Dominik Charousset's avatar Dominik Charousset

Port incubator to latest CAF 0.18 API

parent edc7ca34
...@@ -46,7 +46,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD) ...@@ -46,7 +46,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare( FetchContent_Declare(
actor_framework actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG 9caa83be4 GIT_TAG ac0f23bc
) )
FetchContent_Populate(actor_framework) FetchContent_Populate(actor_framework)
set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE) set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE)
......
diff --git a/libcaf_net/CMakeLists.txt b/libcaf_net/CMakeLists.txt
index 8c98f80..be7333c 100644
--- a/libcaf_net/CMakeLists.txt
+++ b/libcaf_net/CMakeLists.txt
@@ -50,8 +50,8 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp
src/message_queue.cpp
src/multiplexer.cpp
- src/net/backend/test.cpp
- src/net/backend/tcp.cpp
+ #src/net/backend/test.cpp
+ #src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
@@ -124,28 +124,28 @@ target_link_libraries(caf-net-test PRIVATE CAF::test)
caf_incubator_add_test_suites(caf-net-test
accept_socket
- application
+ #application
convert_ip_endpoint
datagram_socket
- datagram_transport
- doorman
- endpoint_manager
+ #datagram_transport
+ #doorman
+ #endpoint_manager
header
ip
multiplexer
- net.backend.tcp
- net.basp.message_queue
- net.basp.ping_pong
- net.basp.worker
+ #net.backend.tcp
+ #net.basp.message_queue
+ #net.basp.ping_pong
+ #net.basp.worker
net.length_prefix_framing
network_socket
pipe_socket
socket
socket_guard
- stream_application
+ #stream_application
stream_socket
- stream_transport
- string_application
+ #stream_transport
+ #string_application
tcp_sockets
transport_worker
transport_worker_dispatcher
diff --git a/libcaf_net/src/net/backend/tcp.cpp b/libcaf_net/src/net/backend/tcp.cpp
index d1d2bd9..e84b011 100644
--- a/libcaf_net/src/net/backend/tcp.cpp
+++ b/libcaf_net/src/net/backend/tcp.cpp
@@ -25,6 +25,7 @@
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp"
+#include "caf/net/defaults.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
diff --git a/libcaf_net/src/udp_datagram_socket.cpp b/libcaf_net/src/udp_datagram_socket.cpp
index d6362f3..58fe231 100644
--- a/libcaf_net/src/udp_datagram_socket.cpp
+++ b/libcaf_net/src/udp_datagram_socket.cpp
@@ -114,8 +114,9 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
- auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
- : sizeof(sockaddr_in6);
+ auto len = static_cast<socklen_t>(ep.address().embeds_v4()
+ ? sizeof(sockaddr_in)
+ : sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res);
@@ -172,7 +173,7 @@ variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
message.msg_iov = buf_array;
- message.msg_iovlen = bufs.size();
+ message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res);
}
...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf); ...@@ -83,9 +83,10 @@ CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf);
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) { bool inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len, return f.object(x).fields(f.field("type", x.type),
x.operation_data); f.field("payload_len", x.payload_len),
f.field("operation_data", x.operation_data));
} }
/// @} /// @}
......
...@@ -53,7 +53,8 @@ public: ...@@ -53,7 +53,8 @@ public:
std::vector<strong_actor_ptr> fwd_stack; std::vector<strong_actor_ptr> fwd_stack;
message content; message content;
binary_deserializer source{ctx, payload}; binary_deserializer source{ctx, payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content)) { if (auto err = source.apply_objects(src_node, src_id, dst_id, fwd_stack,
content)) {
CAF_LOG_ERROR("could not deserialize payload: " << CAF_ARG(err)); CAF_LOG_ERROR("could not deserialize payload: " << CAF_ARG(err));
return; return;
} }
......
...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> { ...@@ -55,9 +55,8 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
/// @relates socket /// @relates socket
template <class Inspector> template <class Inspector>
typename Inspector::result_type CAF_NET_EXPORT inspect(Inspector& f, bool inspect(Inspector& f, socket& x) {
socket& x) { return f.object(x).fields(f.field("id", x.id));
return f(x.id);
} }
/// Denotes the invalid socket. /// Denotes the invalid socket.
......
...@@ -62,14 +62,15 @@ error application::write_message( ...@@ -62,14 +62,15 @@ error application::write_message(
if (src != nullptr) { if (src != nullptr) {
auto src_id = src->id(); auto src_id = src->id();
system().registry().put(src_id, src); system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(src->node(), src_id, dst->id(), ptr->msg->stages))
return err; return sink.get_error();
} else { } else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (!sink.apply_objects(node_id{}, actor_id{0}, dst->id(),
return err; ptr->msg->stages))
return sink.get_error();
} }
if (auto err = sink(ptr->msg->content())) if (!sink.apply_objects(ptr->msg->content()))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message, to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload_buf.size()),
...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path, ...@@ -84,8 +85,8 @@ void application::resolve(packet_writer& writer, string_view path,
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();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(path)) { if (!sink.apply_objects(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err)); CAF_LOG_ERROR("unable to serialize path:" << sink.get_error());
return; return;
} }
auto req_id = next_request_id_++; auto req_id = next_request_id_++;
...@@ -108,7 +109,7 @@ void application::local_actor_down(packet_writer& writer, actor_id id, ...@@ -108,7 +109,7 @@ 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();
binary_serializer sink{system(), payload}; binary_serializer sink{system(), payload};
if (auto err = sink(reason)) if (!sink.apply_objects(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();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr, ...@@ -218,8 +219,8 @@ error application::handle_handshake(packet_writer&, header hdr,
node_id peer_id; node_id peer_id;
std::vector<std::string> app_ids; std::vector<std::string> app_ids;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, payload};
if (auto err = source(peer_id, app_ids)) if (!source.apply_objects(peer_id, app_ids))
return err; return source.get_error();
if (!peer_id || app_ids.empty()) if (!peer_id || app_ids.empty())
return ec::invalid_handshake; return ec::invalid_handshake;
auto ids = get_or(system().config(), "caf.middleman.app-identifiers", auto ids = get_or(system().config(), "caf.middleman.app-identifiers",
...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -277,8 +278,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
CAF_ASSERT(rec_hdr.type == message_type::resolve_request); CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0; size_t path_size = 0;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size)) if (!source.begin_sequence(path_size))
return err; return source.get_error();
// We expect the received buffer to contain the path only. // 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;
...@@ -298,8 +299,8 @@ error application::handle_resolve_request(packet_writer& writer, header rec_hdr, ...@@ -298,8 +299,8 @@ 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();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (!sink.apply_objects(aid, ifs))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response, to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr, ...@@ -323,9 +324,9 @@ error application::handle_resolve_response(packet_writer&, header received_hdr,
actor_id aid; actor_id aid;
std::set<std::string> ifs; std::set<std::string> ifs;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs)) { if (!source.apply_objects(aid, ifs)) {
anon_send(i->second, sec::remote_lookup_failed); anon_send(i->second, sec::remote_lookup_failed);
return err; return source.get_error();
} }
if (aid == 0) { if (aid == 0) {
anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs)); anon_send(i->second, strong_actor_ptr{nullptr}, std::move(ifs));
...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer, ...@@ -354,8 +355,8 @@ error application::handle_monitor_message(packet_writer& writer,
error reason = exit_reason::unknown; error reason = exit_reason::unknown;
auto payload = writer.next_payload_buffer(); auto payload = writer.next_payload_buffer();
binary_serializer sink{&executor_, payload}; binary_serializer sink{&executor_, payload};
if (auto err = sink(reason)) if (!sink.apply_objects(reason))
return err; return sink.get_error();
auto hdr = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr, ...@@ -372,17 +373,20 @@ error application::handle_down_message(packet_writer&, header received_hdr,
<< CAF_ARG2("received.size", received.size())); << CAF_ARG2("received.size", received.size()));
error reason; error reason;
binary_deserializer source{&executor_, received}; binary_deserializer source{&executor_, received};
if (auto err = source(reason)) if (!source.apply_objects(reason))
return err; return source.get_error();
proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason)); proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason));
return none; return none;
} }
error application::generate_handshake(byte_buffer& buf) { error application::generate_handshake(byte_buffer& buf) {
binary_serializer sink{&executor_, buf}; binary_serializer sink{&executor_, buf};
return sink(system().node(), if (!sink.apply_objects(system().node(),
get_or(system().config(), "caf.middleman.app-identifiers", get_or(system().config(),
application::default_app_ids())); "caf.middleman.app-identifiers",
application::default_app_ids())))
return sink.get_error();
return none;
} }
} // namespace caf::net::basp } // namespace caf::net::basp
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/deep_to_string.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
...@@ -34,7 +35,7 @@ CAF_TEST(serialization) { ...@@ -34,7 +35,7 @@ CAF_TEST(serialization) {
byte_buffer buf; byte_buffer buf;
{ {
binary_serializer sink{nullptr, buf}; binary_serializer sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none); CAF_CHECK(sink.apply_object(x));
} }
CAF_CHECK_EQUAL(buf.size(), basp::header_size); CAF_CHECK_EQUAL(buf.size(), basp::header_size);
auto buf2 = to_bytes(x); auto buf2 = to_bytes(x);
...@@ -43,7 +44,7 @@ CAF_TEST(serialization) { ...@@ -43,7 +44,7 @@ CAF_TEST(serialization) {
basp::header y; basp::header y;
{ {
binary_deserializer source{nullptr, buf}; binary_deserializer source{nullptr, buf};
CAF_CHECK_EQUAL(source(y), none); CAF_CHECK(source.apply_object(y));
} }
CAF_CHECK_EQUAL(x, y); CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf); auto z = basp::header::from_bytes(buf);
...@@ -53,5 +54,5 @@ CAF_TEST(serialization) { ...@@ -53,5 +54,5 @@ CAF_TEST(serialization) {
CAF_TEST(to_string) { CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4}; basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(to_string(x), "basp::header(handshake, 42, 4)"); CAF_CHECK_EQUAL(deep_to_string(x), "basp::header(handshake, 42, 4)");
} }
...@@ -122,8 +122,8 @@ template <class... Ts> ...@@ -122,8 +122,8 @@ template <class... Ts>
byte_buffer to_buf(const Ts&... xs) { byte_buffer to_buf(const Ts&... xs) {
byte_buffer buf; byte_buffer buf;
binary_serializer sink{nullptr, buf}; binary_serializer sink{nullptr, buf};
if (auto err = sink(xs...)) if (!sink.apply_objects(xs...))
CAF_FAIL(err); CAF_FAIL("to_buf failed: " << sink.get_error());
return buf; return buf;
} }
......
...@@ -65,9 +65,8 @@ struct string_application_header { ...@@ -65,9 +65,8 @@ struct string_application_header {
/// @relates header /// @relates header
template <class Inspector> template <class Inspector>
typename Inspector::result_type bool inspect(Inspector& f, string_application_header& x) {
inspect(Inspector& f, string_application_header& hdr) { return f.fields(f.field("payload", x.payload));
return f(meta::type_name("sa_header"), hdr.payload);
} }
class string_application { class string_application {
......
...@@ -75,8 +75,8 @@ public: ...@@ -75,8 +75,8 @@ public:
std::unique_ptr<endpoint_manager_queue::message> ptr) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto payload_buf = parent.next_payload_buffer(); auto payload_buf = parent.next_payload_buffer();
binary_serializer sink(parent.system(), payload_buf); binary_serializer sink(parent.system(), payload_buf);
if (auto err = sink(ptr->msg->content())) if (!sink.apply_objects(ptr->msg->content()))
CAF_FAIL("serializing failed: " << err); CAF_FAIL("failed to serialize content: " << sink.get_error());
CAF_MESSAGE("before sending: " << CAF_ARG(ptr->msg->content())); CAF_MESSAGE("before sending: " << CAF_ARG(ptr->msg->content()));
parent.write_packet(payload_buf); parent.write_packet(payload_buf);
return none; return none;
...@@ -209,7 +209,7 @@ CAF_TEST(write_message) { ...@@ -209,7 +209,7 @@ CAF_TEST(write_message) {
auto& buf = transport_results->packet_buffer; auto& buf = transport_results->packet_buffer;
binary_deserializer source{sys, buf}; binary_deserializer source{sys, buf};
caf::message received_msg; caf::message received_msg;
CAF_CHECK_EQUAL(source(received_msg), none); CAF_CHECK(source.apply_objects(received_msg));
CAF_MESSAGE(CAF_ARG(received_msg)); CAF_MESSAGE(CAF_ARG(received_msg));
auto received_str = received_msg.get_as<std::string>(0); auto received_str = received_msg.get_as<std::string>(0);
string_view result{received_str}; string_view result{received_str};
......
...@@ -93,8 +93,8 @@ struct header { ...@@ -93,8 +93,8 @@ struct header {
} }
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, header& x) { friend bool inspect(Inspector& f, header& x) {
return f(meta::type_name("header"), x.payload_size); return f.object(x).fields(f.field("payload_size", x.payload_size));
} }
size_t payload_size; size_t payload_size;
...@@ -116,10 +116,13 @@ CAF_TEST(socket creation) { ...@@ -116,10 +116,13 @@ CAF_TEST(socket creation) {
CAF_TEST(read / write using span<byte>) { CAF_TEST(read / write using span<byte>) {
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true)) if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err); CAF_FAIL("setting socket to nonblocking failed: " << err);
CAF_CHECK_EQUAL(read(receive_socket, buf), sec::unavailable_or_would_block); auto read_res = read(receive_socket, buf);
if (CAF_CHECK(holds_alternative<sec>(read_res)))
CAF_CHECK(get<sec>(read_res) == sec::unavailable_or_would_block);
CAF_MESSAGE("sending data to " << to_string(ep)); CAF_MESSAGE("sending data to " << to_string(ep));
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep), auto write_res=write(send_socket, as_bytes(make_span(hello_test)), ep);
hello_test.size()); if (CAF_CHECK(holds_alternative<size_t>(write_res)))
CAF_CHECK_EQUAL(get<size_t>(write_res), hello_test.size());
CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none); CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()}; string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(received, hello_test); CAF_CHECK_EQUAL(received, hello_test);
...@@ -130,21 +133,23 @@ CAF_TEST(read / write using span<byte_buffer*>) { ...@@ -130,21 +133,23 @@ CAF_TEST(read / write using span<byte_buffer*>) {
header hdr{hello_test.size()}; header hdr{hello_test.size()};
byte_buffer hdr_buf; byte_buffer hdr_buf;
binary_serializer sink(sys, hdr_buf); binary_serializer sink(sys, hdr_buf);
if (auto err = sink(hdr)) if (!sink.apply_object(hdr))
CAF_FAIL("serializing payload failed" << err); CAF_FAIL("failed to serialize payload: " << sink.get_error());
auto bytes = as_bytes(make_span(hello_test)); auto bytes = as_bytes(make_span(hello_test));
byte_buffer payload_buf(bytes.begin(), bytes.end()); byte_buffer payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size(); auto packet_size = hdr_buf.size() + payload_buf.size();
std::vector<byte_buffer*> bufs{&hdr_buf, &payload_buf}; std::vector<byte_buffer*> bufs{&hdr_buf, &payload_buf};
CAF_CHECK_EQUAL(write(send_socket, bufs, ep), packet_size); auto write_res = write(send_socket, bufs, ep);
if (CAF_CHECK(holds_alternative<size_t>(write_res)))
CAF_CHECK_EQUAL(get<size_t>(write_res), packet_size);
// receive both as one single packet. // receive both as one single packet.
buf.resize(packet_size); buf.resize(packet_size);
CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none); CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
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;
if (auto err = source(recv_hdr)) if (!source.apply_objects(recv_hdr))
CAF_FAIL("serializing failed: " << err); CAF_FAIL("failed to deserialize header: " << source.get_error());
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