path: root/crates/core/tedge_core/src/reactor.rs

use std::path::Path;
use std::sync::Arc;

use futures::StreamExt;

use itertools::Itertools;
use tedge_api::plugin::BuiltPlugin;
use tedge_api::PluginExt;
use tokio::sync::mpsc::channel;
use tokio::sync::RwLock;
use tokio_util::sync::CancellationToken;
use tracing::debug;
use tracing::error;
use tracing::info_span;
use tracing::trace;
use tracing::trace_span;
use tracing::warn;
use tracing::Instrument;

use crate::communication::CorePluginDirectory;
use crate::communication::PluginDirectory;
use crate::communication::PluginInfo;
use crate::configuration::InstanceConfiguration;
use crate::configuration::PluginInstanceConfiguration;
use crate::configuration::PluginKind;
use crate::core_task::CoreInternalMessage;
use crate::core_task::CorePlugin;
use crate::errors::PluginBuilderInstantiationError;

use crate::errors::PluginConfigurationNotFoundError;
use crate::errors::PluginInstantiationError;
use crate::errors::PluginKindUnknownError;

use crate::errors::TedgeApplicationError;
use crate::plugin_task::PluginTask;
use crate::TedgeApplication;

/// Helper type for running a TedgeApplication
///
/// This type is only introduced for more seperation-of-concerns in the codebase
/// `Reactor::run()` is simply `TedgeApplication::run()`.
pub struct Reactor(pub(crate) TedgeApplication);

impl std::fmt::Debug for Reactor {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

/// Helper type for preparing a PluginTask
struct PluginTaskPrep {
    name: String,
    plugin: BuiltPlugin,
    max_concurrency: usize,
    plugin_msg_comms: tedge_api::address::MessageSender,
    cancellation_token: CancellationToken,
}

impl Reactor {
    /// Run the application
    ///
    /// This function implements running the application. That includes the complete lifecycle of
    /// the application, the plugins that need to be started and stopped accordingly as well as
    /// crash safety concerns.
    pub async fn run(self) -> Result<(), TedgeApplicationError> {
        let max_concurrency = self.0.config().max_concurrency().get();

        // find all PluginBuilder objects that are registered and specified in the configuration to
        // be used to build Plugin instances with.
        //
        // This is then collected into a CorePluginDirectory, our "addressbook type" that can be
        // used to retrieve addresses for message passing.
        let mut directory = tracing::debug_span!("core.build_plugin_directory").in_scope(|| {
            let directory_iter = self.0.config().plugins().iter().map(|(pname, pconfig)| {
                // fetch the types the plugin claims to handle from the plugin builder identified
                // by the "kind" in the configuration of the instance
                let handle_types = self
                    .0
                    .plugin_builders()
                    .get(pconfig.kind().as_ref())
                    .map(|(handle_types, _)| handle_types.get_types().to_vec())
                    .ok_or_else(|| {
                        PluginInstantiationError::KindNotFound(PluginKindUnknownError {
                            name: pconfig.kind().as_ref().to_string(),
                            alternatives: None,
                        })
                    })?;

                Ok((
                    pname.to_string(),
                    PluginInfo::new(handle_types, max_concurrency),
                ))
            });

            CorePluginDirectory::collect_from(directory_iter)
        })?;

        // Start preparing the plugin instantiation...
        let plugin_instantiation_prep = tracing::debug_span!("core.plugin_instantiation_prep")
            .in_scope(|| {
                self.0
                    .config()
                    .plugins()
                    .iter()
                    .map(|(pname, pconfig)| {
                        let receiver = match directory
                            .get_mut(pname)
                            .map(|pinfo| pinfo.communicator.clone())
                        {
                            Some(receiver) => receiver,
                            None => unreachable!(
                            "Could not find existing plugin. This is a FATAL bug, please report it"
                        ),
                        };

                        (pname, pconfig, receiver)
                    })
                    .collect::<Vec<_>>()
            });

        let directory = Arc::new(directory);

        // ... and then instantiate the plugins requested by the user
        let (mut instantiated_plugins, failed_instantiations): (Vec<PluginTaskPrep>, Vec<_>) =
            plugin_instantiation_prep
                .into_iter()
                .map(|(pname, pconfig, communicator)| {
                    {
                        self.instantiate_plugin(
                            pname,
                            self.0.config_path(),
                            pconfig,
                            directory.clone(),
                            communicator,
                            self.0.cancellation_token().child_token(),
                        )
                    }
                    .instrument(info_span!("plugin.instantiate", name = %pname))
                })
                .collect::<futures::stream::FuturesUnordered<_>>()
                .collect::<Vec<Result<_, _>>>()
                .instrument(tracing::debug_span!("core.plugin_instantiation"))
                .await
                .into_iter()
                .partition_result();
        trace!("Plugins instantiated");

        if !failed_instantiations.is_empty() {
            return Err(TedgeApplicationError::PluginInstantiationsError {
                errors: failed_instantiations,
            });
        }

        // Now we need to make sure we start the "CoreTask", which is responsible for handling the
        // communication within the core itself.
        let (internal_sender, mut internal_receiver) = channel(10);
        let core_plugin = CorePlugin::new(internal_sender);
        instantiated_plugins.push(PluginTaskPrep {
            name: "core".to_string(),
            plugin: core_plugin.finish(),
            max_concurrency: 10,
            plugin_msg_comms: directory.get_core_communicator(),
            cancellation_token: self.0.cancellation_token.clone(),
        });

        debug!("Core task instantiated");

        let mut all_plugins: Vec<PluginTask> = instantiated_plugins
            .into_iter()
            .map(|prep| {
                let timeout = self.0.config().plugin_shutdown_timeout();
                let plugin = Arc::new(RwLock::new(prep.plugin));
                PluginTask::new(
                    prep.name,
                    plugin,
                    prep.max_concurrency,
                    prep.plugin_msg_comms,
                    prep.cancellation_token,
                    timeout,
                )
            })
            .collect();

        debug!("Running 'start' for plugins");
        let (_start_results_ok, start_results_err): (Vec<_>, Vec<_>) = all_plugins
            .iter_mut()
            .map(|plugin_task| {
                let span =
                    tracing::debug_span!("plugin.start", plugin.name = %plugin_task.plugin_name());
                plugin_task.run_start().instrument(span)
            })
            .collect::<futures::stream::FuturesOrdered<_>>()
            .collect::<Vec<Result<(), _>>>()
            .instrument(tracing::info_span!("core.mainloop.plugins.start"))
            .await
            .into_iter()
            .partition_result();

        if !start_results_err.is_empty() {
            return Err(TedgeApplicationError::PluginLifecycleErrors {
                errors: start_results_err,
            });
        }

        debug!("Enabling communications for plugins");
        all_plugins
            .iter()
            .map(|plugin_task| {
                let span =
                    tracing::debug_span!("plugin.enable_communication", plugin.name = %plugin_task.plugin_name());
                plugin_task.enable_communications().instrument(span)
            })
            .collect::<futures::stream::FuturesOrdered<_>>()
            .collect::<Vec<Result<(), _>>>()
            .instrument(tracing::info_span