Commit bf53a25f authored by Dominik Charousset's avatar Dominik Charousset

Implement scaffold for resolving paths

parent 7fa556fa
...@@ -18,12 +18,14 @@ ...@@ -18,12 +18,14 @@
#pragma once #pragma once
#include <cstdint>
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp" #include "caf/error.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"
...@@ -33,6 +35,7 @@ ...@@ -33,6 +35,7 @@
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
...@@ -44,6 +47,10 @@ public: ...@@ -44,6 +47,10 @@ public:
using buffer_type = std::vector<byte>; using buffer_type = std::vector<byte>;
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
template <class Parent> template <class Parent>
...@@ -55,7 +62,7 @@ public: ...@@ -55,7 +62,7 @@ public:
return err; return err;
auto hdr = to_bytes(header{message_type::handshake, auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version}); static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(hdr, buf_); parent.write_packet(parent, hdr, buf_, unit);
return none; return none;
} }
...@@ -66,17 +73,25 @@ public: ...@@ -66,17 +73,25 @@ public:
static_cast<uint32_t>(ptr->payload.size()), static_cast<uint32_t>(ptr->payload.size()),
ptr->msg->mid.integer_value()}; ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr); auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), ptr->payload); parent.write_packet(parent, make_span(bytes), ptr->payload, unit);
} }
template <class Parent> template <class Parent>
error handle_data(Parent&, span<const byte> bytes) { error handle_data(Parent& parent, byte_span bytes) {
return handle(bytes); auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(parent, hdr, payload, unit);
return none;
});
return handle(write_packet, bytes);
} }
template <class Parent> template <class Parent>
void resolve(Parent&, const std::string&, actor) { void resolve(Parent& parent, string_view path, actor listener) {
// TODO: implement me auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(parent, hdr, payload, unit);
return none;
});
resolve_remote_path(write_packet, path, listener);
} }
template <class Transport> template <class Transport>
...@@ -91,6 +106,15 @@ public: ...@@ -91,6 +106,15 @@ public:
static expected<std::vector<byte>> serialize(actor_system& sys, static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x); const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept { connection_state state() const noexcept {
return state_; return state_;
} }
...@@ -102,11 +126,19 @@ public: ...@@ -102,11 +126,19 @@ public:
private: private:
// -- message handling ------------------------------------------------------- // -- message handling -------------------------------------------------------
error handle(span<const byte> bytes); error handle(write_packet_callback& write_packet, byte_span bytes);
error handle(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle(header hdr, span<const byte> payload); error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_handshake(header hdr, span<const byte> payload); error handle_resolve_request(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
/// Writes the handshake payload to `buf_`. /// Writes the handshake payload to `buf_`.
error generate_handshake(); error generate_handshake();
...@@ -133,6 +165,12 @@ private: ...@@ -133,6 +165,12 @@ private:
/// Keeps track of which local actors our peer monitors. /// Keeps track of which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_; std::unordered_set<actor_addr> monitored_actors_;
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, response_promise> pending_resolves_;
/// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1;
}; };
} // namespace basp } // namespace basp
......
...@@ -38,8 +38,9 @@ enum class ec : uint8_t { ...@@ -38,8 +38,9 @@ enum class ec : uint8_t {
missing_handshake, missing_handshake,
unexpected_handshake, unexpected_handshake,
version_mismatch, version_mismatch,
unimplemented, unimplemented = 10,
app_identifiers_mismatch, app_identifiers_mismatch,
invalid_payload,
}; };
/// @relates ec /// @relates ec
......
...@@ -30,7 +30,9 @@ ...@@ -30,7 +30,9 @@
#include "caf/expected.hpp" #include "caf/expected.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"
#include "caf/no_stages.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/sec.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/type_erased_tuple.hpp" #include "caf/type_erased_tuple.hpp"
...@@ -47,7 +49,32 @@ expected<std::vector<byte>> application::serialize(actor_system& sys, ...@@ -47,7 +49,32 @@ expected<std::vector<byte>> application::serialize(actor_system& sys,
return result; return result;
} }
error application::handle(span<const byte> bytes) { strong_actor_ptr application::resolve_local_path(string_view) {
return nullptr;
}
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
return;
}
auto req_id = next_request_id_++;
auto hdr = to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(buf_.size()), req_id});
if (auto err = write_packet(hdr, buf_)) {
CAF_LOG_ERROR("unable to serialize path");
return;
}
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
error application::handle(write_packet_callback& write_packet,
byte_span bytes) {
switch (state_) { switch (state_) {
case connection_state::await_handshake_header: { case connection_state::await_handshake_header: {
if (bytes.size() != header_size) if (bytes.size() != header_size)
...@@ -85,7 +112,7 @@ error application::handle(span<const byte> bytes) { ...@@ -85,7 +112,7 @@ error application::handle(span<const byte> bytes) {
return ec::unexpected_number_of_bytes; return ec::unexpected_number_of_bytes;
hdr_ = header::from_bytes(bytes); hdr_ = header::from_bytes(bytes);
if (hdr_.payload_len == 0) if (hdr_.payload_len == 0)
return handle(hdr_, span<const byte>{}); return handle(write_packet, hdr_, byte_span{});
else else
state_ = connection_state::await_payload; state_ = connection_state::await_payload;
return none; return none;
...@@ -93,17 +120,23 @@ error application::handle(span<const byte> bytes) { ...@@ -93,17 +120,23 @@ error application::handle(span<const byte> bytes) {
case connection_state::await_payload: { case connection_state::await_payload: {
if (bytes.size() != hdr_.payload_len) if (bytes.size() != hdr_.payload_len)
return ec::unexpected_number_of_bytes; return ec::unexpected_number_of_bytes;
return handle(hdr_, bytes); state_ = connection_state::await_header;
return handle(write_packet, hdr_, bytes);
} }
default: default:
return ec::illegal_state; return ec::illegal_state;
} }
} }
error application::handle(header hdr, span<const byte>) { error application::handle(write_packet_callback& write_packet, header hdr,
byte_span payload) {
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::resolve_request:
return handle_resolve_request(write_packet, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
case message_type::heartbeat: case message_type::heartbeat:
return none; return none;
default: default:
...@@ -111,7 +144,8 @@ error application::handle(header hdr, span<const byte>) { ...@@ -111,7 +144,8 @@ error application::handle(header hdr, span<const byte>) {
} }
} }
error application::handle_handshake(header hdr, span<const byte> payload) { error application::handle_handshake(write_packet_callback&, header hdr,
byte_span payload) {
if (hdr.type != message_type::handshake) if (hdr.type != message_type::handshake)
return ec::missing_handshake; return ec::missing_handshake;
if (hdr.operation_data != version) if (hdr.operation_data != version)
...@@ -127,6 +161,60 @@ error application::handle_handshake(header hdr, span<const byte> payload) { ...@@ -127,6 +161,60 @@ error application::handle_handshake(header hdr, span<const byte> payload) {
return none; return none;
} }
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{system(), payload};
if (auto err = source.begin_sequence(path_size))
return err;
// We expect the payload to consist only of the path.
if (path_size != source.remaining())
return ec::invalid_payload;
auto remainder = source.remainder();
string_view path{reinterpret_cast<const char*>(remainder.data()),
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
buf_.clear();
actor_id aid = result ? result->id() : 0;
std::set<std::string> ifs;
// TODO: figure out how to obtain messaging interface.
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(aid, ifs))
return err;
auto out_hdr = to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
}
error application::handle_resolve_response(write_packet_callback&, header hdr,
byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(hdr.operation_data);
if (i == pending_resolves_.end()) {
CAF_LOG_ERROR("received unknown ID in resolve_response message");
return none;
}
auto guard = detail::make_scope_guard([&] {
if (i->second.pending())
i->second.deliver(sec::remote_lookup_failed);
pending_resolves_.erase(i);
});
actor_id aid;
std::set<std::string> ifs;
binary_deserializer source{system(), payload};
if (auto err = source(aid, ifs))
return err;
if (aid == 0) {
i->second.deliver(strong_actor_ptr{nullptr});
return none;
}
// TODO: generate proxies and deal with proxy_registry
return ec::unimplemented;
}
error application::generate_handshake() { error application::generate_handshake() {
serializer_impl<buffer_type> sink{system(), buf_}; serializer_impl<buffer_type> sink{system(), buf_};
return sink(system().node(), return sink(system().node(),
......
...@@ -39,6 +39,7 @@ const char* ec_names[] = { ...@@ -39,6 +39,7 @@ const char* ec_names[] = {
"version_mismatch", "version_mismatch",
"unimplemented", "unimplemented",
"app_identifiers_mismatch", "app_identifiers_mismatch",
"invalid_payload",
}; };
} // namespace } // namespace
......
...@@ -63,7 +63,8 @@ struct fixture : test_coordinator_fixture<> { ...@@ -63,7 +63,8 @@ struct fixture : test_coordinator_fixture<> {
input = to_buf(xs...); input = to_buf(xs...);
} }
void write_packet(span<const byte> hdr, span<const byte> payload) { void write_packet(fixture&, span<const byte> hdr, span<const byte> payload,
unit_t) {
output.insert(output.end(), hdr.begin(), hdr.end()); output.insert(output.end(), hdr.begin(), hdr.end());
output.insert(output.end(), payload.begin(), payload.end()); output.insert(output.end(), payload.begin(), payload.end());
} }
...@@ -114,6 +115,29 @@ struct fixture : test_coordinator_fixture<> { ...@@ -114,6 +115,29 @@ struct fixture : test_coordinator_fixture<> {
} // namespace } // namespace
#define MOCK(kind, op, ...) \
do { \
auto payload = to_buf(__VA_ARGS__); \
set_input(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
if (auto err = app.handle_data(*this, input)) \
CAF_FAIL("application-under-test failed to process header: " \
<< sys.render(err)); \
if (auto err = app.handle_data(*this, payload)) \
CAF_FAIL("application-under-test failed to process payload: " \
<< sys.render(err)); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
serializer_impl<buffer_type> source{sys, output}; \
basp::header hdr; \
if (auto err = source(hdr, __VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
output.clear(); \
} while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture) CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(missing handshake) { CAF_TEST(missing handshake) {
...@@ -172,6 +196,19 @@ CAF_TEST(repeated handshake) { ...@@ -172,6 +196,19 @@ CAF_TEST(repeated handshake) {
basp::ec::unexpected_handshake); basp::ec::unexpected_handshake);
} }
CAF_TEST(resolve request without result) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, std::string{"/foo/bar"});
CAF_CHECK_EQUAL(app.state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, 0u);
CAF_CHECK(ifs.empty());
}
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