mirror of
https://github.com/basiliscos/syncspirit.git
synced 2026-06-01 17:47:32 +00:00
272 lines
10 KiB
C++
272 lines
10 KiB
C++
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
// SPDX-FileCopyrightText: 2019-2026 Ivan Baidakou
|
|
|
|
#include "test_supervisor.h"
|
|
#include "model/diff/load/commit.h"
|
|
#include "model/diff/modify/upsert_folder.h"
|
|
#include "model/diff/modify/upsert_folder_info.h"
|
|
#include "model/diff/advance/advance.h"
|
|
#include "model/diff/peer/update_folder.h"
|
|
#include "presentation/folder_entity.h"
|
|
#include "proto/proto-helpers-bep.h"
|
|
#include "proto/proto-helpers-db.h"
|
|
#include "net/names.h"
|
|
|
|
namespace to {
|
|
struct queue {};
|
|
struct on_timer_trigger {};
|
|
} // namespace to
|
|
|
|
template <> inline auto &rotor::supervisor_t::access<to::queue>() noexcept { return queue; }
|
|
|
|
namespace rotor {
|
|
|
|
template <>
|
|
inline auto rotor::actor_base_t::access<to::on_timer_trigger, request_id_t, bool>(request_id_t request_id,
|
|
bool cancelled) noexcept {
|
|
on_timer_trigger(request_id, cancelled);
|
|
}
|
|
|
|
} // namespace rotor
|
|
|
|
using namespace syncspirit::net;
|
|
using namespace syncspirit::test;
|
|
using namespace syncspirit::presentation;
|
|
|
|
supervisor_t::supervisor_t(config_t &cfg) : parent_t(cfg) {
|
|
auto_finish = cfg.auto_finish;
|
|
auto_ack_io = cfg.auto_ack_io;
|
|
make_presentation = cfg.make_presentation;
|
|
configure_callback = cfg.configure_callback;
|
|
sequencer = model::make_sequencer(1234);
|
|
}
|
|
|
|
void supervisor_t::configure(r::plugin::plugin_base_t &plugin) noexcept {
|
|
parent_t::configure(plugin);
|
|
plugin.with_casted<r::plugin::address_maker_plugin_t>([&](auto &p) {
|
|
p.set_identity(std::string(names::coordinator) + ".test", false);
|
|
log = utils::get_logger(identity);
|
|
});
|
|
plugin.with_casted<r::plugin::registry_plugin_t>(
|
|
[&](auto &p) { p.register_name(names::coordinator, get_address()); });
|
|
plugin.with_casted<r::plugin::starter_plugin_t>([&](auto &p) {
|
|
p.subscribe_actor(&supervisor_t::on_model_update);
|
|
p.subscribe_actor(&supervisor_t::on_model_subscribe);
|
|
p.subscribe_actor(&supervisor_t::on_model_unsubscribe);
|
|
p.subscribe_actor(&supervisor_t::on_package);
|
|
p.subscribe_actor(&supervisor_t::on_io);
|
|
});
|
|
if (configure_callback) {
|
|
configure_callback(plugin);
|
|
}
|
|
}
|
|
|
|
void supervisor_t::do_start_timer(const r::pt::time_duration &, r::timer_handler_base_t &handler) noexcept {
|
|
timers.emplace_back(&handler);
|
|
}
|
|
|
|
void supervisor_t::do_cancel_timer(r::request_id_t timer_id) noexcept {
|
|
auto it = timers.begin();
|
|
while (it != timers.end()) {
|
|
auto &handler = *it;
|
|
if (handler->request_id == timer_id) {
|
|
auto &actor_ptr = handler->owner;
|
|
actor_ptr->access<to::on_timer_trigger, r::request_id_t, bool>(timer_id, true);
|
|
on_timer_trigger(timer_id, true);
|
|
timers.erase(it);
|
|
return;
|
|
} else {
|
|
++it;
|
|
}
|
|
}
|
|
assert(0 && "should not happen");
|
|
}
|
|
|
|
void supervisor_t::do_invoke_timer(r::request_id_t timer_id) noexcept {
|
|
LOG_DEBUG(log, "{}, invoking timer {}", identity, timer_id);
|
|
auto predicate = [&](auto &handler) { return handler->request_id == timer_id; };
|
|
auto it = std::find_if(timers.begin(), timers.end(), predicate);
|
|
assert(it != timers.end());
|
|
auto &handler = *it;
|
|
auto &actor_ptr = handler->owner;
|
|
actor_ptr->access<to::on_timer_trigger, r::request_id_t, bool>(timer_id, false);
|
|
timers.erase(it);
|
|
}
|
|
|
|
void supervisor_t::start() noexcept {}
|
|
void supervisor_t::shutdown() noexcept { do_shutdown(); }
|
|
|
|
void supervisor_t::enqueue(r::message_ptr_t message) noexcept {
|
|
locality_leader->access<to::queue>().emplace_back(std::move(message));
|
|
}
|
|
|
|
void supervisor_t::on_model_subscribe(model::message::model_subscription_t &message) noexcept {
|
|
model_subscribers.push_back(message.payload);
|
|
}
|
|
|
|
void supervisor_t::on_model_unsubscribe(model::message::model_unsubscription_t &message) noexcept {
|
|
auto &item = static_cast<const model::payload::model_subscription_t &>(message.payload);
|
|
auto it = std::find(model_subscribers.begin(), model_subscribers.end(), item);
|
|
if (it != model_subscribers.end()) {
|
|
model_subscribers.erase(it);
|
|
}
|
|
}
|
|
|
|
void supervisor_t::on_model_update(model::message::model_update_t &msg) noexcept {
|
|
LOG_TRACE(log, "updating model");
|
|
auto &diff = msg.payload.diff;
|
|
auto r = diff->apply(*this, {});
|
|
if (!r) {
|
|
LOG_ERROR(log, "error updating model: {}", r.assume_error().message());
|
|
do_shutdown(make_error(r.assume_error()));
|
|
}
|
|
|
|
r = diff->visit(*this, nullptr);
|
|
if (!r) {
|
|
LOG_ERROR(log, "error visiting model: {}", r.assume_error().message());
|
|
do_shutdown(make_error(r.assume_error()));
|
|
}
|
|
auto context = model::payload::apply_context_t(&msg, const_cast<void *>(msg.payload.custom));
|
|
for (auto &subscriber : model_subscribers) {
|
|
subscriber.fn(*diff, context, subscriber.custom);
|
|
}
|
|
}
|
|
|
|
void supervisor_t::on_package(bouncer::message::package_t &msg) noexcept {
|
|
LOG_TRACE(log, "on package");
|
|
put(std::move(msg.payload));
|
|
}
|
|
|
|
void supervisor_t::on_io(fs::message::io_commands_t &message) noexcept {
|
|
for (auto &cmd : message.payload.commands) {
|
|
std::visit([&](auto &cmd) { process_io(cmd); }, cmd);
|
|
}
|
|
}
|
|
|
|
auto supervisor_t::consume_errors() noexcept -> io_errors_t { return std::move(io_errors); }
|
|
|
|
auto supervisor_t::operator()(const model::diff::local::io_failure_t &diff, void *custom) noexcept
|
|
-> outcome::result<void> {
|
|
auto &errs = diff.errors;
|
|
std::copy(errs.begin(), errs.end(), std::back_inserter(io_errors));
|
|
return diff.visit_next(*this, custom);
|
|
}
|
|
|
|
auto supervisor_t::operator()(const model::diff::modify::upsert_folder_t &diff, void *custom) noexcept
|
|
-> outcome::result<void> {
|
|
if (make_presentation) {
|
|
auto folder_id = db::get_id(diff.db);
|
|
auto folder = cluster->get_folders().by_id(folder_id);
|
|
if (!folder->get_augmentation()) {
|
|
auto folder_entity = folder_entity_ptr_t(new folder_entity_t(folder));
|
|
folder->set_augmentation(folder_entity);
|
|
}
|
|
}
|
|
return diff.visit_next(*this, custom);
|
|
}
|
|
|
|
auto supervisor_t::operator()(const model::diff::modify::upsert_folder_info_t &diff, void *custom) noexcept
|
|
-> outcome::result<void> {
|
|
auto r = diff.visit_next(*this, custom);
|
|
if (make_presentation) {
|
|
auto &folder = *cluster->get_folders().by_id(diff.folder_id);
|
|
auto &device = *cluster->get_devices().by_sha256(diff.device_id);
|
|
auto folder_info = folder.is_shared_with(device);
|
|
if (&device != cluster->get_device()) {
|
|
auto augmentation = folder.get_augmentation().get();
|
|
auto folder_entity = static_cast<presentation::folder_entity_t *>(augmentation);
|
|
folder_entity->on_insert(*folder_info);
|
|
}
|
|
}
|
|
return r;
|
|
}
|
|
|
|
auto supervisor_t::operator()(const model::diff::advance::advance_t &diff, void *custom) noexcept
|
|
-> outcome::result<void> {
|
|
if (make_presentation) {
|
|
auto folder = cluster->get_folders().by_id(diff.folder_id);
|
|
auto augmentation = folder->get_augmentation().get();
|
|
auto folder_entity = static_cast<presentation::folder_entity_t *>(augmentation);
|
|
if (folder_entity) {
|
|
auto &folder_infos = folder->get_folder_infos();
|
|
auto local_fi = folder_infos.by_device(*cluster->get_device());
|
|
auto file_name = proto::get_name(diff.proto_local);
|
|
auto local_file = local_fi->get_file_infos().by_name(file_name);
|
|
if (local_file) {
|
|
folder_entity->on_insert(*local_file, *local_fi);
|
|
}
|
|
}
|
|
}
|
|
return diff.visit_next(*this, custom);
|
|
}
|
|
|
|
auto supervisor_t::operator()(const model::diff::peer::update_folder_t &diff, void *custom) noexcept
|
|
-> outcome::result<void> {
|
|
if (make_presentation) {
|
|
auto folder = cluster->get_folders().by_id(diff.folder_id);
|
|
auto folder_aug = folder->get_augmentation().get();
|
|
auto folder_entity = static_cast<presentation::folder_entity_t *>(folder_aug);
|
|
|
|
auto &devices_map = cluster->get_devices();
|
|
auto peer = devices_map.by_sha256(diff.peer_id);
|
|
auto folder_info = folder->get_folder_infos().by_device(*peer);
|
|
auto &files_map = folder_info->get_file_infos();
|
|
|
|
for (auto &file : diff.files) {
|
|
auto file_name = proto::get_name(file);
|
|
auto file_info = files_map.by_name(file_name);
|
|
auto augmentation = file_info->get_augmentation().get();
|
|
if (!augmentation) {
|
|
folder_entity->on_insert(*file_info, *folder_info);
|
|
}
|
|
}
|
|
}
|
|
return diff.visit_next(*this, custom);
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::block_request_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io, requesting on '{}' (offset: {}, size: {})", req.path.string(), req.offset,
|
|
req.block_size);
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::remote_copy_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io (ack: {}), remote_copy_t of '{} ({} bytes)'", auto_ack_io, req.path.string(), req.size);
|
|
if (auto_ack_io) {
|
|
req.result = outcome::success();
|
|
}
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::append_block_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io (ack: {}), append_block_t of {}", auto_ack_io, req.path.string());
|
|
if (auto_ack_io) {
|
|
req.result = outcome::success();
|
|
}
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::finish_file_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io (ack: {}), finish_file_t of {}", auto_ack_io, req.path.string());
|
|
if (auto_ack_io) {
|
|
req.result = outcome::success();
|
|
}
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::clone_block_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io (ack: {}), clone_block_t, {} bytes, {}(#{}) -> {}(#{})", auto_ack_io, req.block_size,
|
|
req.source.string(), req.source_offset, req.path.string(), req.target_offset);
|
|
if (auto_ack_io) {
|
|
req.result = outcome::success();
|
|
}
|
|
}
|
|
|
|
void supervisor_t::process_io(fs::payload::update_meta_t &req) noexcept {
|
|
LOG_TRACE(log, "process_io (ack: {}), update_meta_t of {}", auto_ack_io, req.path.string());
|
|
if (auto_ack_io) {
|
|
req.result = outcome::success();
|
|
}
|
|
}
|
|
|
|
auto supervisor_t::apply(const model::diff::load::commit_t &message, void *) noexcept -> outcome::result<void> {
|
|
put(message.commit_message);
|
|
return outcome::success();
|
|
}
|