<commit_before>/// @author <NAME> (@uentity)
/// @date 03.02.2021
/// @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/
#pragma once

#include <bs/python/common.h>
#include <bs/python/result_converter.h>
#include <bs/actor_common.h>
#include <bs/detail/function_view.h>
#include <bs/serialize/cafbind.h>
#include <bs/serialize/propdict.h>

#include "../kernel/radio_subsyst.h"

#include <tuple>

NAMESPACE_BEGIN(blue_sky::python)

template<typename F, typename R, typename... Args, typename... LaunchAsync>
auto pipe_queue_impl(F f, const identity<R (Args...)> _, LaunchAsync... async_tag) {
	return [f = std::make_shared<F>(std::move(f))](Args&&... args) {
		KRADIO.enqueue(
			LaunchAsync{}...,
			[f, argtup = std::make_tuple(std::forward<Args>(args)...)]() mutable {
				std::apply(*f, std::move(argtup));
				return perfect;
			}
		);
	};
};

// run any given callable through kernel's queue
template<typename F, typename... LaunchAsync>
auto pipe_through_queue(F f, LaunchAsync... async_tag) {
	return pipe_queue_impl(std::move(f), identity< deduce_callable_t<F> >{}, async_tag...);
};

// run Python transaction (applied to link/object) through queue
template<typename... Ts>
auto pytr_through_queue(std::function< py::object(Ts...) > tr) {
	return [tr = make_result_converter<tr_result>(std::move(tr), perfect)]
	(caf::event_based_actor* papa, Ts... args) mutable -> caf::result<tr_result::box> {
		// [NOTE] using request.await to stop messages processing while tr is executed
		auto res = papa->make_response_promise<tr_result::box>();
		papa->request(
			KRADIO.queue_actor(), kernel::radio::timeout(true),
			transaction{[tr = std::move(tr), argstup = std::make_tuple(std::forward<Ts>(args)...)]
			() mutable {
				return std::apply(std::move(tr), std::move(argstup));
			}}
		).await(
			[=](tr_result::box tres) mutable { res.deliver(std::move(tres)); },
			[=](const caf::error& er) mutable { res.deliver(pack(tr_result{ forward_caf_error(er) })); }
		);
		return res;
	};
}

NAMESPACE_END(blue_sky::python)<commit_msg>python/queue: tr -> queue decorator also pass callback inside shared_ptr<commit_after>/// @author <NAME> (@uentity)
/// @date 03.02.2021
/// @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/
#pragma once

#include <bs/python/common.h>
#include <bs/python/result_converter.h>
#include <bs/actor_common.h>
#include <bs/detail/function_view.h>
#include <bs/serialize/cafbind.h>
#include <bs/serialize/propdict.h>

#include "../kernel/radio_subsyst.h"

#include <tuple>

NAMESPACE_BEGIN(blue_sky::python)

template<typename F, typename R, typename... Args, typename... LaunchAsync>
auto pipe_queue_impl(F f, const identity<R (Args...)> _, LaunchAsync... async_tag) {
	return [f = std::make_shared<F>(std::move(f))](Args&&... args) {
		KRADIO.enqueue(
			LaunchAsync{}...,
			[f, argtup = std::make_tuple(std::forward<Args>(args)...)]() mutable {
				std::apply(*f, std::move(argtup));
				return perfect;
			}
		);
	};
};

// run any given callable through kernel's queue
template<typename F, typename... LaunchAsync>
auto pipe_through_queue(F f, LaunchAsync... async_tag) {
	return pipe_queue_impl(std::move(f), identity< deduce_callable_t<F> >{}, async_tag...);
};

// run Python transaction (applied to link/object) through queue
template<typename... Ts>
auto pytr_through_queue(std::function< py::object(Ts...) > tr) {
	const auto make_ctr = [&] {
		return make_result_converter<tr_result>(std::move(tr), perfect);
	};
	using F = decltype(make_ctr());

	return [tr = std::make_shared<F>(make_ctr())]
	(caf::event_based_actor* papa, Ts... args) mutable -> caf::result<tr_result::box> {
		// [NOTE] using request.await to stop messages processing while tr is executed
		auto res = papa->make_response_promise<tr_result::box>();
		papa->request(
			KRADIO.queue_actor(), kernel::radio::timeout(true),
			transaction{[tr = std::move(tr), argstup = std::make_tuple(std::forward<Ts>(args)...)]
			() mutable {
				auto r = std::apply(std::move(*tr), std::move(argstup));
				tr.reset();
				return r;
			}}
		).await(
			[=](tr_result::box tres) mutable { res.deliver(std::move(tres)); },
			[=](const caf::error& er) mutable { res.deliver(pack(tr_result{ forward_caf_error(er) })); }
		);
		return res;
	};
}

NAMESPACE_END(blue_sky::python)
{'fraction_non_alphanumeric': 0.18140343254399305, 'fraction_numerical': 0.005213990875515968, 'mean_word_length': 4.553679131483715, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 53, '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': '499998', 'n_tokens_mistral': 1726, 'n_tokens_neox': 1605, 'n_words': 367}