<commit_before>/// @date 14.10.2020
/// @copyright
/// This Source Code Form is subject to the terms of the Mozilla Public License,
/// v. 2.0. If a copy of the MPL was not distributed with this file,
/// You can obtain one at https://mozilla.org/MPL/2.0/

#include <bs/tree/map_link.h>
#include <bs/meta.h>
#include <bs/log.h>

#include <bs/serialize/cafbind.h>
#include <bs/serialize/tree.h>

#include "map_engine.h"
#include "nil_engine.h"

#include <algorithm>
#include <memory_resource>

NAMESPACE_BEGIN(blue_sky::tree)
using ei = engine::impl;

// setup synchronized pool allocator for link impls
static auto impl_pool = std::pmr::synchronized_pool_resource{};
static auto map_link_alloc = std::pmr::polymorphic_allocator<map_link_impl>(&impl_pool);
static auto map_node_alloc = std::pmr::polymorphic_allocator<map_node_impl>(&impl_pool);

map_link::map_link(
	mapper_f mf, uuid tag, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) : super([&] {
		const auto make_lmapper_impl = [&](link_mapper_f lmf) -> sp_engine_impl {
			return std::allocate_shared<map_link_impl>(
				map_link_alloc, std::move(lmf), tag, std::move(name), src_node, dest_node, update_on, opts, f
			);
		};
		const auto make_nmapper_impl = [&](node_mapper_f nmf) -> sp_engine_impl {
			return std::allocate_shared<map_node_impl>(
				map_node_alloc, std::move(nmf), tag, std::move(name), src_node, dest_node, update_on, opts, f
			);
		};

		return visit(meta::overloaded{

			[&](link_mapper_f lmf) -> sp_engine_impl { return make_lmapper_impl(std::move(lmf)); },

			[&](simple_link_mapper_f lmf) -> sp_engine_impl {
				return make_lmapper_impl([lmf = std::move(lmf)](link src, link dest, caf::event_based_actor*) {
					return caf::result<link>{ lmf(src, dest) };
				});
			},

			[&](node_mapper_f nmf) -> sp_engine_impl { return make_nmapper_impl(std::move(nmf)); },

			[&](simple_node_mapper_f nmf) -> sp_engine_impl {
				return make_nmapper_impl([nmf = std::move(nmf)](node src, node dest, caf::event_based_actor*) {
					nmf(src, dest);
					return caf::result<void>{};
				});
			},

		}, std::move(mf));
	}())
{}

map_link::map_link(
	mapper_f mf, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) : map_link(
	std::move(mf), gen_uuid(), std::move(name), std::move(src_node), std::move(dest_node), update_on, opts, f
) {}


map_link::map_link(mapper_f mf, const map_link& rhs, link_or_node src_node, link_or_node dest_node) :
	map_link(
		std::move(mf), ei::pimpl(rhs).tag_, ei::pimpl(rhs).name_, std::move(src_node),
		dest_node.index() == 0 || std::get<1>(dest_node).is_nil() ?
			ei::pimpl(rhs).out_ : std::get<1>(dest_node),
		ei::pimpl(rhs).update_on_, ei::pimpl(rhs).opts_, ei::pimpl(rhs).flags_
	)
{
	// ensure that dest node is relinked to this new link
	pimpl()->propagate_handle(ei::pimpl(*this).out_);
	rs_reset(Req::DataNode, rhs.req_status(Req::DataNode));
}

map_link::map_link(const link& rhs) : super(rhs, type_id_()) {}

auto map_link::type_id_() -> std::string_view { return "map_link"; }

auto map_link::tag() const -> uuid {
	return ei::pimpl(*this).tag_;
}

auto map_link::input() const -> node {
	return ei::pimpl<map_impl_base>(*this).in_;
}

auto map_link::output() const -> node {
	return ei::pimpl<map_impl_base>(*this).out_;
}

auto map_link::l_target() const -> const link_mapper_f* {
	auto& simpl = ei::pimpl<map_impl_base>(*this);
	if(simpl.is_link_mapper)
		return &ei::pimpl<map_link_impl>(*this).mf_;
	return nullptr;
}

auto map_link::n_target() const -> const node_mapper_f* {
	auto& simpl = ei::pimpl<map_impl_base>(*this);
	if(!simpl.is_link_mapper)
		return &ei::pimpl<map_node_impl>(*this).mf_;
	return nullptr;
}

/*-----------------------------------------------------------------------------
 *  bundled mappers
 *-----------------------------------------------------------------------------*/
auto make_otid_filter(
	std::vector<std::string> allowed_otids, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) -> map_link {
	std::sort(allowed_otids.begin(), allowed_otids.end());
	// [NOTE] dest link is ignored
	return map_link(
		[otids = std::move(allowed_otids)](link src, link /* dest */, caf::event_based_actor* worker)
		-> caf::result<link> {
			auto res = worker->make_response_promise<link>();
			auto src_actor = src.actor();

			worker->request(src_actor, kernel::radio::timeout(), a_lnk_otid{})
			.then([=](const std::string& otid) mutable {
				if(std::binary_search(otids.begin(), otids.end(), otid))
					res.delegate(src_actor, a_clone{}, false);
				else
					res.deliver(link{});
			});
			return link{};
		}, std::move(name), std::move(src_node), std::move(dest_node), update_on, opts, f
	);
}

NAMESPACE_END(blue_sky::tree)<commit_msg>map_link/otid_filter: output contain weak links to target object<commit_after>/// @date 14.10.2020
/// @copyright
/// This Source Code Form is subject to the terms of the Mozilla Public License,
/// v. 2.0. If a copy of the MPL was not distributed with this file,
/// You can obtain one at https://mozilla.org/MPL/2.0/

#include <bs/tree/map_link.h>
#include <bs/meta.h>
#include <bs/log.h>

#include <bs/serialize/cafbind.h>
#include <bs/serialize/tree.h>

#include "map_engine.h"
#include "nil_engine.h"

#include <algorithm>
#include <memory_resource>

NAMESPACE_BEGIN(blue_sky::tree)
using ei = engine::impl;

// setup synchronized pool allocator for link impls
static auto impl_pool = std::pmr::synchronized_pool_resource{};
static auto map_link_alloc = std::pmr::polymorphic_allocator<map_link_impl>(&impl_pool);
static auto map_node_alloc = std::pmr::polymorphic_allocator<map_node_impl>(&impl_pool);

map_link::map_link(
	mapper_f mf, uuid tag, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) : super([&] {
		const auto make_lmapper_impl = [&](link_mapper_f lmf) -> sp_engine_impl {
			return std::allocate_shared<map_link_impl>(
				map_link_alloc, std::move(lmf), tag, std::move(name), src_node, dest_node, update_on, opts, f
			);
		};
		const auto make_nmapper_impl = [&](node_mapper_f nmf) -> sp_engine_impl {
			return std::allocate_shared<map_node_impl>(
				map_node_alloc, std::move(nmf), tag, std::move(name), src_node, dest_node, update_on, opts, f
			);
		};

		return visit(meta::overloaded{

			[&](link_mapper_f lmf) -> sp_engine_impl { return make_lmapper_impl(std::move(lmf)); },

			[&](simple_link_mapper_f lmf) -> sp_engine_impl {
				return make_lmapper_impl([lmf = std::move(lmf)](link src, link dest, caf::event_based_actor*) {
					return caf::result<link>{ lmf(src, dest) };
				});
			},

			[&](node_mapper_f nmf) -> sp_engine_impl { return make_nmapper_impl(std::move(nmf)); },

			[&](simple_node_mapper_f nmf) -> sp_engine_impl {
				return make_nmapper_impl([nmf = std::move(nmf)](node src, node dest, caf::event_based_actor*) {
					nmf(src, dest);
					return caf::result<void>{};
				});
			},

		}, std::move(mf));
	}())
{}

map_link::map_link(
	mapper_f mf, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) : map_link(
	std::move(mf), gen_uuid(), std::move(name), std::move(src_node), std::move(dest_node), update_on, opts, f
) {}


map_link::map_link(mapper_f mf, const map_link& rhs, link_or_node src_node, link_or_node dest_node) :
	map_link(
		std::move(mf), ei::pimpl(rhs).tag_, ei::pimpl(rhs).name_, std::move(src_node),
		dest_node.index() == 0 || std::get<1>(dest_node).is_nil() ?
			ei::pimpl(rhs).out_ : std::get<1>(dest_node),
		ei::pimpl(rhs).update_on_, ei::pimpl(rhs).opts_, ei::pimpl(rhs).flags_
	)
{
	// ensure that dest node is relinked to this new link
	pimpl()->propagate_handle(ei::pimpl(*this).out_);
	rs_reset(Req::DataNode, rhs.req_status(Req::DataNode));
}

map_link::map_link(const link& rhs) : super(rhs, type_id_()) {}

auto map_link::type_id_() -> std::string_view { return "map_link"; }

auto map_link::tag() const -> uuid {
	return ei::pimpl(*this).tag_;
}

auto map_link::input() const -> node {
	return ei::pimpl<map_impl_base>(*this).in_;
}

auto map_link::output() const -> node {
	return ei::pimpl<map_impl_base>(*this).out_;
}

auto map_link::l_target() const -> const link_mapper_f* {
	auto& simpl = ei::pimpl<map_impl_base>(*this);
	if(simpl.is_link_mapper)
		return &ei::pimpl<map_link_impl>(*this).mf_;
	return nullptr;
}

auto map_link::n_target() const -> const node_mapper_f* {
	auto& simpl = ei::pimpl<map_impl_base>(*this);
	if(!simpl.is_link_mapper)
		return &ei::pimpl<map_node_impl>(*this).mf_;
	return nullptr;
}

/*-----------------------------------------------------------------------------
 *  bundled mappers
 *-----------------------------------------------------------------------------*/
auto make_otid_filter(
	std::vector<std::string> allowed_otids, std::string name, link_or_node src_node, link_or_node dest_node,
	Event update_on, TreeOpts opts, Flags f
) -> map_link {
	namespace kradio = kernel::radio;
	std::sort(allowed_otids.begin(), allowed_otids.end());
	// [NOTE] dest link is ignored
	return map_link(
		[otids = std::move(allowed_otids)](link src, link /* dest */, caf::event_based_actor* worker)
		-> caf::result<link> {
			auto res = worker->make_response_promise<link>();
			auto src_actor = src.actor();

			worker->set_error_handler([=](auto* worker, auto& er) mutable {
				res.deliver(link{});
				worker->default_error_handler(worker, er);
			});

			worker->request(src_actor, kradio::timeout(true), a_data{}, true)
			.then([=](obj_or_errbox maybe_obj) mutable {
				if(maybe_obj && std::binary_search(otids.begin(), otids.end(), (*maybe_obj)->type_id()))
					worker->request(src_actor, kradio::timeout(), a_lnk_name{})
					.then([=, obj = *std::move(maybe_obj)](std::string src_name) mutable {
						res.deliver(link{ weak_link(std::move(src_name), obj) });
					});
				else
					res.deliver(link{});
			});
			return res;
		}, std::move(name), std::move(src_node), std::move(dest_node), update_on, opts, f
	);
}

NAMESPACE_END(blue_sky::tree)
{'fraction_non_alphanumeric': 0.1935007385524372, 'fraction_numerical': 0.0029542097488921715, 'mean_word_length': 4.49269875608437, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 89, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1180635', 'n_tokens_mistral': 4024, 'n_tokens_neox': 3758, 'n_words': 728}