bayou 0.1.3

iot real-time analytics
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

use bayou::commands::Cli;
use bayou::config::Settings;

#[macro_use]
extern crate log;

use chrono::prelude::*;
use clap::Parser;
use file_rotate::suffix::{DateFrom, FileLimit};
use rumqttc::{AsyncClient, ConnectionError, Incoming, MqttOptions, QoS};
// use config::Config;
use file_rotate::{compression::Compression, suffix::AppendTimestamp, ContentLimit, FileRotate, TimeFrequency};
// use tokio_cron_scheduler::{Job, JobScheduler, JobSchedulerError};
use std::io::Write;
use std::process::exit;
use std::str;
use std::time::Duration;

async fn mqtt_stream_topic(cfg: &Settings) {
  let mut mqttoptions = MqttOptions::new("bayou", &cfg.mqtt.host, cfg.mqtt.port);
  mqttoptions.set_keep_alive(Duration::from_secs(5));

  info!(
    "[{}] Starting listeners on [{}:{}]",
    cfg.mqtt.topic, cfg.mqtt.host, cfg.mqtt.port
  );
  let mut log = FileRotate::new(
    &cfg.logs.path,
    // AppendCount::new(10),
    AppendTimestamp::with_format("%Y%m%dT%H", FileLimit::MaxFiles(cfg.logs.files), DateFrom::Now),
    ContentLimit::Time(TimeFrequency::Hourly),
    Compression::OnRotate(1),
    None,
  );

  let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
  client.subscribe(&cfg.mqtt.topic, QoS::AtMostOnce).await.unwrap();
  info!("[{}] Subscribed", cfg.mqtt.topic);

  loop {
    let res = eventloop.poll().await;
    match res {
      Ok(notification) => {
        match notification {
          rumqttc::Event::Incoming(Incoming::Publish(msg)) => {
            let s = match str::from_utf8(&msg.payload) {
              Ok(v) => v,
              Err(e) => panic!("Invalid UTF-8 sequence: {}", e),
            };
            // let msg: Packet = notification as Packet;
            // info!("[{}] {}", msg.topic, s);
            writeln!(log, "{} {} {}", Utc::now(), msg.topic, s).unwrap();
          }
          _ => {
            // println!("NOTIFICATION {:?}", notification);
          }
        };
      }
      Err(ConnectionError::Io(error)) => {
        if error.kind() == std::io::ErrorKind::ConnectionAborted
          || error.kind() == std::io::ErrorKind::ConnectionRefused
          || error.kind() == std::io::ErrorKind::ConnectionReset
        {
          println!("Failed to connect to the server. Error: {error:?}");
          return;
        }
        println!("Connection IO error: {error:?}");
        exit(1);
      }
      Err(ConnectionError::NetworkTimeout) => {
        println!("Network timeout");
      }
      Err(ce) => {
        println!("Connection Err  {}", ce);
        exit(1);
      }
    }
  }
}

// async fn s3_archive(cfg: &Settings) -> Result<(), JobSchedulerError> {
//     let mut sched = JobScheduler::new().await?;
//     sched.add(
//         Job::new("1/10 * * * * *", |_uuid, _l| {
//             println!("I run every 10 seconds");
//         })?
//     ).await?;
//     sched.start().await?;

//     // Wait while the jobs run
//     tokio::time::sleep(Duration::from_secs(100)).await;

//     Ok(())
// }

#[tokio::main]
async fn main() {
  // RUST_LOG=debug
  env_logger::init();
  let cli = Cli::parse();
  let settings = Settings::new(cli.config).unwrap();
  info!("{:?}", settings);

  mqtt_stream_topic(&settings).await;
}