Commit bbb7839a authored by Joseph Noir's avatar Joseph Noir

Handle udp client handshake

parent bf227239
...@@ -198,9 +198,13 @@ public: ...@@ -198,9 +198,13 @@ public:
void write_client_handshake(execution_unit* ctx, void write_client_handshake(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side); buffer_type& buf, const node_id& remote_side);
/// Start handshake ... /// 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
void write_udp_server_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& remote_side);
/// 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,
const node_id& dest_node, actor_id aid); const node_id& dest_node, actor_id aid);
......
...@@ -1254,6 +1254,13 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -1254,6 +1254,13 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len, auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa), no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa),
sa_len); sa_len);
{
std::string host;
uint16_t port;
std::tie(host,port) = sender_from_sockaddr(sa, sa_len);
std::cerr << "[SD] Sent " << sres << " bytes to "
<< host << ":" << port << std::endl;
}
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres)); CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return false; return false;
...@@ -1263,12 +1270,19 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -1263,12 +1270,19 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
} }
bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len, bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len,
sockaddr_storage& sender_addr, socklen_t& sender_len) { sockaddr_storage& sa, socklen_t& sa_len) {
CAF_LOG_TRACE(CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(fd));
sender_len = sizeof(sockaddr); sa_len = sizeof(sockaddr);
auto sres = ::recvfrom(fd, buf, len, 0, auto sres = ::recvfrom(fd, buf, len, 0,
reinterpret_cast<struct sockaddr *>(&sender_addr), reinterpret_cast<struct sockaddr *>(&sa),
&sender_len); &sa_len);
{
std::string host;
uint16_t port;
std::tie(host,port) = sender_from_sockaddr(sa, sa_len);
std::cerr << "[SD] Received " << sres << " bytes from "
<< host << ":" << port << std::endl;
}
// TODO: Check if sres > len and do some error handling ... // TODO: Check if sres > len and do some error handling ...
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres)); CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
......
...@@ -281,13 +281,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -281,13 +281,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
std::cerr << "Received invalid header!" << std::endl; std::cerr << "Received invalid header!" << std::endl;
return err(); return err();
} }
if (hdr.payload_len > 0) {
// TODO: handle payload
}
CAF_LOG_DEBUG(CAF_ARG(hdr)); CAF_LOG_DEBUG(CAF_ARG(hdr));
/* std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) {
payload = &pl_buf;
if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload");
std::cerr << "Received invalid payload" << std::endl;
return err();
}
std::cerr << "Received " << payload->size() << " bytes payload."
<< std::endl;
}
// TODO: handle payload
// 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_) {
std::cerr << "[I] Needs forwarding?" << std::endl;
/*
CAF_LOG_DEBUG("forward message"); CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node); auto path = lookup(hdr.dest_node);
if (path) { if (path) {
...@@ -315,24 +325,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -315,24 +325,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
notify<hook::message_forwarding_failed>(hdr, payload); notify<hook::message_forwarding_failed>(hdr, payload);
} }
return await_header; return await_header;
*/
} }
// 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 to ourselves // handle message to ourselves
switch (hdr.operation) { switch (hdr.operation) {
case message_type::udp_server_handshake: { case message_type::udp_server_handshake: {
std::cerr << "[I] message_type::udp_server_handshake" << std::endl; std::cerr << "[I] Received UDP server handshake" << std::endl;
break; break;
} }
case message_type::udp_client_handshake: { case message_type::udp_client_handshake: {
std::cerr << "[I] message_type::udp_client_handshake" << std::endl; std::cerr << "[I] Received UDP client handshake" << std::endl;
// udp scribes start with a client handshake when sending the first // udp scribes start with a client handshake when sending the first
// message, should be answered by a server handshake! // message, should be answered by a server handshake!
/*
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()) {
...@@ -343,6 +352,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -343,6 +352,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
return err(); return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) { if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch"); CAF_LOG_ERROR("app identifier mismatch");
std::cerr << "app identifier mismatch" << std::endl;
return err(); return err();
} }
e = bd(aid, sigs); e = bd(aid, sigs);
...@@ -350,11 +360,13 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -350,11 +360,13 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
return err(); return err();
} else { } else {
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_ERROR("fail to receive the app identifier");
std::cerr << "fail to receive the app identifier" << std::endl;
return err(); return err();
} }
// close self connection after handshake is done // close self connection after handshake is done
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
std::cerr << "close connection to self immediately" << std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return err();
} }
...@@ -362,11 +374,16 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -362,11 +374,16 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
if (tbl_.lookup_hdl(hdr.source_node)) { if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a " CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node)); "direct connection: " << CAF_ARG(hdr.source_node));
std::cerr << "close connection since we already have a "
"direct connection: " << to_string(hdr.source_node)
<< std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return err();
} }
// add direct route to this node and remove any indirect entry // add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
std::cerr << "new direct connection: " << to_string(hdr.source_node)
<< std::endl;
tbl_.add(dm.handle, hdr.source_node); tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node); // auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response // write handshake as client in response
...@@ -375,11 +392,10 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -375,11 +392,10 @@ 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); write_udp_server_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::server_handshake: { case message_type::server_handshake: {
...@@ -740,9 +756,53 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -740,9 +756,53 @@ void instance::write_client_handshake(execution_unit* ctx,
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(CAF_ARG(port));
using namespace detail;
published_actor* pa = nullptr;
if (port) {
auto i = published_actors_.find(*port);
if (i != published_actors_.end())
pa = &i->second;
}
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& ref = callee_.system().config().middleman_app_identifier;
auto e = sink(const_cast<std::string&>(ref));
if (e)
return e;
//if (pa) {
// auto i = pa->first ? pa->first->id() : invalid_actor_id;
// return sink(i, pa->second);
//} else {
auto aid = invalid_actor_id;
std::set<std::string> tmp;
return sink(aid, tmp);
//}
});
header hdr{message_type::udp_client_handshake, 0, 0, version, header hdr{message_type::udp_client_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id}; this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr); write(ctx, buf, hdr, &writer);
}
void instance::write_udp_server_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) -> error {
auto& str = callee_.system().config().middleman_app_identifier;
return sink(const_cast<std::string&>(str));
});
header hdr{message_type::udp_server_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id};
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,
......
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