Commit 8816a06b authored by Dominik Charousset's avatar Dominik Charousset

Handle disconnect during handshake, fix #192

parent 39bb4b35
...@@ -15,6 +15,7 @@ set (LIBCAF_IO_SRCS ...@@ -15,6 +15,7 @@ set (LIBCAF_IO_SRCS
src/default_multiplexer.cpp src/default_multiplexer.cpp
src/publish.cpp src/publish.cpp
src/publish_local_groups.cpp src/publish_local_groups.cpp
src/remote_actor.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/remote_group.cpp src/remote_group.cpp
src/manager.cpp src/manager.cpp
......
...@@ -67,7 +67,6 @@ class basp_broker : public broker, public actor_namespace::backend { ...@@ -67,7 +67,6 @@ class basp_broker : public broker, public actor_namespace::backend {
struct client_handshake_data { struct client_handshake_data {
id_type remote_id; id_type remote_id;
std::promise<abstract_actor_ptr>* result; std::promise<abstract_actor_ptr>* result;
std::string* error_msg;
const std::set<std::string>* expected_ifs; const std::set<std::string>* expected_ifs;
}; };
......
...@@ -28,12 +28,26 @@ ...@@ -28,12 +28,26 @@
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/remote_actor_impl.hpp"
#include "caf/io/typed_remote_actor_helper.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
const std::string& host, uint16_t port);
template <class List>
struct typed_remote_actor_helper;
template <template <class...> class List, class... Ts>
struct typed_remote_actor_helper<List<Ts...>> {
using return_type = typed_actor<Ts...>;
template <class... Vs>
return_type operator()(Vs&&... vs) {
auto iface = return_type::message_types();
auto tmp = remote_actor_impl(std::move(iface), std::forward<Vs>(vs)...);
return actor_cast<return_type>(tmp);
}
};
/** /**
* Establish a new connection to the actor at `host` on given `port`. * Establish a new connection to the actor at `host` on given `port`.
* @param host Valid hostname or IP address. * @param host Valid hostname or IP address.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_TYPED_REMOTE_ACTOR_HELPER_HPP
#define CAF_IO_TYPED_REMOTE_ACTOR_HELPER_HPP
#include "caf/actor_cast.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_actor.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/io/remote_actor_impl.hpp"
namespace caf {
namespace io {
template <class List>
struct typed_remote_actor_helper;
template <template <class...> class List, class... Ts>
struct typed_remote_actor_helper<List<Ts...>> {
using return_type = typed_actor<Ts...>;
template <class... Vs>
return_type operator()(Vs&&... vs) {
auto iface = return_type::message_types();
auto tmp = remote_actor_impl(std::move(iface), std::forward<Vs>(vs)...);
return actor_cast<return_type>(tmp);
}
};
} // namespace io
} // namespace caf
#endif // CAF_IO_TYPED_REMOTE_ACTOR_HELPER_HPP
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/exception.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
...@@ -68,7 +69,15 @@ behavior basp_broker::make_behavior() { ...@@ -68,7 +69,15 @@ behavior basp_broker::make_behavior() {
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
CAF_LOGM_TRACE("make_behavior$connection_closed_msg", CAF_LOGM_TRACE("make_behavior$connection_closed_msg",
CAF_MARG(msg.handle, id)); CAF_MARG(msg.handle, id));
CAF_REQUIRE(m_ctx.count(msg.handle) > 0); auto j = m_ctx.find(msg.handle);
if (j != m_ctx.end()) {
auto hd = j->second.handshake_data;
if (hd) {
network_error err{"disconnect during handshake"};
hd->result->set_exception(std::make_exception_ptr(err));
}
m_ctx.erase(j);
}
// purge handle from all routes // purge handle from all routes
std::vector<id_type> lost_connections; std::vector<id_type> lost_connections;
for (auto& kvp : m_routes) { for (auto& kvp : m_routes) {
...@@ -97,7 +106,6 @@ behavior basp_broker::make_behavior() { ...@@ -97,7 +106,6 @@ behavior basp_broker::make_behavior() {
p->kill_proxy(exit_reason::remote_link_unreachable); p->kill_proxy(exit_reason::remote_link_unreachable);
} }
} }
m_ctx.erase(msg.handle);
}, },
// received from underlying broker implementation // received from underlying broker implementation
[=](const acceptor_closed_msg& msg) { [=](const acceptor_closed_msg& msg) {
...@@ -439,6 +447,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -439,6 +447,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
auto e = what.end(); auto e = what.end();
tmp += *i++; tmp += *i++;
while (i != e) { while (i != e) {
tmp += ",";
tmp += *i++; tmp += *i++;
} }
tmp += ">"; tmp += ">";
...@@ -446,7 +455,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -446,7 +455,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
}; };
auto iface_str = tostr(remote_ifs); auto iface_str = tostr(remote_ifs);
auto expected_str = tostr(ifs); auto expected_str = tostr(ifs);
auto& error_msg = *(ctx.handshake_data->error_msg); std::string error_msg;
if (ifs.empty()) { if (ifs.empty()) {
error_msg = "expected remote actor to be a " error_msg = "expected remote actor to be a "
"dynamically typed actor but found " "dynamically typed actor but found "
...@@ -466,7 +475,8 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -466,7 +475,8 @@ basp_broker::handle_basp_header(connection_context& ctx,
+ iface_str; + iface_str;
} }
// abort with error // abort with error
ctx.handshake_data->result->set_value(nullptr); std::runtime_error err{error_msg};
ctx.handshake_data->result->set_exception(std::make_exception_ptr(err));
return close_connection; return close_connection;
} }
auto nid = ctx.handshake_data->remote_id; auto nid = ctx.handshake_data->remote_id;
......
...@@ -17,8 +17,7 @@ ...@@ -17,8 +17,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_IO_REMOTE_ACTOR_IMPL_HPP #include "caf/io/remote_actor.hpp"
#define CAF_IO_REMOTE_ACTOR_IMPL_HPP
#include <set> #include <set>
#include <ios> #include <ios>
...@@ -40,14 +39,11 @@ ...@@ -40,14 +39,11 @@
namespace caf { namespace caf {
namespace io { namespace io {
template <class... Ts>
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs, abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
auto mm = middleman::instance(); auto mm = middleman::instance();
std::string error_msg; std::promise<abstract_actor_ptr> res;
std::promise<abstract_actor_ptr> result_promise; basp_broker::client_handshake_data hdata{invalid_node_id, &res, &ifs};
basp_broker::client_handshake_data hdata{invalid_node_id, &result_promise,
&error_msg, &ifs};
mm->run_later([&] { mm->run_later([&] {
try { try {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
...@@ -55,17 +51,12 @@ abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs, ...@@ -55,17 +51,12 @@ abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
bro->init_client(hdl, &hdata); bro->init_client(hdl, &hdata);
} }
catch (...) { catch (...) {
result_promise.set_exception(std::current_exception()); res.set_exception(std::current_exception());
} }
}); });
auto result = result_promise.get_future().get(); return res.get_future().get();
if (!result) {
throw std::runtime_error(error_msg);
}
return result;
} }
} // namespace io } // namespace io
} // namespace caf } // namespace caf
#endif // CAF_IO_REMOTE_ACTOR_IMPL_HPP
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