Commit 356af0d2 authored by Jakob Otto's avatar Jakob Otto

Include review feedback

parent e55cc605
......@@ -19,6 +19,7 @@
#pragma once
#include <map>
#include <mutex>
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
......@@ -39,6 +40,10 @@ namespace caf::net::backend {
/// @warning this backend is *not* thread safe.
class CAF_NET_EXPORT tcp : public middleman_backend {
public:
using peer_map = std::map<node_id, endpoint_manager_ptr>;
using emplace_return_type = std::pair<peer_map::iterator, bool>;
// -- constructors, destructors, and assignment operators --------------------
tcp(middleman& mm);
......@@ -51,7 +56,7 @@ public:
void stop() override;
expected<endpoint_manager_ptr> connect(const uri& locator) override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
endpoint_manager_ptr peer(const node_id& id) override;
......@@ -63,19 +68,14 @@ public:
// -- properties -------------------------------------------------------------
tcp_stream_socket socket(const node_id& peer_id) {
return socket_cast<tcp_stream_socket>(peers_[peer_id]->handle());
}
uint16_t port() const noexcept override {
return listening_port_;
}
uint16_t port() const noexcept override;
template <class Handle>
expected<endpoint_manager_ptr>
emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
nonblocking(socket_handle, true);
if (auto err = nonblocking(socket_handle, true))
return err;
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(
......@@ -85,8 +85,15 @@ public:
return err;
}
mpx->register_reading(mgr);
peers_[peer_id] = std::move(mgr);
return peers_[peer_id];
emplace_return_type res;
{
const std::lock_guard<std::mutex> lock(lock_);
res = peers_.emplace(peer_id, mgr);
}
if (res.second)
return std::move(mgr);
else
return make_error(sec::runtime_error, "peer_id already exists");
}
private:
......@@ -94,11 +101,13 @@ private:
middleman& mm_;
std::map<node_id, endpoint_manager_ptr> peers_;
peer_map peers_;
proxy_registry proxies_;
uint16_t listening_port_;
std::mutex lock_;
};
} // namespace caf::net::backend
......@@ -51,7 +51,7 @@ public:
endpoint_manager_ptr peer(const node_id& id) override;
expected<endpoint_manager_ptr> connect(const uri& locator) override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override;
......@@ -65,9 +65,7 @@ public:
return get_peer(peer_id).first;
}
uint16_t port() const noexcept override {
return 0;
}
uint16_t port() const noexcept override;
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
......
......@@ -82,6 +82,8 @@ public:
// Publishes an actor.
template <class Handle>
void publish(Handle whom, const std::string& path) {
// TODO: Currently, we can't get the interface from the registry. Either we
// change that, or we need to associate the handle with the interface.
system().registry().put(path, whom);
}
......@@ -89,18 +91,19 @@ public:
void resolve(const uri& locator, const actor& listener);
template <class Handle = actor, class Duration = std::chrono::seconds>
expected<Handle> remote_actor(const uri& locator, Duration timeout_duration
= std::chrono::seconds(5)) {
expected<Handle>
remote_actor(const uri& locator, Duration timeout = std::chrono::seconds(5)) {
scoped_actor self{sys_};
resolve(locator, self);
Handle handle;
error err;
self->receive(
[&handle](strong_actor_ptr& ptr, const std::set<std::string>&) {
// TODO: This cast is not type-safe.
handle = actor_cast<Handle>(std::move(ptr));
},
[&err](const error& e) { err = e; },
after(timeout_duration) >>
after(timeout) >>
[&err] {
err = make_error(sec::runtime_error,
"manager did not respond with a proxy.");
......
......@@ -47,7 +47,7 @@ public:
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node.
virtual expected<endpoint_manager_ptr> connect(const uri& locator) = 0;
virtual expected<endpoint_manager_ptr> get_or_connect(const uri& locator) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
......
......@@ -18,6 +18,7 @@
#include "caf/net/backend/tcp.hpp"
#include <mutex>
#include <string>
#include "caf/net/actor_proxy_impl.hpp"
......@@ -56,7 +57,8 @@ error tcp::init() {
if (!acceptor)
return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor);
nonblocking(acc_guard.socket(), true);
if (auto err = nonblocking(acc_guard.socket(), true))
return err;
auto port = local_port(*acceptor);
if (!port)
return port.error();
......@@ -82,17 +84,21 @@ void tcp::stop() {
peers_.clear();
}
expected<endpoint_manager_ptr> tcp::connect(const uri& locator) {
auto auth = locator.authority();
auto host = auth.host;
expected<endpoint_manager_ptr> tcp::get_or_connect(const uri& locator) {
if (auto auth = locator.authority_only()) {
auto id = make_node_id(*auth);
if (auto ptr = peer(id))
return ptr;
auto host = locator.authority().host;
if (auto hostname = get_if<std::string>(&host)) {
for (const auto& addr : ip::resolve(*hostname)) {
ip_endpoint ep{addr, auth.port};
ip_endpoint ep{addr, locator.authority().port};
auto sock = make_connected_tcp_stream_socket(ep);
if (!sock)
continue;
else
return emplace(make_node_id(*locator.authority_only()), *sock);
return emplace(id, *sock);
}
}
}
return sec::cannot_connect_to_node;
......@@ -103,12 +109,11 @@ endpoint_manager_ptr tcp::peer(const node_id& id) {
}
void tcp::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id) {
if (auto id = locator.authority_only()) {
auto p = peer(make_node_id(*id));
if (p == nullptr) {
CAF_LOG_INFO("connecting to " << CAF_ARG(locator));
auto res = connect(locator);
auto res = get_or_connect(locator);
if (!res)
anon_send(listener, error(sec::cannot_connect_to_node));
else
......@@ -132,7 +137,12 @@ void tcp::set_last_hop(node_id*) {
// nop
}
uint16_t tcp::port() const noexcept {
return listening_port_;
}
endpoint_manager_ptr tcp::get_peer(const node_id& id) {
const std::lock_guard<std::mutex> lock(lock_);
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
......
......@@ -55,9 +55,11 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
expected<endpoint_manager_ptr> test::connect(const uri&) {
expected<endpoint_manager_ptr> test::get_or_connect(const uri& locator) {
if (auto ptr = peer(make_node_id(*locator.authority_only())))
return ptr;
return make_error(sec::runtime_error,
"function not implemented in test_backend");
"connecting not implemented in test backend");
}
void test::resolve(const uri& locator, const actor& listener) {
......@@ -80,6 +82,10 @@ void test::set_last_hop(node_id*) {
// nop
}
uint16_t test::port() const noexcept {
return 0;
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
......
......@@ -97,9 +97,8 @@ void* middleman::subtype_ptr() {
}
expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
auto ptr = backend(locator.scheme());
if (auto ret = ptr->connect(locator))
return ret;
if (auto ptr = backend(locator.scheme()))
return ptr->get_or_connect(locator);
else
return basp::ec::invalid_scheme;
}
......@@ -123,8 +122,7 @@ middleman_backend* middleman::backend(string_view scheme) const noexcept {
}
expected<uint16_t> middleman::port(string_view scheme) const {
auto ptr = backend(scheme);
if (ptr != nullptr)
if (auto ptr = backend(scheme))
return ptr->port();
else
return basp::ec::invalid_scheme;
......
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