feat: make python module

Author: d-perl

src/main.rs src/bin/app.rssrc/main.rs renamed to src/bin/app.rs

@@ -1,17 +1,6 @@
-use std::process::exit;
-
-use tokio::sync::mpsc;
-
-mod redis_logs;
-use crate::redis_logs::{LogMsg, producer_loop};
-
-mod elastic_push;
-use crate::elastic_push::consumer_loop;
-
-mod config;
-use crate::config::IngestorConfig;
-
+use bec_log_ingestor::{config::IngestorConfig, main_loop};
 use clap::Parser;
+use std::process::exit;
 
 #[derive(clap::Parser, Debug)]
 struct Args {
@@ -37,16 +26,6 @@ fn entry() -> IngestorConfig {
     IngestorConfig::from_file(path)
 }
 
-async fn main_loop(config: IngestorConfig) {
-    println!("Starting log ingestor with config: \n {:?}", &config);
-
-    let (tx, mut rx) = mpsc::unbounded_channel::<LogMsg>();
-    let producer = tokio::spawn(producer_loop(tx, config.redis.clone()));
-    consumer_loop(&mut rx, config.elastic.clone()).await;
-
-    let _ = tokio::join!(producer);
-}
-
 #[tokio::main]
 async fn main() {
     let config = entry();

src/config.rs src/lib/config.rssrc/config.rs renamed to src/lib/config.rs

@@ -82,7 +82,10 @@ pub struct IngestorConfig {
 
 impl IngestorConfig {
     /// Parse a toml file for an IngestorConfig. Assumes the file exists and is readable.
-    pub fn from_file(path: std::path::PathBuf) -> Self {
+    pub fn from_file<P>(path: P) -> Self
+    where
+        P: AsRef<std::path::Path>,
+    {
         let mut file = std::fs::File::open(path).expect("Cannot open supplied config file!");
         let mut contents = String::new();
         file.read_to_string(&mut contents)

src/elastic_push.rs src/lib/elastic_push.rssrc/elastic_push.rs renamed to src/lib/elastic_push.rs

@@ -1,12 +1,7 @@
+use crate::{config::ElasticConfig, redis_logs::LogMsg};
 use elasticsearch::{Elasticsearch, http::request::JsonBody};
-use tokio::sync::mpsc;
-
 use std::{error::Error, iter::once};
-
-use crate::{
-    config::ElasticConfig,
-    redis_logs::{LogMsg, LogRecord},
-};
+use tokio::sync::mpsc;
 
 fn elastic_client(config: &ElasticConfig) -> Result<Elasticsearch, Box<dyn Error>> {
     let url = elasticsearch::http::Url::parse(&config.url.full_url())?;
@@ -83,7 +78,7 @@ pub async fn consumer_loop(rx: &mut mpsc::UnboundedReceiver<LogMsg>, config: Ela
 
 #[cfg(test)]
 mod tests {
-    use crate::config::UrlPort;
+    use crate::{config::UrlPort, redis_logs::LogRecord};
 
     use super::*;
     use serde::{Deserialize, Serialize};

src/example_config.toml src/lib/example_config.tomlsrc/example_config.toml renamed to src/lib/example_config.toml

@@ -0,0 +1,34 @@
+use tokio::sync::mpsc;
+pub mod config;
+pub mod elastic_push;
+pub mod redis_logs;
+use redis_logs::{LogMsg, producer_loop};
+
+use elastic_push::consumer_loop;
+
+use config::IngestorConfig;
+
+pub async fn main_loop(config: IngestorConfig) {
+    println!("Starting log ingestor with config: \n {:?}", &config);
+
+    let (tx, mut rx) = mpsc::unbounded_channel::<LogMsg>();
+    let producer = tokio::spawn(producer_loop(tx, config.redis.clone()));
+    consumer_loop(&mut rx, config.elastic.clone()).await;
+
+    let _ = tokio::join!(producer);
+}
+
+#[pyo3::pymodule]
+mod bec_log_ingestor {
+    use crate::config::IngestorConfig;
+    use pyo3::prelude::*;
+    use tokio::runtime::Runtime;
+
+    /// Run forever. Will block forever.
+    #[pyfunction]
+    fn run_with_config(filename: String) {
+        let config = IngestorConfig::from_file(filename);
+        let rt = Runtime::new().unwrap();
+        rt.block_on(super::main_loop(config));
+    }
+}

src/lib/lib.rs

@@ -1,12 +1,11 @@
+use crate::config::RedisConfig;
 use chrono::TimeZone;
 use redis::Commands;
 use rmp_serde;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 use tokio::sync::mpsc;
 
-use crate::config::RedisConfig;
-
 #[derive(Debug, PartialEq, Deserialize, Serialize, Clone)]
 pub struct Elapsed {
     pub repr: String,