Commit 51f336ad authored by Dominik Charousset's avatar Dominik Charousset

Implement proxy lookups and actor messaging

parent eed1e92b
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#pragma once #pragma once
#include <cstdint> #include <cstdint>
#include <memory>
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
...@@ -33,6 +34,7 @@ ...@@ -33,6 +34,7 @@
#include "caf/net/basp/message_type.hpp" #include "caf/net/basp/message_type.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -52,6 +54,12 @@ public: ...@@ -52,6 +54,12 @@ public:
using write_packet_callback = callback<byte_span, byte_span>; using write_packet_callback = callback<byte_span, byte_span>;
using proxy_registry_ptr = std::shared_ptr<proxy_registry>;
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry_ptr proxies);
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
template <class Parent> template <class Parent>
...@@ -135,6 +143,9 @@ private: ...@@ -135,6 +143,9 @@ private:
error handle_handshake(write_packet_callback& write_packet, header hdr, error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload); byte_span payload);
error handle_actor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_request(write_packet_callback& write_packet, header hdr, error handle_resolve_request(write_packet_callback& write_packet, header hdr,
byte_span payload); byte_span payload);
...@@ -172,6 +183,9 @@ private: ...@@ -172,6 +183,9 @@ private:
/// Ascending ID generator for requests to our peer. /// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1; uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry_ptr proxies_;
}; };
} // namespace basp } // namespace basp
......
...@@ -42,6 +42,11 @@ namespace caf { ...@@ -42,6 +42,11 @@ namespace caf {
namespace net { namespace net {
namespace basp { namespace basp {
application::application(proxy_registry_ptr proxies)
: proxies_(std::move(proxies)) {
// nop
}
expected<std::vector<byte>> application::serialize(actor_system& sys, expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) { const type_erased_tuple& x) {
std::vector<byte> result; std::vector<byte> result;
...@@ -152,6 +157,8 @@ error application::handle(write_packet_callback& write_packet, header hdr, ...@@ -152,6 +157,8 @@ error application::handle(write_packet_callback& write_packet, header hdr,
switch (hdr.type) { switch (hdr.type) {
case message_type::handshake: case message_type::handshake:
return ec::unexpected_handshake; return ec::unexpected_handshake;
case message_type::actor_message:
return handle_actor_message(write_packet, hdr, payload);
case message_type::resolve_request: case message_type::resolve_request:
return handle_resolve_request(write_packet, hdr, payload); return handle_resolve_request(write_packet, hdr, payload);
case message_type::resolve_response: case message_type::resolve_response:
...@@ -180,6 +187,32 @@ error application::handle_handshake(write_packet_callback&, header hdr, ...@@ -180,6 +187,32 @@ error application::handle_handshake(write_packet_callback&, header hdr,
return none; return none;
} }
error application::handle_actor_message(write_packet_callback&, header hdr,
byte_span payload) {
// Deserialize payload.
actor_id src;
actor_id dst;
std::vector<strong_actor_ptr> fwd_stack;
message content;
binary_deserializer source{system(), payload};
if (auto err = source(src, dst, fwd_stack, content))
return err;
// Sanity checks.
if (dst == 0)
return ec::invalid_payload;
// Try to fetch the receiver.
auto src_hdl = system().registry().get(dst);
if (src_hdl == nullptr) {
CAF_LOG_DEBUG("no actor found for given ID, drop message");
return caf::none;
}
// Ship the message.
src_hdl->get()->eq_impl(make_message_id(hdr.operation_data),
proxies_->get_or_put(peer_id_, src), nullptr,
std::move(content));
return none;
}
error application::handle_resolve_request(write_packet_callback& write_packet, error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) { header hdr, byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_request); CAF_ASSERT(hdr.type == message_type::resolve_request);
...@@ -227,11 +260,11 @@ error application::handle_resolve_response(write_packet_callback&, header hdr, ...@@ -227,11 +260,11 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
if (auto err = source(aid, ifs)) if (auto err = source(aid, ifs))
return err; return err;
if (aid == 0) { if (aid == 0) {
i->second.deliver(strong_actor_ptr{nullptr}); i->second.deliver(strong_actor_ptr{nullptr}, std::move(ifs));
return none; return none;
} }
// TODO: generate proxies and deal with proxy_registry i->second.deliver(proxies_->get_or_put(peer_id_, aid), std::move(ifs));
return ec::unimplemented; return none;
} }
error application::generate_handshake() { error application::generate_handshake() {
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <vector> #include <vector>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/basp/connection_state.hpp" #include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
...@@ -40,10 +41,10 @@ using namespace caf::net; ...@@ -40,10 +41,10 @@ using namespace caf::net;
namespace { namespace {
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
using buffer_type = std::vector<byte>; using buffer_type = std::vector<byte>;
fixture() { fixture() : app(std::make_shared<proxy_registry>(sys, *this)) {
REQUIRE_OK(app.init(*this)); REQUIRE_OK(app.init(*this));
uri mars_uri; uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri)); REQUIRE_OK(parse("tcp://mars", mars_uri));
...@@ -104,6 +105,17 @@ struct fixture : test_coordinator_fixture<> { ...@@ -104,6 +105,17 @@ struct fixture : test_coordinator_fixture<> {
return sys; return sys;
} }
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
using impl_type = forwarding_actor_proxy;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &sys, cfg, self);
}
void set_last_hop(node_id*) override {
// nop
}
buffer_type input; buffer_type input;
buffer_type output; buffer_type output;
...@@ -199,6 +211,19 @@ CAF_TEST(repeated handshake) { ...@@ -199,6 +211,19 @@ CAF_TEST(repeated handshake) {
basp::ec::unexpected_handshake); basp::ec::unexpected_handshake);
} }
CAF_TEST(actor message) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
CAF_REQUIRE_EQUAL(self->mailbox().size(), 0u);
MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!"));
allow((atom_value, strong_actor_ptr),
from(_).to(self).with(atom("monitor"), _));
expect((std::string), from(_).to(self).with("hello world!"));
}
CAF_TEST(resolve request without result) { CAF_TEST(resolve request without result) {
handle_handshake(); handle_handshake();
consume_handshake(); consume_handshake();
...@@ -242,6 +267,39 @@ CAF_TEST(resolve request on name with result) { ...@@ -242,6 +267,39 @@ CAF_TEST(resolve request on name with result) {
CAF_CHECK(ifs.empty()); CAF_CHECK(ifs.empty());
} }
CAF_TEST(resolve response with invalid actor handle) {
handle_handshake();
consume_handshake();
app.resolve(*this, "foo/bar", self);
std::string path;
RECEIVE(basp::message_type::resolve_request, 1u, path);
CAF_CHECK_EQUAL(path, "foo/bar");
actor_id aid = 0;
std::set<std::string> ifs;
MOCK(basp::message_type::resolve_response, 1u, aid, ifs);
self->receive([&](strong_actor_ptr& hdl, std::set<std::string>& hdl_ifs) {
CAF_CHECK_EQUAL(hdl, nullptr);
CAF_CHECK_EQUAL(ifs, hdl_ifs);
});
}
CAF_TEST(resolve response with valid actor handle) {
handle_handshake();
consume_handshake();
app.resolve(*this, "foo/bar", self);
std::string path;
RECEIVE(basp::message_type::resolve_request, 1u, path);
CAF_CHECK_EQUAL(path, "foo/bar");
actor_id aid = 42;
std::set<std::string> ifs;
MOCK(basp::message_type::resolve_response, 1u, aid, ifs);
self->receive([&](strong_actor_ptr& hdl, std::set<std::string>& hdl_ifs) {
CAF_REQUIRE(hdl != nullptr);
CAF_CHECK_EQUAL(ifs, hdl_ifs);
CAF_CHECK_EQUAL(hdl->id(), aid);
});
}
CAF_TEST(heartbeat message) { CAF_TEST(heartbeat message) {
handle_handshake(); handle_handshake();
consume_handshake(); consume_handshake();
......
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