Commit 31434abc authored by Jakob Otto's avatar Jakob Otto

Cleanup basp::application

parent 03756ae4
......@@ -57,8 +57,6 @@ public:
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
......@@ -90,11 +88,8 @@ public:
return none;
}
template <class Parent>
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
return write(parent, std::move(ptr));
}
error write_message(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
......@@ -107,37 +102,11 @@ public:
return none;
}
// TODO: unessecary indirection
template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) {
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement `packet_writer`");
resolve_remote_path(parent, path, listener);
}
void resolve(packet_writer& writer, string_view path, const actor& listener);
// TODO: can be packet_writer&?
template <class Parent>
void new_proxy(Parent& parent, actor_id id) {
header hdr{message_type::monitor_message, 0, static_cast<uint64_t>(id)};
auto header_buf = parent.next_header_buffer();
to_bytes(hdr, header_buf);
parent.write_packet(header_buf);
}
static void new_proxy(packet_writer& writer, actor_id id);
// TODO: can be packet_writer&?
template <class Parent>
void local_actor_down(Parent& parent, actor_id id, error reason) {
auto header_buf = parent.next_header_buffer();
auto payload_buf = parent.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_buf};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
header hdr{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()),
static_cast<uint64_t>(id)};
to_bytes(hdr, header_buf);
parent.write_packet(header_buf, payload_buf);
}
void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
......@@ -148,16 +117,13 @@ public:
// nop
}
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x);
static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(packet_writer& writer, string_view path,
const actor& listener);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
......@@ -169,11 +135,6 @@ public:
}
private:
// -- handling of outgoing messages ------------------------------------------
error write(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, packet_writer& writer, byte_span bytes);
......@@ -198,7 +159,7 @@ private:
byte_span payload);
/// Writes the handshake payload to `buf_`.
error generate_handshake(std::vector<byte>& buf);
error generate_handshake(buffer_type& buf);
// -- member variables -------------------------------------------------------
......
......@@ -48,6 +48,81 @@ application::application(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
error application::write_message(
packet_writer& writer, std::unique_ptr<endpoint_manager_queue::message> ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_elem_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_elem_buf};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
// TODO: Is this size correct?
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_elem_buf.size()
+ ptr->payload.size()),
ptr->msg->mid.integer_value()},
header_buf);
// TODO: OK to move payload out of message?
writer.write_packet(header_buf, payload_elem_buf, ptr->payload);
return none;
}
void application::resolve(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return;
}
auto req_id = next_request_id_++;
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload_buf.size()), req_id},
header_buf);
writer.write_packet(header_buf, payload_buf);
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
void application::new_proxy(packet_writer& writer, actor_id id) {
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::monitor_message, 0, static_cast<uint64_t>(id)},
header_buf);
writer.write_packet(header_buf);
}
void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) {
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_buf};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()),
static_cast<uint64_t>(id)},
header_buf);
writer.write_packet(header_buf, payload_buf);
}
expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
......@@ -79,60 +154,6 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
return nullptr;
}
void application::resolve_remote_path(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return;
}
auto req_id = next_request_id_++;
to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload_buf.size()), req_id},
header_buf);
writer.write_packet(header_buf, payload_buf);
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::write(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto header_buf = writer.next_header_buffer();
auto payload_elem_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_elem_buf};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
// TODO: Is this size correct?
header hdr{message_type::actor_message,
static_cast<uint32_t>(payload_elem_buf.size()
+ ptr->payload.size()),
ptr->msg->mid.integer_value()};
to_bytes(hdr, header_buf);
// TODO: OK to move payload out of message?
writer.write_packet(header_buf, payload_elem_buf, ptr->payload);
return none;
}
error application::handle(size_t& next_read_size, packet_writer& writer,
byte_span bytes) {
CAF_LOG_TRACE(CAF_ARG(state_) << CAF_ARG2("bytes.size", bytes.size()));
......@@ -276,8 +297,6 @@ error application::handle_resolve_request(packet_writer& writer, header hdr,
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
actor_id aid;
std::set<std::string> ifs;
if (result) {
......@@ -287,9 +306,11 @@ error application::handle_resolve_request(packet_writer& writer, header hdr,
aid = 0;
}
// TODO: figure out how to obtain messaging interface.
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(aid, ifs))
return err;
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload_buf.size()),
hdr.operation_data},
......@@ -340,11 +361,11 @@ error application::handle_monitor_message(packet_writer& writer, header hdr,
});
} else {
error reason = exit_reason::unknown;
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(reason))
return err;
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()),
hdr.operation_data},
......@@ -366,7 +387,6 @@ error application::handle_down_message(packet_writer&, header hdr,
}
error application::generate_handshake(std::vector<byte>& buf) {
buf.clear(); // TODO: really necessary?
serializer_impl<buffer_type> sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
......
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