Commit 1f0eeec0 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Integrate review feedback

parent f4de15a9
...@@ -215,9 +215,6 @@ private: ...@@ -215,9 +215,6 @@ private:
/// Re-usable buffer for storing payloads. /// Re-usable buffer for storing payloads.
buffer_type buf_; buffer_type buf_;
/// Stores our own ID.
node_id id_;
/// Stores the ID of our peer. /// Stores the ID of our peer.
node_id peer_id_; node_id peer_id_;
......
...@@ -39,7 +39,7 @@ public: ...@@ -39,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer_ptr parent); pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent);
~pollset_updater() override; ~pollset_updater() override;
...@@ -61,7 +61,6 @@ public: ...@@ -61,7 +61,6 @@ public:
private: private:
msg_buf buf_; msg_buf buf_;
size_t buf_size_; size_t buf_size_;
std::mutex write_lock_;
}; };
} // namespace net } // namespace net
......
...@@ -228,9 +228,9 @@ error application::handle_actor_message(write_packet_callback&, header hdr, ...@@ -228,9 +228,9 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
byte_span payload) { byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload. // Deserialize payload.
actor_id src_id; actor_id src_id = 0;
node_id src_node; node_id src_node;
actor_id dst_id; actor_id dst_id = 0;
std::vector<strong_actor_ptr> fwd_stack; std::vector<strong_actor_ptr> fwd_stack;
message content; message content;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, payload};
......
...@@ -55,7 +55,7 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() { ...@@ -55,7 +55,7 @@ endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
void endpoint_manager::resolve(uri locator, actor listener) { void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result; using intrusive::inbox_result;
using event_type = endpoint_manager_queue::event; using event_type = endpoint_manager_queue::event;
auto ptr = new event_type(std::move(locator), std::move(listener)); auto ptr = new event_type(std::move(locator), listener);
if (!enqueue(ptr)) if (!enqueue(ptr))
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down)); make_error(sec::request_receiver_down));
......
...@@ -33,8 +33,6 @@ namespace caf { ...@@ -33,8 +33,6 @@ namespace caf {
namespace net { namespace net {
namespace backend { namespace backend {
using transport_type = stream_transport<basp::application>;
test::test(middleman& mm) test::test(middleman& mm)
: middleman_backend("test"), mm_(mm), proxies_(mm.system(), *this) { : middleman_backend("test"), mm_(mm), proxies_(mm.system(), *this) {
// nop // nop
...@@ -80,8 +78,10 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first, ...@@ -80,8 +78,10 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
auto mgr = make_endpoint_manager(mpx, mm_.system(), auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second, transport_type{second,
basp::application{proxies_}}); basp::application{proxies_}});
if (auto err = mgr->init()) if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << mm_.system().render(err));
CAF_RAISE_ERROR("mgr->init() failed"); CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr); mpx->register_reading(mgr);
auto& result = peers_[peer_id]; auto& result = peers_[peer_id];
result = std::make_pair(first, std::move(mgr)); result = std::make_pair(first, std::move(mgr));
...@@ -93,8 +93,11 @@ test::peer_entry& test::get_peer(const node_id& id) { ...@@ -93,8 +93,11 @@ test::peer_entry& test::get_peer(const node_id& id) {
if (i != peers_.end()) if (i != peers_.end())
return i->second; return i->second;
auto sockets = make_stream_socket_pair(); auto sockets = make_stream_socket_pair();
if (!sockets) if (!sockets) {
CAF_LOG_ERROR("make_stream_socket_pair failed: "
<< mm_.system().render(sockets.error()));
CAF_RAISE_ERROR("make_stream_socket_pair failed"); CAF_RAISE_ERROR("make_stream_socket_pair failed");
}
return emplace(id, sockets->first, sockets->second); return emplace(id, sockets->first, sockets->second);
} }
......
...@@ -60,8 +60,10 @@ void middleman::stop() { ...@@ -60,8 +60,10 @@ void middleman::stop() {
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) if (auto err = mpx_->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << system().render(err));
CAF_RAISE_ERROR("mpx->init failed"); CAF_RAISE_ERROR("mpx->init failed");
}
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri)); auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node); sys_.node_.swap(this_node);
...@@ -69,8 +71,10 @@ void middleman::init(actor_system_config& cfg) { ...@@ -69,8 +71,10 @@ void middleman::init(actor_system_config& cfg) {
CAF_RAISE_ERROR("no valid entry for middleman.this-node found"); CAF_RAISE_ERROR("no valid entry for middleman.this-node found");
} }
for (auto& backend : backends_) for (auto& backend : backends_)
if (auto err = backend->init()) if (auto err = backend->init()) {
CAF_LOG_ERROR("failed to initialize backend: " << system().render(err));
CAF_RAISE_ERROR("failed to initialize backend"); CAF_RAISE_ERROR("failed to initialize backend");
}
} }
middleman::module::id_t middleman::id() const { middleman::module::id_t middleman::id() const {
......
...@@ -27,8 +27,8 @@ namespace caf { ...@@ -27,8 +27,8 @@ namespace caf {
namespace net { namespace net {
pollset_updater::pollset_updater(pipe_socket read_handle, pollset_updater::pollset_updater(pipe_socket read_handle,
multiplexer_ptr parent) const multiplexer_ptr& parent)
: super(read_handle, std::move(parent)), buf_size_(0) { : super(read_handle, parent), buf_size_(0) {
mask_ = operation::read; mask_ = operation::read;
nonblocking(read_handle, true); nonblocking(read_handle, true);
} }
......
...@@ -26,19 +26,15 @@ ...@@ -26,19 +26,15 @@
#include <vector> #include <vector>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/backend/test.hpp" #include "caf/net/backend/test.hpp"
#include "caf/net/basp/connection_state.hpp" #include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.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/net/stream_transport.hpp" #include "caf/net/stream_transport.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
using namespace caf; using namespace caf;
......
...@@ -85,7 +85,7 @@ public: ...@@ -85,7 +85,7 @@ public:
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, string_view path, actor listener) { void resolve(Parent& parent, string_view path, const actor& listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
......
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