Commit 30fe093d authored by Joseph Noir's avatar Joseph Noir

Enable simple remote actor test using UDP

parent 1961b3fd
......@@ -50,6 +50,9 @@ public:
/// Returns the current input buffer.
virtual std::vector<char>& rd_buf() = 0;
/// Returns the local port
virtual uint16_t local_port() const = 0;
void io_failure(execution_unit* ctx, network::operation op) override;
using dgram_doorman_base::new_endpoint;
......
......@@ -612,7 +612,7 @@ public:
inline buffer_type& rd_buf() {
return rd_buf_;
}
inline const std::string& host() const {
return host_;
}
......
......@@ -187,6 +187,9 @@ struct new_datagram_msg {
dgram_scribe_handle handle;
// Buffer containing received data.
std::vector<char> buf;
// Port of the doorman that accepted the handshake
// TODO: get rid of this because it is only sometime used
optional<uint16_t> port;
};
/// @relates new_datagram_msg
......
......@@ -302,7 +302,7 @@ uint16_t abstract_broker::local_port(dgram_scribe_handle hdl) {
uint16_t abstract_broker::local_port(dgram_doorman_handle hdl) {
auto i = dgram_doormans_.find(hdl);
return i != dgram_doormans_.end() ? i->second->port() : 0;
return i != dgram_doormans_.end() ? i->second->local_port() : 0;
}
accept_handle abstract_broker::hdl_by_port(uint16_t port) {
......
......@@ -1042,6 +1042,12 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
uint16_t port() const override {
return acceptor_.port();
}
uint16_t local_port() const override {
auto x = local_port_of_fd(acceptor_.fd());
if (!x)
return 0;
return *x;
}
void launch() override {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
......@@ -2120,6 +2126,7 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
return std::move(p.error());
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
std::cerr << "[NDDI] Bound to " << *p << std::endl;
return std::make_pair(sguard.release(), *p);
}
......
......@@ -59,16 +59,18 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx,
if (detached())
return false;
using delegate_t = new_datagram_msg; // dgram_delegate_msg;
using tmp_t = mailbox_element_vals<delegate_t>;
using tmp_t = mailbox_element_vals<delegate_t>;
auto guard = parent_;
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
buf.resize(num_bytes);
// std::vector<char> msg_buf;
// msg_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(),
mailbox_element::forwarding_stack{},
delegate_t{endpoint, std::move(buf)}};
delegate_t{endpoint, std::move(buf),
parent()->local_port(hdl())}};
invoke_mailbox_element_impl(ctx,tmp);
return true;
}
......
......@@ -363,9 +363,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
}
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
std::cerr << "[I] Before finalize handshake" << std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs);
std::cerr << "[I] After finalize handshake" << std::endl;
flush(*path);
break;
}
......@@ -400,17 +398,23 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
}
write_udp_server_handshake(ctx, path->wr_buf, hdr.source_node, none);
if (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] message_type::server_handshake" << std::endl;
std::cerr << "[I] Received TCP server_handshake (ignored)" << std::endl;
break;
}
case message_type::client_handshake: {
std::cerr << "[I] Received TCP client handshake" << std::endl;
std::cerr << "[I] Received TCP client handshake (ignored)" << std::endl;
break;
}
case message_type::dispatch_message: {
......@@ -688,39 +692,6 @@ void instance::write_udp_client_handshake(execution_unit* ctx,
header hdr{message_type::udp_client_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr, &writer);
/*
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,
this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr, &writer);
*/
}
void instance::write_udp_server_handshake(execution_unit* ctx,
......@@ -731,6 +702,7 @@ void instance::write_udp_server_handshake(execution_unit* ctx,
using namespace detail;
published_actor* pa = nullptr;
if (port) {
std::cerr << "LOOKING FOR ACTOR ON PORT " << *port << std::endl;
auto i = published_actors_.find(*port);
if (i != published_actors_.end())
pa = &i->second;
......@@ -760,16 +732,6 @@ void instance::write_udp_server_handshake(execution_unit* ctx,
pa && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id};
write(ctx, buf, hdr, &writer);
/*
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,
......
......@@ -276,7 +276,6 @@ expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
return 0;
}
uint16_t port() const override {
return mpx_->port(hdl());
}
......@@ -384,11 +383,18 @@ expected<void> test_multiplexer::assign_dgram_doorman(abstract_broker* ptr,
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
uint16_t local_port() const override {
std::cerr << "test_multiplexer::assign_dgram_scribe::impl::local_port "
<< "not implementd." << std::endl;
abort();
return 0;
}
void launch() override {
// nop
}
......
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