feat: implement source tail

src/lib.rs

@@ -1,4 +1,6 @@
 pub mod config;
 pub mod filter;
+pub mod parser;
 pub mod record;
 pub mod sink;
+pub mod source;

src/parser.rs

@@ -0,0 +1,28 @@
+use crate::record::LogLine;
+use serde_json::json;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+pub async fn run_parser(mut rx: Receiver<String>, tx: Sender<LogLine>) {
+    while let Some(line) = rx.recv().await {
+        let log_line = parse_line(&line);
+        if tx.send(log_line).await.is_err() {
+            break;
+        }
+    }
+}
+
+pub fn parse_line(line: &str) -> LogLine {
+    match serde_json::from_str::<LogLine>(line) {
+        Ok(value) if value.is_object() => value,
+
+        Ok(value) => {
+            eprintln!("logtap: line was valid JSON, but is not object: {value}");
+            json!({ "message": line, "level": "UNKNOWN", "parse_issue": "not_an_object" })
+        }
+
+        Err(err) => {
+            eprintln!("logtap: falied to process as JSON: {err}");
+            json!({ "message": line, "level": "UNKNOWN", "parse_issue": "invalid_json" })
+        }
+    }
+}

src/source.rs

@@ -0,0 +1,36 @@
+use std::fs::File;
+use std::io::{BufRead, BufReader, Seek, SeekFrom};
+use tokio::time::{Duration, sleep};
+
+pub async fn run_source(
+    cfg: crate::config::Config,
+    tx: tokio::sync::mpsc::Sender<String>,
+) -> anyhow::Result<()> {
+    let mut offset: u64 = 0;
+
+    loop {
+        let file = File::open(&cfg.source_path)?;
+        let mut reader = BufReader::new(file);
+
+        reader.seek(SeekFrom::Start(offset))?;
+
+        let mut line = String::new();
+        loop {
+            line.clear();
+            let bytes_read = reader.read_line(&mut line)?;
+
+            if bytes_read == 0 {
+                break;
+            }
+
+            offset += bytes_read as u64;
+            let trimmed = line.trim_end().to_string();
+
+            if tx.send(trimmed).await.is_err() {
+                return Ok(());
+            }
+        }
+
+        sleep(Duration::from_millis(500)).await;
+    }
+}

tests/parser_test.rs

@@ -0,0 +1,52 @@
+use logtap::parser::parse_line;
+use serde_json::Value;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_run_parse() {
+        let line = r#"{"level": "info", "msg": "server started"}"#;
+
+        let result = parse_line(line);
+
+        assert_eq!(
+            result.get("level"),
+            Some(&Value::String("info".to_string()))
+        );
+        assert_eq!(
+            result.get("msg"),
+            Some(&Value::String("server started".to_string()))
+        );
+    }
+
+    #[test]
+    fn parseline_with_text() {
+        let line = "isso aqui nao e json";
+
+        let result = parse_line(line);
+
+        assert_eq!(
+            result.get("message"),
+            Some(&Value::String(line.to_string()))
+        );
+        assert_eq!(
+            result.get("level"),
+            Some(&Value::String("UNKNOWN".to_string()))
+        );
+        assert_eq!(
+            result.get("parse_issue"),
+            Some(&Value::String("invalid_json".to_string()))
+        );
+    }
+
+    #[test]
+    fn parse_empty_json_line_returns_empty_object() {
+        let line = "{}";
+
+        let result = parse_line(line);
+
+        assert!(result.as_object().is_some_and(|obj| obj.is_empty()));
+    }
+}

tests/source_test.rs

@@ -0,0 +1,27 @@
+use std::path::PathBuf;
+
+use logtap::config::Config;
+use logtap::source::run_source;
+use tokio::sync::mpsc;
+
+#[tokio::test]
+async fn test_run_source() {
+    let (tx, mut rx) = mpsc::channel::<String>(100);
+
+    let cfg = Config {
+        source_path: PathBuf::from("teste.log"),
+        sink_url: "http://localhost:8080".to_string(),
+        batch_size: 100,
+        flush_interval_secs: 5,
+        channel_capacity: 100,
+        filter_rules: vec![],
+    };
+
+    tokio::spawn(async move {
+        run_source(cfg, tx).await.unwrap();
+    });
+
+    while let Some(line) = rx.recv().await {
+        println!("{line}");
+    }
+}