Merge pull request #495 from LnL7/lapin-log-message-collector

convert log-message-collector to easylapin

ofborg/src/bin/evaluation-filter.rs

@@ -56,7 +56,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         no_wait: false,
     })?;
 
-    let handle = chan.consume(
+    let handle = easylapin::WorkerChannel(chan).consume(
         tasks::evaluationfilter::EvaluationFilterWorker::new(cfg.acl()),
         easyamqp::ConsumeConfig {
             queue: queue_name.clone(),

ofborg/src/bin/github-comment-filter.rs

@@ -57,7 +57,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         no_wait: false,
     })?;
 
-    let handle = chan.consume(
+    let handle = easylapin::WorkerChannel(chan).consume(
         tasks::githubcommentfilter::GitHubCommentWorker::new(cfg.acl(), cfg.github()),
         easyamqp::ConsumeConfig {
             queue: "build-inputs".to_owned(),

ofborg/src/bin/github-comment-poster.rs

@@ -46,7 +46,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         no_wait: false,
     })?;
 
-    let handle = chan.consume(
+    let handle = easylapin::WorkerChannel(chan).consume(
         tasks::githubcommentposter::GitHubCommentPoster::new(cfg.github_app_vendingmachine()),
         easyamqp::ConsumeConfig {
             queue: "build-results".to_owned(),

ofborg/src/bin/log-message-collector.rs

@@ -1,78 +1,73 @@
-use ofborg::config;
-use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
-use ofborg::tasks;
-use ofborg::worker;
-
 use std::env;
+use std::error::Error;
 use std::path::PathBuf;
 
+use async_std::task;
 use tracing::info;
 
-fn main() {
+use ofborg::config;
-    let cfg = config::load(env::args().nth(1).unwrap().as_ref());
+use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
+use ofborg::easylapin;
+use ofborg::tasks;
+
+fn main() -> Result<(), Box<dyn Error>> {
     ofborg::setup_log();
 
-    let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
+    let arg = env::args()
-    info!("Connected to rabbitmq");
+        .nth(1)
+        .expect("usage: log-message-collector <config>");
+    let cfg = config::load(arg.as_ref());
 
-    let mut channel = session.open_channel(1).unwrap();
+    let conn = easylapin::from_config(&cfg.rabbitmq)?;
+    let mut chan = task::block_on(conn.create_channel())?;
 
-    channel
+    chan.declare_exchange(easyamqp::ExchangeConfig {
-        .declare_exchange(easyamqp::ExchangeConfig {
+        exchange: "logs".to_owned(),
-            exchange: "logs".to_owned(),
+        exchange_type: easyamqp::ExchangeType::Topic,
-            exchange_type: easyamqp::ExchangeType::Topic,
+        passive: false,
-            passive: false,
+        durable: true,
-            durable: true,
+        auto_delete: false,
-            auto_delete: false,
+        no_wait: false,
-            no_wait: false,
+        internal: false,
-            internal: false,
+    })?;
-        })
-        .unwrap();
 
     let queue_name = "".to_owned();
-    channel
+    chan.declare_queue(easyamqp::QueueConfig {
-        .declare_queue(easyamqp::QueueConfig {
+        queue: queue_name.clone(),
+        passive: false,
+        durable: false,
+        exclusive: true,
+        auto_delete: true,
+        no_wait: false,
+    })?;
+
+    chan.bind_queue(easyamqp::BindQueueConfig {
+        queue: queue_name.clone(),
+        exchange: "logs".to_owned(),
+        routing_key: Some("*.*".to_owned()),
+        no_wait: false,
+    })?;
+
+    // Regular channel, we want prefetching here.
+    let handle = chan.consume(
+        tasks::log_message_collector::LogMessageCollector::new(
+            PathBuf::from(cfg.log_storage.clone().unwrap().path),
+            100,
+        ),
+        easyamqp::ConsumeConfig {
             queue: queue_name.clone(),
-            passive: false,
+            consumer_tag: format!("{}-log-collector", cfg.whoami()),
-            durable: false,
+            no_local: false,
-            exclusive: true,
+            no_ack: false,
-            auto_delete: true,
             no_wait: false,
-        })
+            exclusive: false,
-        .unwrap();
+        },
+    )?;
 
-    channel
+    info!("Fetching jobs from {}", &queue_name);
-        .bind_queue(easyamqp::BindQueueConfig {
+    task::block_on(handle);
-            queue: queue_name.clone(),
-            exchange: "logs".to_owned(),
-            routing_key: Some("*.*".to_owned()),
-            no_wait: false,
-        })
-        .unwrap();
 
-    let mut channel = channel
+    drop(conn); // Close connection.
-        .consume(
-            worker::new(tasks::log_message_collector::LogMessageCollector::new(
-                PathBuf::from(cfg.log_storage.clone().unwrap().path),
-                100,
-            )),
-            easyamqp::ConsumeConfig {
-                queue: queue_name,
-                consumer_tag: format!("{}-log-collector", cfg.whoami()),
-                no_local: false,
-                no_ack: false,
-                no_wait: false,
-                exclusive: false,
-            },
-        )
-        .unwrap();
-
-    channel.start_consuming();
-
-    info!("Finished consuming?");
-
-    channel.close(200, "Bye").unwrap();
-    info!("Closed the channel");
-    session.close(200, "Good Bye");
     info!("Closed the session... EOF");
+    Ok(())
 }

ofborg/src/bin/mass-rebuilder.rs

@@ -53,7 +53,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         no_wait: false,
     })?;
 
-    let handle = chan.consume(
+    let handle = easylapin::WorkerChannel(chan).consume(
         tasks::evaluate::EvaluationWorker::new(
             cloner,
             &nix,

ofborg/src/easylapin.rs

@@ -86,8 +86,6 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop<Channel> {
     type Handle = Pin<Box<dyn Future<Output = ()> + 'a>>;
 
     fn consume(self, mut worker: W, config: ConsumeConfig) -> Result<Self::Handle, Self::Error> {
-        task::block_on(self.basic_qos(1, BasicQosOptions::default()))?;
-
         let mut consumer = task::block_on(self.basic_consume(
             &config.queue,
             &config.consumer_tag,
@@ -117,6 +115,20 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop<Channel> {
     }
 }
 
+/// Same as a regular channel, but without prefetching,
+/// used for services with multiple instances.
+pub struct WorkerChannel(pub CloseOnDrop<Channel>);
+
+impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for WorkerChannel {
+    type Error = lapin::Error;
+    type Handle = Pin<Box<dyn Future<Output = ()> + 'a>>;
+
+    fn consume(self, worker: W, config: ConsumeConfig) -> Result<Self::Handle, Self::Error> {
+        task::block_on(self.0.basic_qos(1, BasicQosOptions::default()))?;
+        self.0.consume(worker, config)
+    }
+}
+
 struct ChannelNotificationReceiver<'a> {
     channel: &'a mut CloseOnDrop<lapin::Channel>,
     deliver: &'a Delivery,