Merge pull request #493 from LnL7/lapin-stats

lapin stats

ofborg/src/bin/stats.rs

@@ -1,34 +1,32 @@
-use ofborg::easyamqp::{ChannelExt, ConsumerExt};
-use ofborg::{config, easyamqp, stats, tasks, worker};
-
 use std::env;
+use std::error::Error;
 use std::thread;
 
-use amqp::Basic;
+use async_std::task;
 use hyper::server::{Request, Response, Server};
 use tracing::info;
 
-fn main() {
+use ofborg::easyamqp::{ChannelExt, ConsumerExt};
-    let cfg = config::load(env::args().nth(1).unwrap().as_ref());
+use ofborg::{config, easyamqp, easylapin, stats, tasks};
+
+fn main() -> Result<(), Box<dyn Error>> {
     ofborg::setup_log();
 
-    info!("Hello, world!");
+    let arg = env::args().nth(1).expect("usage: stats <config>");
+    let cfg = config::load(arg.as_ref());
 
-    let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
+    let conn = easylapin::from_config(&cfg.rabbitmq)?;
-    info!("Connected to rabbitmq");
+    let mut chan = task::block_on(conn.create_channel())?;
 
-    let events = stats::RabbitMQ::from_amqp(
+    let events = stats::RabbitMQ::from_lapin(
         &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()),
-        session.open_channel(3).unwrap(),
+        task::block_on(conn.create_channel())?,
     );
 
     let metrics = stats::MetricCollector::new();
-
     let collector = tasks::statscollector::StatCollectorWorker::new(events, metrics.clone());
 
-    let mut channel = session.open_channel(1).unwrap();
+    chan.declare_exchange(easyamqp::ExchangeConfig {
-    channel
-        .declare_exchange(easyamqp::ExchangeConfig {
         exchange: "stats".to_owned(),
         exchange_type: easyamqp::ExchangeType::Fanout,
         passive: false,
@@ -36,33 +34,27 @@ fn main() {
         auto_delete: false,
         no_wait: false,
         internal: false,
-        })
+    })?;
-        .unwrap();
 
-    channel
+    let queue_name = String::from("stats-events");
-        .declare_queue(easyamqp::QueueConfig {
+    chan.declare_queue(easyamqp::QueueConfig {
-            queue: "stats-events".to_owned(),
+        queue: queue_name.clone(),
         passive: false,
         durable: true,
         exclusive: false,
         auto_delete: false,
         no_wait: false,
-        })
+    })?;
-        .unwrap();
 
-    channel
+    chan.bind_queue(easyamqp::BindQueueConfig {
-        .bind_queue(easyamqp::BindQueueConfig {
+        queue: queue_name.clone(),
-            queue: "stats-events".to_owned(),
         exchange: "stats".to_owned(),
         routing_key: None,
         no_wait: false,
-        })
+    })?;
-        .unwrap();
 
-    channel.basic_prefetch(1).unwrap();
+    let handle = chan.consume(
-    let mut channel = channel
+        collector,
-        .consume(
-            worker::new(collector),
         easyamqp::ConsumeConfig {
             queue: "stats-events".to_owned(),
             consumer_tag: format!("{}-prometheus-stats-collector", cfg.whoami()),
@@ -71,26 +63,21 @@ fn main() {
             no_wait: false,
             exclusive: false,
         },
-        )
+    )?;
-        .unwrap();
 
     thread::spawn(|| {
         let addr = "0.0.0.0:9898";
         info!("listening addr {:?}", addr);
-        Server::http(addr)
+        Server::http(addr)?.handle(move |_: Request, res: Response| {
-            .unwrap()
-            .handle(move |_: Request, res: Response| {
             res.send(metrics.prometheus_output().as_bytes()).unwrap();
-            })
+        })?;
-            .unwrap();
+        Ok::<_, Box<dyn Error + Sync + Send + '_>>(())
     });
 
-    channel.start_consuming();
+    info!("Fetching jobs from {}", &queue_name);
+    task::block_on(handle);
 
-    info!("Finished consuming?");
+    drop(conn); // Close connection.
-
-    channel.close(200, "Bye").unwrap();
-    info!("Closed the channel");
-    session.close(200, "Good Bye");
     info!("Closed the session... EOF");
+    Ok(())
 }