NixOS · May 24, 2020 · May 24, 2020 · May 24, 2020
Showing with 56 additions and 32 deletions.

+24 −18 ofborg/src/bin/build-faker.rs

+25 −13 ofborg/src/bin/log-message-generator.rs

+7 −1 ofborg/src/easylapin.rs
diff --git a/ofborg/src/bin/build-faker.rs b/ofborg/src/bin/build-faker.rs
@@ -1,24 +1,25 @@
+use std::env;
+use std::error::Error;
+
+use async_std::task;
+use lapin::message::Delivery;
+use lapin::BasicProperties;
+
 use ofborg::commentparser;
 use ofborg::config;
-use ofborg::easyamqp;
+use ofborg::easylapin;
 use ofborg::message::{buildjob, Pr, Repo};
-use ofborg::notifyworker::{self, NotificationReceiver};
+use ofborg::notifyworker::NotificationReceiver;
 use ofborg::worker;
 
-use std::env;
-
-use tracing::info;
-
-fn main() {
-    let cfg = config::load(env::args().nth(1).unwrap().as_ref());
+fn main() -> Result<(), Box<dyn Error>> {
     ofborg::setup_log();
 
-    info!("Hello, world!");
-
-    let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
-    info!("Connected to rabbitmq");
+    let arg = env::args().nth(1).expect("usage: build-faker <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())?;
 
     let repo_msg = Repo {
         clone_url: "https://github.com/nixos/ofborg.git".to_owned(),
@@ -46,7 +47,15 @@ fn main() {
     };
 
     {
-        let mut recv = notifyworker::ChannelNotificationReceiver::new(&mut channel, 0);
+        let deliver = Delivery {
+            delivery_tag: 0,
+            exchange: "no-exchange".into(),
+            routing_key: "".into(),
+            redelivered: false,
+            properties: BasicProperties::default(),
+            data: vec![],
+        };
+        let mut recv = easylapin::ChannelNotificationReceiver::new(&mut chan, &deliver);
 
         for _i in 1..2 {
             recv.tell(worker::publish_serde_action(
@@ -57,8 +66,5 @@ fn main() {
         }
     }
 
-    channel.close(200, "Bye").unwrap();
-    info!("Closed the channel");
-    session.close(200, "Good Bye");
-    info!("Closed the session... EOF");
+    Ok(())
 }
diff --git a/ofborg/src/bin/log-message-generator.rs b/ofborg/src/bin/log-message-generator.rs
@@ -1,26 +1,38 @@
-use ofborg::config;
-use ofborg::easyamqp;
-use ofborg::message::{buildjob, Pr, Repo};
-use ofborg::notifyworker;
-use ofborg::tasks::build;
-
 use std::env;
+use std::error::Error;
 use std::thread;
 use std::time::Duration;
 
+use async_std::task;
+use lapin::message::Delivery;
+use lapin::BasicProperties;
 use tracing::info;
 
-fn main() {
-    let cfg = config::load(env::args().nth(1).unwrap().as_ref());
+use ofborg::config;
+use ofborg::easylapin;
+use ofborg::message::{buildjob, Pr, Repo};
+use ofborg::tasks::build;
+
+fn main() -> Result<(), Box<dyn Error>> {
     ofborg::setup_log();
 
-    let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
-    info!("Connected to rabbitmq");
+    let arg = env::args()
+        .nth(1)
+        .expect("usage: log-message-generator <config>");
+    let cfg = config::load(arg.as_ref());
 
-    info!("About to open channel #1");
-    let mut chan = session.open_channel(1).unwrap();
+    let conn = easylapin::from_config(&cfg.rabbitmq)?;
+    let mut chan = task::block_on(conn.create_channel())?;
 
-    let mut receiver = notifyworker::ChannelNotificationReceiver::new(&mut chan, 0);
+    let deliver = Delivery {
+        delivery_tag: 0,
+        exchange: "no-exchange".into(),
+        routing_key: "".into(),
+        redelivered: false,
+        properties: BasicProperties::default(),
+        data: vec![],
+    };
+    let mut receiver = easylapin::ChannelNotificationReceiver::new(&mut chan, &deliver);
     let job = buildjob::BuildJob {
         attrs: vec![],
         pr: Pr {

diff --git a/ofborg/src/easylapin.rs b/ofborg/src/easylapin.rs
@@ -129,11 +129,17 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for WorkerChannel {
     }
 }
 
-struct ChannelNotificationReceiver<'a> {
+pub struct ChannelNotificationReceiver<'a> {
     channel: &'a mut CloseOnDrop<lapin::Channel>,
     deliver: &'a Delivery,
 }
 
+impl<'a> ChannelNotificationReceiver<'a> {
+    pub fn new(channel: &'a mut CloseOnDrop<lapin::Channel>, deliver: &'a Delivery) -> Self {
+        ChannelNotificationReceiver { channel, deliver }
+    }
+}
+
 impl<'a> NotificationReceiver for ChannelNotificationReceiver<'a> {
     fn tell(&mut self, action: Action) {
         task::block_on(action_deliver(self.channel, self.deliver, action))