Commit ff2dca97 authored by Jakob Otto's avatar Jakob Otto

Fix merge conflict

parents cc1efee4 37fc6104
......@@ -50,7 +50,7 @@ if(CAF_INC_ENABLE_STANDALONE_BUILD)
FetchContent_Declare(
actor_framework
GIT_REPOSITORY https://github.com/actor-framework/actor-framework.git
GIT_TAG e3f7e64c8
GIT_TAG c5f6c835c
)
FetchContent_Populate(actor_framework)
set(CAF_ENABLE_EXAMPLES OFF CACHE BOOL "" FORCE)
......
......@@ -22,6 +22,7 @@
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
......@@ -85,7 +86,11 @@ public:
auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
this->read_buf_.resize(num_bytes);
this->next_layer_.handle_data(*this, this->read_buf_, std::move(ep));
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_,
std::move(ep))) {
CAF_LOG_ERROR("handle_data failed: " << err);
return false;
}
prepare_next_read();
} else {
auto err = get<sec>(ret);
......
......@@ -65,7 +65,8 @@ public:
max_(1024),
rd_flag_(net::receive_policy_flag::exactly) {
CAF_ASSERT(handle != invalid_socket);
nodelay(handle, true);
if (auto err = nodelay(handle, true))
CAF_LOG_ERROR("nodelay failed: " << err);
}
// -- member functions -------------------------------------------------------
......
......@@ -209,7 +209,7 @@ public:
private:
// -- utility functions ------------------------------------------------------
static byte_buffer 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());
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
......@@ -75,7 +76,8 @@ public:
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.write_message(writer, std::move(msg));
if (auto err = application_.write_message(writer, std::move(msg)))
CAF_LOG_ERROR("write_message failed: " << err);
}
template <class Parent>
......@@ -91,8 +93,8 @@ public:
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) {
void
local_actor_down(Parent& parent, const node_id&, actor_id id, error reason) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, id, std::move(reason));
}
......@@ -113,7 +115,7 @@ private:
};
template <class Application, class IdType = unit_t>
using transport_worker_ptr = std::shared_ptr<
transport_worker<Application, IdType>>;
using transport_worker_ptr
= std::shared_ptr<transport_worker<Application, IdType>>;
} // namespace caf::net
......@@ -89,7 +89,8 @@ uint16_t test::port() const noexcept {
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
nonblocking(second, true);
if (auto err = nonblocking(second, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(mpx, mm_.system(),
......
......@@ -32,7 +32,8 @@ pollset_updater::pollset_updater(pipe_socket read_handle,
const multiplexer_ptr& parent)
: super(read_handle, parent), buf_size_(0) {
mask_ = operation::read;
nonblocking(read_handle, true);
if (auto err = nonblocking(read_handle, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
}
pollset_updater::~pollset_updater() {
......
......@@ -65,7 +65,8 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
tcp_accept_socket sock{fd};
// sguard closes the socket in case of exception
auto sguard = make_socket_guard(tcp_accept_socket{fd});
child_process_inherit(sock, false);
if (auto err = child_process_inherit(sock, false))
return err;
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
......@@ -73,8 +74,9 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
using sockaddr_type = typename std::conditional<
Family == AF_INET, sockaddr_in, sockaddr_in6>::type;
using sockaddr_type =
typename std::conditional<Family == AF_INET, sockaddr_in,
sockaddr_in6>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
detail::family_of(sa) = Family;
......
......@@ -38,8 +38,9 @@ bool ip_connect(stream_socket fd, std::string host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
<< CAF_ARG(fd.id) << CAF_ARG(host) << CAF_ARG(port));
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
using sockaddr_type = typename std::conditional<
Family == AF_INET, sockaddr_in, sockaddr_in6>::type;
using sockaddr_type =
typename std::conditional<Family == AF_INET, sockaddr_in,
sockaddr_in6>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
inet_pton(Family, host.c_str(), &detail::addr_of(sa));
......@@ -60,7 +61,8 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0));
tcp_stream_socket sock{fd};
child_process_inherit(sock, false);
if (auto err = child_process_inherit(sock, false))
return err;
auto sguard = make_socket_guard(sock);
if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(sock, to_string(node.address()), node.port())) {
......
......@@ -115,13 +115,14 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
return none;
}
template <class Parent>
......@@ -221,7 +222,7 @@ CAF_TEST(resolve and proxy communication) {
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr);
auto& transport = mgr_impl->transport();
transport.add_new_worker(make_node_id(uri), ep);
CAF_CHECK_EQUAL(transport.add_new_worker(make_node_id(uri), ep), none);
run();
mgr->resolve(uri, self);
run();
......
......@@ -21,6 +21,7 @@
#include "caf/net/doorman.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/error.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
......@@ -66,13 +67,14 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
return none;
}
template <class Parent>
......
......@@ -163,7 +163,7 @@ CAF_TEST(send and receive) {
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(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
sec::unavailable_or_would_block);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
......@@ -189,7 +189,7 @@ CAF_TEST(resolve and proxy communication) {
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(nonblocking(sockets.second, true), none);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf});
......
......@@ -88,13 +88,14 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto payload_buf = parent.next_payload_buffer();
binary_serializer sink{parent.system(), payload_buf};
if (auto err = sink(msg->msg->payload))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
return none;
}
template <class Parent>
......
......@@ -97,11 +97,11 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
// Ignore proxy announcement messages.
if (ptr->msg == nullptr)
return;
return none;
auto header_buf = parent.next_header_buffer();
auto payload_buf = parent.next_payload_buffer();
binary_serializer payload_sink{parent.system(), payload_buf};
......@@ -112,6 +112,7 @@ public:
= header_sink(header_type{static_cast<uint32_t>(payload_buf.size())}))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(header_buf, payload_buf);
return none;
}
private:
......@@ -206,7 +207,7 @@ CAF_TEST(receive) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<byte_buffer>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
CAF_CHECK_EQUAL(nonblocking(sockets.second, true), none);
CAF_CHECK_EQUAL(read(sockets.second, read_buf),
sec::unavailable_or_would_block);
CAF_MESSAGE("adding both endpoint managers");
......
......@@ -71,14 +71,15 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto payload_buf = parent.next_payload_buffer();
binary_serializer sink(parent.system(), payload_buf);
if (auto err = sink(ptr->msg->content()))
CAF_FAIL("serializing failed: " << err);
CAF_MESSAGE("before sending: " << CAF_ARG(ptr->msg->content()));
parent.write_packet(payload_buf);
return none;
}
template <class Parent>
......@@ -188,7 +189,7 @@ CAF_TEST(construction and initialization) {
CAF_TEST(handle_data) {
auto test_span = as_bytes(make_span(hello_test));
worker.handle_data(transport, test_span);
CAF_CHECK_EQUAL(worker.handle_data(transport, test_span), none);
auto& buf = application_results->data_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(result, hello_test);
......@@ -208,7 +209,7 @@ CAF_TEST(write_message) {
auto& buf = transport_results->packet_buffer;
binary_deserializer source{sys, buf};
caf::message received_msg;
CAF_CHECK(!source(received_msg));
CAF_CHECK_EQUAL(source(received_msg), none);
CAF_MESSAGE(CAF_ARG(received_msg));
auto received_str = received_msg.get_as<std::string>(0);
string_view result{received_str};
......
......@@ -66,11 +66,12 @@ public:
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
error 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<byte_buffer>(0);
parent.write_packet(data);
return none;
}
template <class Parent>
......@@ -249,7 +250,8 @@ struct fixture : host_fixture {
};
#define CHECK_HANDLE_DATA(testcase) \
dispatcher.handle_data(dummy, span<const byte>{}, testcase.ep); \
CAF_CHECK_EQUAL( \
dispatcher.handle_data(dummy, span<const byte>{}, testcase.ep), none); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
......
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