Commit af262298 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #479 from ufownl/topic/app-identifier

Add app identifier config parameter
parents 0f810ffb 24571158
......@@ -39,6 +39,8 @@ relaxed-sleep-duration=10000
enable-automatic-connections=false
; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend='default'
; application identifier of this node
app-identifier=''
; maximum number of consecutive I/O reads per broker
max-consecutive-reads=50
; heartbeat message interval in ms (0 disables heartbeating)
......
......@@ -242,6 +242,7 @@ public:
// Config parameters of middleman.
atom_value middleman_network_backend;
std::string middleman_app_identifier;
bool middleman_enable_automatic_connections;
size_t middleman_max_consecutive_reads;
size_t middleman_heartbeat_interval;
......
......@@ -131,6 +131,8 @@ actor_system_config::actor_system_config()
opt_group{options_, "middleman"}
.add(middleman_network_backend, "network-backend",
"sets the network backend to either 'default' or 'asio' (if available)")
.add(middleman_app_identifier, "app-identifier",
"sets the application identifier of this node")
.add(middleman_enable_automatic_connections, "enable-automatic-connections",
"enables or disables automatic connection management (off per default)")
.add(middleman_max_consecutive_reads, "max-consecutive-reads",
......
......@@ -84,7 +84,6 @@ bool client_handshake_valid(const header& hdr) {
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
......
......@@ -22,6 +22,7 @@
#include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/io/basp/version.hpp"
......@@ -116,7 +117,16 @@ connection_state instance::handle(execution_unit* ctx,
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
bd >> remote_appid;
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
bd >> aid >> sigs;
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err();
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
......@@ -153,6 +163,18 @@ connection_state instance::handle(execution_unit* ctx,
<< CAF_ARG(hdr.source_node));
break;
}
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
bd >> remote_appid;
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err();
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(dm.handle, hdr.source_node);
......@@ -366,9 +388,12 @@ void instance::write_server_handshake(execution_unit* ctx,
}
CAF_LOG_DEBUG_IF(! pa && port, "no actor published");
auto writer = make_callback([&](serializer& sink) {
sink << callee_.system().config().middleman_app_identifier;
if (pa) {
auto i = pa->first ? pa->first->id() : invalid_actor_id;
sink << i << pa->second;
} else {
sink << invalid_actor_id << std::set<std::string>{};
}
});
header hdr{message_type::server_handshake, 0, 0, version,
......@@ -382,9 +407,12 @@ void instance::write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
auto writer = make_callback([&](serializer& sink) {
sink << callee_.system().config().middleman_app_identifier;
});
header hdr{message_type::client_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr);
write(ctx, buf, hdr, &writer);
}
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
......
......@@ -131,9 +131,10 @@ public:
auto i = pending_.find(key);
if (i == pending_.end())
return;
monitor(addr);
if (nid && addr)
if (nid && addr) {
monitor(addr);
cached_.emplace(key, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(ok_atom::value, std::move(nid),
std::move(addr), std::move(sigs));
for (auto& promise : i->second)
......
......@@ -287,33 +287,27 @@ public:
mpx_->accept_connection(src);
// technically, the server handshake arrives
// before we send the client handshake
auto m = mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0,
remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id});
if (published_actor_id != invalid_actor_id)
m.expect(hdl,
basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{invalid_node_id},
published_actor_id, invalid_actor_id,
published_actor_id,
published_actor_ifs);
else
m.expect(hdl,
basp::message_type::server_handshake, no_flags, any_vals,
basp::version, this_node(), node_id{invalid_node_id},
invalid_actor_id, invalid_actor_id);
mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0,
remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id}, std::string{})
.expect(hdl,
basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{invalid_node_id},
published_actor_id, invalid_actor_id, std::string{},
published_actor_id,
published_actor_ifs)
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
m.expect(hdl,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data,
this_node(), remote_node(i),
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_addr>{},
make_message(sys_atom::value, get_atom::value, "info"));
.expect(hdl,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
no_operation_data,
this_node(), remote_node(i),
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_addr>{},
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i));
......@@ -489,7 +483,7 @@ CAF_TEST(non_empty_server_handshake) {
basp::header expected{basp::message_type::server_handshake, 0, 0,
basp::version, this_node(), invalid_node_id,
self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr,
to_buf(expected_buf, expected, nullptr, std::string{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
}
......@@ -595,12 +589,13 @@ CAF_TEST(remote_actor_and_send) {
{basp::message_type::server_handshake, 0, 0, basp::version,
remote_node(0), invalid_node_id,
pseudo_remote(0)->id(), invalid_actor_id},
std::string{},
pseudo_remote(0)->id(),
uint32_t{0})
.expect(remote_hdl(0),
basp::message_type::client_handshake, no_flags, no_payload,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id)
invalid_actor_id, invalid_actor_id, std::string{})
.expect(remote_hdl(0),
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
......@@ -823,12 +818,13 @@ CAF_TEST(automatic_connection) {
{basp::message_type::server_handshake, 0, 0, basp::version,
remote_node(0), invalid_node_id,
pseudo_remote(0)->id(), invalid_actor_id},
std::string{},
pseudo_remote(0)->id(),
uint32_t{0})
.expect(remote_hdl(0),
basp::message_type::client_handshake, no_flags, no_payload,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id);
invalid_actor_id, invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(0)), invalid_node_id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(1)), invalid_node_id);
CAF_CHECK_EQUAL(tbl().lookup_direct(remote_node(0)).id(), remote_hdl(0).id());
......
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