<commit_before>#[macro_use]
extern crate diesel;

#[macro_use]
extern crate lazy_static;

mod db;
mod homectl_core;
mod integrations;
mod utils;

// use db::{actions::find_floorplans, establish_connection};
use anyhow::{Context, Result};
use async_std::{prelude::*, task};
use homectl_core::{
    devices::Devices,
    events::*,
    groups::Groups,
    integration::IntegrationActionDescriptor,
    integrations::Integrations,
    rules::Rules,
    scene::{CycleScenesDescriptor, SceneDescriptor},
    scenes::Scenes,
};
use std::error::Error;

#[derive(Clone)]
struct State {
    integrations: Integrations,
    groups: Groups,
    scenes: Scenes,
    devices: Devices,
    rules: Rules,
}

#[async_std::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let (config, opaque_integrations_configs) = homectl_core::config::read_config()?;

    // println!("Using config:");
    // println!("{:#?}", config);

    let (sender, mut receiver) = mk_channel();

    let mut integrations = Integrations::new(sender.clone());
    let groups = Groups::new(config.groups);
    let scenes = Scenes::new(config.scenes, groups);
    let devices = Devices::new(sender.clone(), scenes);
    let rules = Rules::new(config.routines, sender.clone());

    for (id, integration_config) in &config.integrations {
        let opaque_integration_config: &config::Value = opaque_integrations_configs
            .get(id)
            .with_context(|| format!("Expected to find config for integration with id {}", id))?;

        integrations
            .load_integration(&integration_config.plugin, id, opaque_integration_config)
            .await?;
    }

    let _: Result<()> = {
        integrations.run_register_pass().await?;
        integrations.run_start_pass().await?;

        Ok(())
    };

    // let state = State {
    //     integrations: Arc::new(Mutex::new(integrations)),
    //     groups,
    //     scenes,
    //     devices: Arc::new(Mutex::new(devices)),
    //     rules: Arc::new(Mutex::new(rules)),
    // };

    loop {
        let msg = receiver
            .next()
            .await
            .expect("Expected sender end of channel to never be dropped");

        // println!("got msg: {:#?}", msg);

        // TODO: Need to figure out a way of not locking such large chunks of
        // state across .await points

        // let state = state.clone();

        // Maybe instead of cloning all these structs we should pass state
        // around in an Arc<Mutex<>> and only lock it when necessary
        let mut devices = devices.clone();
        let rules = rules.clone();
        let mut integrations = integrations.clone();

        task::spawn(async move {
            let result: Result<()> = match &msg {
                Message::IntegrationDeviceRefresh { device } => {
                    devices.handle_integration_device_refresh(device).await;
                    Ok(())
                }
                Message::DeviceUpdate {
                    old_state,
                    new_state,
                    old,
                    new,
                } => {
                    rules
                        .handle_device_update(old_state, new_state, old, new)
                        .await;

                    Ok(())
                }
                Message::SetDeviceState { device, set_scene } => {
                    devices.set_device_state(&device, *set_scene).await;

                    Ok(())
                }
                Message::SetIntegrationDeviceState { device } => {
                    integrations.set_integration_device_state(&device).await
                }
                Message::ActivateScene(SceneDescriptor { scene_id }) => {
                    devices.activate_scene(&scene_id).await;

                    Ok(())
                }
                Message::CycleScenes(CycleScenesDescriptor { scenes }) => {
                    devices.cycle_scenes(&scenes).await;

                    Ok(())
                }
                Message::RunIntegrationAction(IntegrationActionDescriptor {
                    integration_id,
                    payload,
                }) => {
                    integrations
                        .run_integration_action(integration_id, payload)
                        .await
                }
            };

            if let Err(err) = result {
                println!("Error while handling message:");
                println!("Msg: {:#?}", msg);
                println!("Error: {:#?}", err);
            }
        });
    }
}<commit_msg>Clone only structs necessary for handling msg<commit_after>#[macro_use]
extern crate diesel;

#[macro_use]
extern crate lazy_static;

mod db;
mod homectl_core;
mod integrations;
mod utils;

// use db::{actions::find_floorplans, establish_connection};
use anyhow::{Context, Result};
use async_std::{prelude::*, task};
use homectl_core::{
    devices::Devices,
    events::*,
    groups::Groups,
    integration::IntegrationActionDescriptor,
    integrations::Integrations,
    rules::Rules,
    scene::{CycleScenesDescriptor, SceneDescriptor},
    scenes::Scenes,
};
use std::{error::Error, sync::Arc};

#[derive(Clone)]
struct State {
    integrations: Integrations,
    groups: Groups,
    scenes: Scenes,
    devices: Devices,
    rules: Rules,
}

#[async_std::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let (config, opaque_integrations_configs) = homectl_core::config::read_config()?;

    // println!("Using config:");
    // println!("{:#?}", config);

    let (sender, mut receiver) = mk_channel();

    let mut integrations = Integrations::new(sender.clone());
    let groups = Groups::new(config.groups);
    let scenes = Scenes::new(config.scenes, groups.clone());
    let devices = Devices::new(sender.clone(), scenes.clone());
    let rules = Rules::new(config.routines, sender.clone());

    for (id, integration_config) in &config.integrations {
        let opaque_integration_config: &config::Value = opaque_integrations_configs
            .get(id)
            .with_context(|| format!("Expected to find config for integration with id {}", id))?;

        integrations
            .load_integration(&integration_config.plugin, id, opaque_integration_config)
            .await?;
    }

    let _: Result<()> = {
        integrations.run_register_pass().await?;
        integrations.run_start_pass().await?;

        Ok(())
    };

    let state = State {
        integrations,
        groups,
        scenes,
        devices,
        rules,
    };

    let state = Arc::new(state);

    loop {
        let msg = receiver
            .next()
            .await
            .expect("Expected sender end of channel to never be dropped");

        // println!("got msg: {:#?}", msg);

        // TODO: Need to figure out a way of not locking such large chunks of
        // state across .await points

        // let state = state.clone();

        // Maybe instead of cloning all these structs we should pass state
        // around to functions in an Arc<Mutex<>> and only lock it when
        // necessary
        let state = Arc::clone(&state);

        task::spawn(async move {
            let result: Result<()> = match &msg {
                Message::IntegrationDeviceRefresh { device } => {
                    let mut devices = state.devices.clone();
                    devices.handle_integration_device_refresh(device).await;
                    Ok(())
                }
                Message::DeviceUpdate {
                    old_state,
                    new_state,
                    old,
                    new,
                } => {
                    state.rules
                        .handle_device_update(old_state, new_state, old, new)
                        .await;

                    Ok(())
                }
                Message::SetDeviceState { device, set_scene } => {
                    let mut devices = state.devices.clone();
                    devices.set_device_state(&device, *set_scene).await;

                    Ok(())
                }
                Message::SetIntegrationDeviceState { device } => {
                    let mut integrations = state.integrations.clone();
                    integrations.set_integration_device_state(&device).await
                }
                Message::ActivateScene(SceneDescriptor { scene_id }) => {
                    let mut devices = state.devices.clone();
                    devices.activate_scene(&scene_id).await;

                    Ok(())
                }
                Message::CycleScenes(CycleScenesDescriptor { scenes }) => {
                    let mut devices = state.devices.clone();
                    devices.cycle_scenes(&scenes).await;

                    Ok(())
                }
                Message::RunIntegrationAction(IntegrationActionDescriptor {
                    integration_id,
                    payload,
                }) => {
                    let mut integrations = state.integrations.clone();
                    integrations
                        .run_integration_action(integration_id, payload)
                        .await
                }
            };

            if let Err(err) = result {
                println!("Error while handling message:");
                println!("Msg: {:#?}", msg);
                println!("Error: {:#?}", err);
            }
        });
    }
}
{'fraction_non_alphanumeric': 0.12087445611800912, 'fraction_numerical': 0.0, 'mean_word_length': 1.760398359695372, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 31, 'https://': 0, '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': '2673485', 'n_tokens_mistral': 2538, 'n_tokens_neox': 2430, 'n_words': 547}