Commit 39ca8e49 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Implement down and monitor messages

parent bcbb9f5e
...@@ -81,7 +81,18 @@ public: ...@@ -81,7 +81,18 @@ public:
template <class Parent> template <class Parent>
error write_message(Parent& parent, error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) { std::unique_ptr<endpoint_manager::message> ptr) {
// TODO: avoid extra copy of the payload // There is one special case: actor_proxy_impl sends a message without
// mailbox element on construction to trigger monitoring messages.
if (ptr->msg == nullptr) {
CAF_ASSERT(ptr->payload.empty());
CAF_ASSERT(ptr->receiver != nullptr);
CAF_ASSERT(ptr->receiver->node() == peer_id_);
header hdr{message_type::monitor_message, 0,
static_cast<uint64_t>(ptr->receiver->id())};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), span<const byte>{});
return none;
}
buf_.clear(); buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_}; serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender; const auto& src = ptr->msg->sender;
...@@ -97,6 +108,7 @@ public: ...@@ -97,6 +108,7 @@ public:
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err; return err;
} }
// TODO: avoid extra copy of the payload
buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end()); buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end());
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()), header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
ptr->msg->mid.integer_value()}; ptr->msg->mid.integer_value()};
...@@ -177,6 +189,9 @@ private: ...@@ -177,6 +189,9 @@ private:
error handle_resolve_response(write_packet_callback& write_packet, header hdr, error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload); byte_span payload);
error handle_down_message(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();
......
...@@ -113,7 +113,8 @@ public: ...@@ -113,7 +113,8 @@ public:
using queue_type = intrusive::drr_queue<message_policy>; using queue_type = intrusive::drr_queue<message_policy>;
task_size_type task_size(const message& x) const noexcept { task_size_type task_size(const message& x) const noexcept {
return x.payload.size(); // Return at least 1 if the payload is empty.
return x.payload.size() + static_cast<task_size_type>(x.payload.empty());
} }
}; };
......
...@@ -27,7 +27,8 @@ namespace net { ...@@ -27,7 +27,8 @@ namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst) actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) { : super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// nop CAF_ASSERT(dst_ != nullptr);
dst_->enqueue(nullptr, ctrl(), {});
} }
actor_proxy_impl::~actor_proxy_impl() { actor_proxy_impl::~actor_proxy_impl() {
......
...@@ -155,6 +155,8 @@ error application::handle(write_packet_callback& write_packet, header hdr, ...@@ -155,6 +155,8 @@ error application::handle(write_packet_callback& write_packet, header hdr,
return handle_resolve_response(write_packet, hdr, payload); return handle_resolve_response(write_packet, hdr, payload);
case message_type::heartbeat: case message_type::heartbeat:
return none; return none;
case message_type::down_message:
return handle_down_message(write_packet, hdr, payload);
default: default:
return ec::unimplemented; return ec::unimplemented;
} }
...@@ -271,6 +273,16 @@ error application::handle_resolve_response(write_packet_callback&, header hdr, ...@@ -271,6 +273,16 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
return none; return none;
} }
error application::handle_down_message(write_packet_callback&, header hdr,
byte_span payload) {
error reason;
binary_deserializer source{system(), payload};
if (auto err = source(reason))
return err;
proxies_->erase(peer_id_, hdr.operation_data, std::move(reason));
return none;
}
error application::generate_handshake() { error application::generate_handshake() {
buf_.clear(); buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_}; serializer_impl<buffer_type> sink{system(), buf_};
......
...@@ -164,37 +164,52 @@ struct fixture : test_coordinator_fixture<>, ...@@ -164,37 +164,52 @@ struct fixture : test_coordinator_fixture<>,
stream_socket sock; stream_socket sock;
basp::application* app; basp::application* app;
unit_t no_payload;
}; };
} // namespace } // namespace
#define MOCK(kind, op, ...) \ #define MOCK(kind, op, ...) \
do { \ do { \
auto payload = to_buf(__VA_ARGS__); \ CAF_MESSAGE("mock " << kind); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \ if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
write(sock, make_span(payload)); \ std::tuple<unit_t>>::value) { \
auto payload = to_buf(__VA_ARGS__); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
write(sock, make_span(payload)); \
} else { \
mock(basp::header{kind, 0, op}); \
} \
run(); \ run(); \
} while (false) } while (false)
#define RECEIVE(msg_type, op_data, ...) \ #define RECEIVE(msg_type, op_data, ...) \
do { \ do { \
CAF_MESSAGE("receive " << msg_type); \
buffer_type buf(basp::header_size); \ buffer_type buf(basp::header_size); \
if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \ if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \ CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \ auto hdr = basp::header::from_bytes(buf); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \ CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \ CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
buf.resize(hdr.payload_len); \ if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
if (fetch_size(read(sock, make_span(buf))) != size_t{hdr.payload_len}) \ std::tuple<unit_t>>::value) { \
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \ buf.resize(hdr.payload_len); \
binary_deserializer source{sys, buf}; \ if (fetch_size(read(sock, make_span(buf))) != size_t{hdr.payload_len}) \
if (auto err = source(__VA_ARGS__)) \ CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
CAF_FAIL("failed to receive data: " << sys.render(err)); \ binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
} else { \
if (hdr.payload_len != 0) \
CAF_FAIL("unexpected payload"); \
} \
} while (false) } while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture) CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(actor message) { CAF_TEST(actor message and down message) {
handle_handshake(); handle_handshake();
consume_handshake(); consume_handshake();
sys.registry().put(self->id(), self); sys.registry().put(self->id(), self);
...@@ -202,9 +217,19 @@ CAF_TEST(actor message) { ...@@ -202,9 +217,19 @@ CAF_TEST(actor message) {
MOCK(basp::message_type::actor_message, make_message_id().integer_value(), MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{}, mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!")); make_message("hello world!"));
allow((atom_value, strong_actor_ptr), MOCK(basp::message_type::monitor_message, 42u, no_payload);
from(_).to(self).with(atom("monitor"), _)); strong_actor_ptr proxy;
expect((std::string), from(_).to(self).with("hello world!")); self->receive([&](const std::string& str) {
CAF_CHECK_EQUAL(str, "hello world!");
proxy = self->current_sender();
CAF_REQUIRE_NOT_EQUAL(proxy, nullptr);
self->monitor(proxy);
});
MOCK(basp::message_type::down_message, 42u,
error{exit_reason::user_shutdown});
expect((down_msg),
from(_).to(self).with(down_msg{actor_cast<actor_addr>(proxy),
exit_reason::user_shutdown}));
} }
CAF_TEST(resolve request without result) { CAF_TEST(resolve request without result) {
......
...@@ -100,6 +100,9 @@ public: ...@@ -100,6 +100,9 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) { std::unique_ptr<net::endpoint_manager::message> msg) {
// Ignore proxy announcement messages.
if (msg->msg == nullptr)
return;
header_type header{static_cast<uint32_t>(msg->payload.size())}; header_type header{static_cast<uint32_t>(msg->payload.size())};
std::vector<byte> payload(msg->payload.begin(), msg->payload.end()); std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload)); parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
......
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