caryatid_module_playback/
playback.rs

1//! Caryatid Playback module
2
3use anyhow::{Error, Result};
4use caryatid_sdk::{module, Context, MessageBounds};
5use config::Config;
6use std::fs::read_to_string;
7use std::path::{Path, PathBuf};
8use std::sync::Arc;
9use tracing::{error, info};
10
11/// Playback module
12/// Parameterised by the outer message enum used on the bus
13#[module(message_type(M), name = "playback", description = "Message playback")]
14pub struct Playback<M: MessageBounds + for<'a> serde::Deserialize<'a>>;
15
16impl<M: MessageBounds + for<'a> serde::Deserialize<'a>> Playback<M> {
17    async fn init(&self, context: Arc<Context<M>>, config: Arc<Config>) -> Result<()> {
18        match config.get_string("topic") {
19            Ok(topic) => {
20                match config.get_string("path") {
21                    Ok(path) => {
22                        let path = PathBuf::from(&path);
23                        let dir = Path::new(&path);
24                        if !dir.is_dir() {
25                            error!("Path does not exist or is not directory for Playback module");
26                            return Err(Error::msg(
27                                "Path does not exist or is not directory for Playback module",
28                            ));
29                        }
30                        info!("Creating message playback on '{}'", topic);
31                        context.run(Self::play_files(context.clone(), topic, path));
32                    }
33                    _ => error!("No path given for Playback module"),
34                };
35            }
36
37            _ => error!("No topic given for Playback module - no effect"),
38        }
39
40        Ok(())
41    }
42
43    async fn play_files(context: Arc<Context<M>>, topic: String, path: PathBuf) -> Result<()> {
44        let mut num: u64 = 0;
45        loop {
46            let filename = path.join(format!("{}.json", num));
47            match read_to_string(&filename) {
48                Ok(file) => {
49                    match serde_json::from_str::<Arc<M>>(&file) {
50                        Ok(message) => {
51                            if let Err(error) = context.message_bus.publish(&topic, message).await {
52                                error!("Failed to publish message: {}", error);
53                            }
54                        }
55                        Err(error) => {
56                            error!(
57                                "Failed to playback message to file {:?}: {}. Aborting playback",
58                                &filename, error
59                            );
60                            break;
61                        }
62                    };
63                }
64                Err(error) => {
65                    error!(
66                        "Failed to playback message to file {:?}: {}. Aborting playback",
67                        &filename, error
68                    );
69                    break;
70                }
71            };
72            num += 1;
73        }
74        Ok(())
75    }
76}
caryatid_module_playback/playback.rs

caryatid_module_playback/
playback.rs
