Merge pull request #484 from LnL7/lapin-comment-filter

lapin github-comment-filter

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

@@ -1,25 +1,26 @@
-use ofborg::config;
-use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
-use ofborg::tasks;
-use ofborg::worker;
-
 use std::env;
+use std::error::Error;
 
-use amqp::Basic;
+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();
 
-    info!("Hello, world!");
+    let arg = env::args()
+        .nth(1)
+        .expect("usage: github-comment-filter <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 mut channel = session.open_channel(1).unwrap();
+    chan.declare_exchange(easyamqp::ExchangeConfig {
-    channel
-        .declare_exchange(easyamqp::ExchangeConfig {
         exchange: "github-events".to_owned(),
         exchange_type: easyamqp::ExchangeType::Topic,
         passive: false,
@@ -27,11 +28,9 @@ fn main() {
         auto_delete: false,
         no_wait: false,
         internal: false,
-        })
+    })?;
-        .unwrap();
 
-    channel
+    chan.declare_exchange(easyamqp::ExchangeConfig {
-        .declare_exchange(easyamqp::ExchangeConfig {
         exchange: "build-jobs".to_owned(),
         exchange_type: easyamqp::ExchangeType::Fanout,
         passive: false,
@@ -39,36 +38,27 @@ fn main() {
         auto_delete: false,
         no_wait: false,
         internal: false,
-        })
+    })?;
-        .unwrap();
 
-    channel
+    let queue_name = "build-inputs";
-        .declare_queue(easyamqp::QueueConfig {
+    chan.declare_queue(easyamqp::QueueConfig {
-            queue: "build-inputs".to_owned(),
+        queue: queue_name.to_owned(),
         passive: false,
         durable: true,
         exclusive: false,
         auto_delete: false,
         no_wait: false,
-        })
+    })?;
-        .unwrap();
 
-    channel
+    chan.bind_queue(easyamqp::BindQueueConfig {
-        .bind_queue(easyamqp::BindQueueConfig {
         queue: "build-inputs".to_owned(),
         exchange: "github-events".to_owned(),
         routing_key: Some("issue_comment.*".to_owned()),
         no_wait: false,
-        })
+    })?;
-        .unwrap();
 
-    channel.basic_prefetch(1).unwrap();
+    let handle = chan.consume(
-    let mut channel = channel
+        tasks::githubcommentfilter::GitHubCommentWorker::new(cfg.acl(), cfg.github()),
-        .consume(
-            worker::new(tasks::githubcommentfilter::GitHubCommentWorker::new(
-                cfg.acl(),
-                cfg.github(),
-            )),
         easyamqp::ConsumeConfig {
             queue: "build-inputs".to_owned(),
             consumer_tag: format!("{}-github-comment-filter", cfg.whoami()),
@@ -77,15 +67,12 @@ fn main() {
             no_wait: false,
             exclusive: false,
         },
-        )
+    )?;
-        .unwrap();
 
-    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(())
 }

ofborg/src/tasks/githubcommentfilter.rs

@@ -4,7 +4,7 @@ use crate::ghevent;
 use crate::message::{buildjob, evaluationjob, Pr, Repo};
 use crate::worker;
 
-use tracing::{error, info};
+use tracing::{debug_span, error, info};
 use uuid::Uuid;
 
 pub struct GitHubCommentWorker {
@@ -34,7 +34,12 @@ impl worker::SimpleWorker for GitHubCommentWorker {
         }
     }
 
+    // FIXME: remove with rust/cargo update
+    #[allow(clippy::cognitive_complexity)]
     fn consumer(&mut self, job: &ghevent::IssueComment) -> worker::Actions {
+        let span = debug_span!("job", pr = ?job.issue.number);
+        let _enter = span.enter();
+
         if job.action == ghevent::IssueCommentAction::Deleted {
             return vec![worker::Action::Ack];
         }