Commit bd9f84ce authored by Joseph Noir's avatar Joseph Noir

Unify message handling for UDP and TCP

parent 66e1ca2b
...@@ -102,9 +102,7 @@ inline bool operator!=(const header& lhs, const header& rhs) { ...@@ -102,9 +102,7 @@ inline bool operator!=(const header& lhs, const header& rhs) {
/// Checks whether given header contains a handshake. /// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) { inline bool is_handshake(const header& hdr) {
return hdr.operation == message_type::server_handshake return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake || hdr.operation == message_type::client_handshake;
|| hdr.operation == message_type::udp_client_handshake
|| hdr.operation == message_type::udp_server_handshake;
} }
/// Checks wheter given header contains a heartbeat. /// Checks wheter given header contains a heartbeat.
......
...@@ -21,6 +21,8 @@ ...@@ -21,6 +21,8 @@
#define CAF_IO_BASP_INSTANCE_HPP #define CAF_IO_BASP_INSTANCE_HPP
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp" // ---
#include "caf/io/hook.hpp" #include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
...@@ -199,12 +201,12 @@ public: ...@@ -199,12 +201,12 @@ public:
buffer_type& buf, const node_id& remote_side); buffer_type& buf, const node_id& remote_side);
/// Start handshake ... TODO /// Start handshake ... TODO
void write_udp_client_handshake(execution_unit* ctx, buffer_type& buf); //void write_udp_client_handshake(execution_unit* ctx, buffer_type& buf);
/// Answer client handshake ... TODO /// Answer client handshake ... TODO
void write_udp_server_handshake(execution_unit* ctx, buffer_type& buf, //void write_udp_server_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& remote_side, // const node_id& remote_side,
optional<uint16_t> port); // optional<uint16_t> port);
/// Writes an `announce_proxy` to `buf`. /// Writes an `announce_proxy` to `buf`.
void write_announce_proxy(execution_unit* ctx, buffer_type& buf, void write_announce_proxy(execution_unit* ctx, buffer_type& buf,
...@@ -233,6 +235,167 @@ public: ...@@ -233,6 +235,167 @@ public:
return callee_.system(); return callee_.system();
} }
template <class Handle>
bool handle_msg(execution_unit* ctx, const Handle& hdl, header& hdr,
std::vector<char>* payload, bool tcp_based,
optional<uint16_t> port) {
std::cerr << "Handling " << to_string(hdr.operation) << " msg" << std::endl;
auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len;
};
// handle message to ourselves
switch (hdr.operation) {
case message_type::server_handshake: {
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return false;
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
e = bd(aid, sigs);
if (e)
return false;
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false;
}
// close this connection if we already have a direct connection
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false;
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
//auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
}
if (tcp_based)
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
break;
}
case message_type::client_handshake: {
auto is_known_node = tbl_.lookup_hdl(hdr.source_node);
if (is_known_node && tcp_based) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
break;
}
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return false;
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
}
if (!is_known_node) {
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
}
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
}
if (!tcp_based)
write_server_handshake(ctx, path->wr_buf, port);
if (!is_known_node) {
//auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
}
flush(*path);
break;
}
case message_type::dispatch_message: {
if (!payload_valid())
return false;
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
//auto last_hop = tbl_.lookup_node(dm.handle);
//if (hdr.source_node != none
// && hdr.source_node != this_node_
// && last_hop != hdr.source_node
// && tbl_.lookup_direct(hdr.source_node) == invalid_connection_handle
// && tbl_.add_indirect(last_hop, hdr.source_node))
// callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag)) {
auto e = bd(receiver_name);
if (e)
return false;
}
auto e = bd(forwarding_stack, msg);
if (e)
return false;
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
break;
}
case message_type::announce_proxy:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
break;
case message_type::kill_proxy: {
if (!payload_valid())
return false;
binary_deserializer bd{ctx, *payload};
error fail_state;
auto e = bd(fail_state);
if (e)
return false;
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
break;
}
case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
callee_.handle_heartbeat(hdr.source_node);
break;
}
default:
CAF_LOG_ERROR("invalid operation");
return false;
}
return true;
}
private: private:
routing_table tbl_; routing_table tbl_;
published_actor_map published_actors_; published_actor_map published_actors_;
......
...@@ -68,11 +68,7 @@ enum class message_type : uint8_t { ...@@ -68,11 +68,7 @@ enum class message_type : uint8_t {
/// in order to detect disconnects. /// in order to detect disconnects.
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x05, heartbeat = 0x05
// TODO: udp handshakes
udp_client_handshake = 0x06,
udp_server_handshake = 0x07
}; };
/// @relates message_type /// @relates message_type
......
...@@ -132,7 +132,7 @@ public: ...@@ -132,7 +132,7 @@ public:
/// including the node that is assigned as direct path for `hdl`. /// including the node that is assigned as direct path for `hdl`.
// void erase_direct(const endpoint_handle& hdl, erase_callback& cb); // void erase_direct(const endpoint_handle& hdl, erase_callback& cb);
void erase(const endpoint_handle& hdl, erase_callback& cb); void erase(const endpoint_handle& hdl, erase_callback& cb);
/* /*
/// Removes any entry for indirect connection to `dest` and returns /// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`. /// `true` if `dest` had an indirect route, otherwise `false`.
......
...@@ -596,12 +596,8 @@ behavior basp_broker::make_behavior() { ...@@ -596,12 +596,8 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(msg.handle)); << CAF_ARG(msg.handle));
ctx.callback->deliver(make_error(sec::disconnect_during_handshake)); ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
} }
// TODO: some error?
} else { } else {
if (ctx.callback) { // TODO: Do we need to handle available callbacks here?
// TODO: if callback available, handle that
}
// TODO: Is there configuration necessary?
configure_datagram_size(msg.handle, 1500); configure_datagram_size(msg.handle, 1500);
} }
}, },
...@@ -778,7 +774,7 @@ behavior basp_broker::make_behavior() { ...@@ -778,7 +774,7 @@ behavior basp_broker::make_behavior() {
ctx.remote_port = port; ctx.remote_port = port;
ctx.callback = rp; ctx.callback = rp;
auto& bi = state.instance; auto& bi = state.instance;
bi.write_udp_client_handshake(context(), wr_buf(hdl)); bi.write_client_handshake(context(), wr_buf(hdl), none);
flush(hdl); flush(hdl);
configure_datagram_size(hdl, 1500); configure_datagram_size(hdl, 1500);
} else { } else {
......
...@@ -2089,7 +2089,7 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host, ...@@ -2089,7 +2089,7 @@ expected<native_socket> new_dgram_scribe_impl(const std::string& host,
: new_ip_acceptor_impl<AF_INET6>(fd, 0, nullptr); : new_ip_acceptor_impl<AF_INET6>(fd, 0, nullptr);
if (!p) if (!p)
return std::move(p.error()); return std::move(p.error());
std::cerr << "[NDSI] Bound to " << *p << std::endl; //std::cerr << "[NDSI] Bound to " << *p << std::endl;
return sguard.release(); return sguard.release();
} }
...@@ -2126,7 +2126,7 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) { ...@@ -2126,7 +2126,7 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
return std::move(p.error()); return std::move(p.error());
// ok, no errors so far // ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p)); CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
std::cerr << "[NDDI] Bound to " << *p << std::endl; //std::cerr << "[NDDI] Bound to " << *p << std::endl;
return std::make_pair(sguard.release(), *p); return std::make_pair(sguard.release(), *p);
} }
......
...@@ -64,9 +64,6 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx, ...@@ -64,9 +64,6 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx,
auto& buf = rd_buf(); auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes); CAF_ASSERT(buf.size() >= num_bytes);
buf.resize(num_bytes); buf.resize(num_bytes);
std::cerr << "Delegating message, doorman on " << parent()->local_port(hdl())
<< " and scribe on " << parent()->local_port(endpoint)
<< std::endl;
tmp_t tmp{strong_actor_ptr{}, message_id::make(), tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, mailbox_element::forwarding_stack{},
delegate_t{endpoint, std::move(buf), delegate_t{endpoint, std::move(buf),
......
...@@ -71,37 +71,22 @@ bool zero(T val) { ...@@ -71,37 +71,22 @@ bool zero(T val) {
return val == 0; return val == 0;
} }
// TODO: Remove comments
bool server_handshake_valid(const header& hdr) { bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& !valid(hdr.dest_node) //&& !valid(hdr.dest_node)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& !zero(hdr.operation_data); && !zero(hdr.operation_data);
} }
// TODO: Remove comments
bool client_handshake_valid(const header& hdr) { bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& valid(hdr.dest_node) //&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node && hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor) && zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor);
&& zero(hdr.operation_data); //&& zero(hdr.operation_data);
}
bool udp_server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
//&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& !zero(hdr.operation_data);
}
bool udp_client_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& !valid(hdr.dest_node)
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& !zero(hdr.operation_data);
} }
bool dispatch_message_valid(const header& hdr) { bool dispatch_message_valid(const header& hdr) {
...@@ -158,10 +143,6 @@ bool valid(const header& hdr) { ...@@ -158,10 +143,6 @@ bool valid(const header& hdr) {
return kill_proxy_instance_valid(hdr); return kill_proxy_instance_valid(hdr);
case message_type::heartbeat: case message_type::heartbeat:
return heartbeat_valid(hdr); return heartbeat_valid(hdr);
case message_type::udp_server_handshake:
return udp_server_handshake_valid(hdr);
case message_type::udp_client_handshake:
return udp_client_handshake_valid(hdr);
} }
} }
......
...@@ -113,6 +113,9 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -113,6 +113,9 @@ connection_state instance::handle(execution_unit* ctx,
} }
return await_header; return await_header;
} }
if (!handle_msg(ctx, dm.handle, hdr, payload, true, none))
return err();
/*
// function object for checking payload validity // function object for checking payload validity
auto payload_valid = [&]() -> bool { auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len; return payload != nullptr && payload->size() == hdr.payload_len;
...@@ -200,16 +203,14 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -200,16 +203,14 @@ connection_state instance::handle(execution_unit* ctx,
return err(); return err();
// in case the sender of this message was received via a third node, // in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source // we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_node(dm.handle); //auto last_hop = tbl_.lookup_node(dm.handle);
/* // TODO: if we don't need this anymore, what does the rest do here?
* TODO: if we don't need this anymore, what does the rest do here? //if (hdr.source_node != none
if (hdr.source_node != none // && hdr.source_node != this_node_
&& hdr.source_node != this_node_ // && last_hop != hdr.source_node
&& last_hop != hdr.source_node // && tbl_.lookup_hdl(hdr.source_node))
&& tbl_.lookup_hdl(hdr.source_node)) // && tbl_.add_indirect(last_hop, hdr.source_node))
&& tbl_.add_indirect(last_hop, hdr.source_node)) // callee_.learned_new_node_indirectly(hdr.source_node);
callee_.learned_new_node_indirectly(hdr.source_node);
*/
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0); auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack; std::vector<strong_actor_ptr> forwarding_stack;
...@@ -254,8 +255,10 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -254,8 +255,10 @@ connection_state instance::handle(execution_unit* ctx,
} }
default: default:
CAF_LOG_ERROR("invalid operation"); CAF_LOG_ERROR("invalid operation");
std::cerr << "[I] Invalid message type" << std::endl;
return err(); return err();
} }
*/
return await_header; return await_header;
} }
...@@ -271,6 +274,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -271,6 +274,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
return false; return false;
}; };
// Split message into hdr and payload // Split message into hdr and payload
// Loop until all messages in the datagram have been processed.
auto dgram_itr = dm.buf.begin(); auto dgram_itr = dm.buf.begin();
do { do {
// extract header // extract header
...@@ -283,7 +287,8 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -283,7 +287,8 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
auto e = bd(hdr); auto e = bd(hdr);
if (e || !valid(hdr)) { if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr)); CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
std::cerr << "Received invalid header!" << std::endl; std::cerr << "[I] Received invalid header of type: "
<< to_string(hdr.operation) << std::endl;
return err(); return err();
} }
CAF_LOG_DEBUG(CAF_ARG(hdr)); CAF_LOG_DEBUG(CAF_ARG(hdr));
...@@ -292,17 +297,9 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -292,17 +297,9 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
std::move_iterator<itr_t>(dgram_itr + std::move_iterator<itr_t>(dgram_itr +
hdr.payload_len)}; hdr.payload_len)};
dgram_itr += hdr.payload_len; dgram_itr += hdr.payload_len;
// deserialize handshake
std::vector<char>* payload = nullptr; std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) { if (hdr.payload_len > 0) {
payload = &pl_buf; payload = &pl_buf;
//if (payload->size() != hdr.payload_len) {
// CAF_LOG_WARNING("received invalid payload");
// // TODO: This kind of should not happen ...
// std::cerr << "Received invalid payload with size " << payload->size()
// << " although header says " << hdr.payload_len << std::endl;
// return err();
//}
} }
// needs forwarding? // needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) { if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
...@@ -310,14 +307,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -310,14 +307,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
// TODO: anything to do here? (i.e., do we still need forwarding?) // TODO: anything to do here? (i.e., do we still need forwarding?)
return err(); return err();
} }
if (!handle_msg(ctx, dm.handle, hdr, payload, false, dm.port))
return err();
/*
// function object for checking payload validity // function object for checking payload validity
auto payload_valid = [&]() -> bool { auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len; return payload != nullptr && payload->size() == hdr.payload_len;
}; };
// handle message ourselves // handle message ourselves
switch (hdr.operation) { switch (hdr.operation) {
case message_type::udp_server_handshake: { case message_type::server_handshake: {
std::cerr << "[I] Received UDP server handshake" << std::endl; std::cerr << "[I] Received UDP server handshake" << std::endl;
// TODO: test if this is the second time the handshake takes place ...
//if (tbl_.lookup(hdr.source_node)) {
// std::cerr << "[I] Second handshake received from node" << std::endl;
// break;
//}
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
if (payload_valid()) { if (payload_valid()) {
...@@ -361,70 +367,57 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -361,70 +367,57 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
CAF_LOG_ERROR("no route to host after server handshake"); CAF_LOG_ERROR("no route to host after server handshake");
return err(); return err();
} }
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node); callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path); flush(*path);
break; break;
} }
case message_type::udp_client_handshake: { case message_type::client_handshake: {
std::cerr << "[I] Received UDP client handshake" << std::endl; std::cerr << "[I] Received UDP client handshake" << std::endl;
if (tbl_.lookup_hdl(hdr.source_node)) { if (!tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("received second client handshake:" std::cerr << "[I] Node not found, is a new client" << std::endl;
<< CAF_ARG(hdr.source_node)); if (payload_valid()) {
break; std::cerr << "[I] Has valid payload of " << hdr.payload_len
} << " bytes" << std::endl;
if (payload_valid()) { binary_deserializer bd{ctx, *payload};
binary_deserializer bd{ctx, *payload}; std::string remote_appid;
std::string remote_appid; auto e = bd(remote_appid);
auto e = bd(remote_appid); if (e)
if (e) return err();
return err(); if (remote_appid !=
if (remote_appid != callee_.system().config().middleman_app_identifier) {
callee_.system().config().middleman_app_identifier) { CAF_LOG_ERROR("app identifier mismatch");
CAF_LOG_ERROR("app identifier mismatch"); std::cerr << "[I] App identifier mismatch" << std::endl;
return err();
}
} else {
CAF_LOG_ERROR("failed to receive the app identifier");
std::cerr << "[I] Failed to receive the app identifier" << std::endl;
return err(); return err();
} }
} else { // add direct route to this node and remove any indirect entry
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
return err(); tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
} }
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(dm.handle, hdr.source_node);
// write handshake as server in response // write handshake as server in response
auto path = tbl_.lookup(hdr.source_node); auto path = tbl_.lookup(hdr.source_node);
if (!path) { if (!path) {
CAF_LOG_ERROR("no route to host after server handshake"); CAF_LOG_ERROR("no route to host after server handshake");
return err(); return err();
} }
if (dm.port) { write_server_handshake(ctx, path->wr_buf, dm.port);
write_udp_server_handshake(ctx, path->wr_buf, hdr.source_node,
*dm.port);
} else {
write_udp_server_handshake(ctx, path->wr_buf, hdr.source_node,
none);
}
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
break;
}
case message_type::server_handshake: {
std::cerr << "[I] Received TCP server_handshake (ignored)" << std::endl;
break;
}
case message_type::client_handshake: {
std::cerr << "[I] Received TCP client handshake (ignored)" << std::endl;
break; break;
} }
case message_type::dispatch_message: { case message_type::dispatch_message: {
std::cerr << "[I] Received dispatch message" << std::endl; std::cerr << "[I] Received dispatch message" << std::endl;
if (!payload_valid()) if (!payload_valid())
return err(); return err();
// in case the sender of this message was received via a third node, // in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source // we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_node(dm.handle); //auto last_hop = tbl_.lookup_node(dm.handle);
// TODO: if we don't need this anymore, what does the rest do here? // TODO: if we don't need this anymore, what does the rest do here?
//if (hdr.source_node != none //if (hdr.source_node != none
// && hdr.source_node != this_node_ // && hdr.source_node != this_node_
...@@ -446,15 +439,15 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -446,15 +439,15 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
if (e) if (e)
return err(); return err();
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg)); CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
if (hdr.has(header::named_receiver_flag)) if (hdr.has(header::named_receiver_flag)) {
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name, callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
message_id::from_integer_value(hdr.operation_data), message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg); forwarding_stack, msg);
else } else {
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor, callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data), message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg); forwarding_stack, msg);
//return err(); }
break; break;
} }
case message_type::announce_proxy: case message_type::announce_proxy:
...@@ -479,7 +472,6 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -479,7 +472,6 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
std::cerr << "[I] Received hearbeat: " << to_string(hdr.source_node) std::cerr << "[I] Received hearbeat: " << to_string(hdr.source_node)
<< std::endl; << std::endl;
callee_.handle_heartbeat(hdr.source_node); callee_.handle_heartbeat(hdr.source_node);
//return true;
break; break;
} }
default: default:
...@@ -487,8 +479,8 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -487,8 +479,8 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
std::cerr << "[I] Invalid operation" << std::endl; std::cerr << "[I] Invalid operation" << std::endl;
return err(); return err();
} }
*/
} while (dgram_itr != dm.buf.end()); } while (dgram_itr != dm.buf.end());
// TODO: Is this reachable?
return true; return true;
}; };
...@@ -553,8 +545,7 @@ size_t instance::remove_published_actor(uint16_t port, ...@@ -553,8 +545,7 @@ size_t instance::remove_published_actor(uint16_t port,
return 1; return 1;
} }
size_t instance::remove_published_actor(const actor_addr& whom, size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port,
uint16_t port,
removed_published_actor* cb) { removed_published_actor* cb) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
size_t result = 0; size_t result = 0;
...@@ -636,8 +627,7 @@ void instance::write(execution_unit* ctx, buffer_type& buf, ...@@ -636,8 +627,7 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
} }
} }
void instance::write_server_handshake(execution_unit* ctx, void instance::write_server_handshake(execution_unit* ctx, buffer_type& buf,
buffer_type& out_buf,
optional<uint16_t> port) { optional<uint16_t> port) {
CAF_LOG_TRACE(CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(port));
using namespace detail; using namespace detail;
...@@ -666,11 +656,10 @@ void instance::write_server_handshake(execution_unit* ctx, ...@@ -666,11 +656,10 @@ void instance::write_server_handshake(execution_unit* ctx,
this_node_, none, this_node_, none,
pa && pa->first ? pa->first->id() : invalid_actor_id, pa && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id}; invalid_actor_id};
write(ctx, out_buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
void instance::write_client_handshake(execution_unit* ctx, void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf,
buffer_type& buf,
const node_id& remote_side) { const node_id& remote_side) {
CAF_LOG_TRACE(CAF_ARG(remote_side)); CAF_LOG_TRACE(CAF_ARG(remote_side));
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
...@@ -682,6 +671,7 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -682,6 +671,7 @@ void instance::write_client_handshake(execution_unit* ctx,
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
/*
void instance::write_udp_client_handshake(execution_unit* ctx, void instance::write_udp_client_handshake(execution_unit* ctx,
buffer_type& buf) { buffer_type& buf) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -702,17 +692,11 @@ void instance::write_udp_server_handshake(execution_unit* ctx, ...@@ -702,17 +692,11 @@ void instance::write_udp_server_handshake(execution_unit* ctx,
using namespace detail; using namespace detail;
published_actor* pa = nullptr; published_actor* pa = nullptr;
if (port) { if (port) {
std::cerr << "LOOKING FOR ACTOR ON PORT " << *port << std::endl;
auto i = published_actors_.find(*port); auto i = published_actors_.find(*port);
if (i != published_actors_.end()) if (i != published_actors_.end())
pa = &i->second; pa = &i->second;
} }
CAF_LOG_DEBUG_IF(!pa && port, "no actor published"); CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
if (!pa && port) {
std::cerr << "No actor published." << std::endl;
} else {
std::cerr << "Found locally published actor." << std::endl;
}
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto& ref = callee_.system().config().middleman_app_identifier; auto& ref = callee_.system().config().middleman_app_identifier;
auto e = sink(const_cast<std::string&>(ref)); auto e = sink(const_cast<std::string&>(ref));
...@@ -733,6 +717,7 @@ void instance::write_udp_server_handshake(execution_unit* ctx, ...@@ -733,6 +717,7 @@ void instance::write_udp_server_handshake(execution_unit* ctx,
invalid_actor_id}; invalid_actor_id};
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
*/
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf, void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid) { const node_id& dest_node, actor_id aid) {
......
...@@ -33,9 +33,7 @@ const char* message_type_strings[] = { ...@@ -33,9 +33,7 @@ const char* message_type_strings[] = {
"dispatch_message", "dispatch_message",
"announce_proxy_instance", "announce_proxy_instance",
"kill_proxy_instance", "kill_proxy_instance",
"heartbeat", "heartbeat"
"udp_client_handshake",
"udp_server_handshake"
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -122,12 +122,12 @@ CAF_TEST(test_datagram_remote_actor) { ...@@ -122,12 +122,12 @@ CAF_TEST(test_datagram_remote_actor) {
auto res3 = io::uri::make(string(uri_no_port) + ":" + to_string(*res2)); auto res3 = io::uri::make(string(uri_no_port) + ":" + to_string(*res2));
CAF_CHECK(res3); CAF_CHECK(res3);
CAF_MESSAGE("Published pong on: " + to_string(*res3) + "."); CAF_MESSAGE("Published pong on: " + to_string(*res3) + ".");
/*
CAF_SET_LOGGER_SYS(&server_side); CAF_SET_LOGGER_SYS(&server_side);
CAF_MESSAGE("Local call to remote actor should acquire the actor."); CAF_MESSAGE("Local call to remote actor should acquire the actor.");
auto res4 = server_side_mm.remote_actor(*res3); auto res4 = server_side_mm.remote_actor(*res3);
CAF_CHECK(res4); CAF_CHECK(res4);
*/
CAF_MESSAGE("Checking from different actor system next."); CAF_MESSAGE("Checking from different actor system next.");
auto res5 = client_side_mm.remote_actor(*res3); auto res5 = client_side_mm.remote_actor(*res3);
CAF_CHECK(res5); CAF_CHECK(res5);
......
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