the-distro/ofborg
2
0
Fork 1
Code Issues 1 Pull requests 1 Packages Projects Releases Wiki Activity

convert mass rebuilder to lapin

Browse source
This commit is contained in:
Daiderd Jordan 2020-05-21 21:09:33 +02:00
parent 3ce9f3edd1
commit 9120d16bd7
No known key found for this signature in database
GPG key ID: D02435D05B810C96
1 changed files with 51 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
ofborg/src/bin/mass-rebuilder.rs
View file

@ -1,20 +1,26 @@
use ofborg::checkout;
use ofborg::config;
use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
use ofborg::stats;
use ofborg::tasks;
use ofborg::worker;
use std::env; use std::env;
use std::error::Error;
use std::path::Path; use std::path::Path;
use std::process; use std::process;
use amqp::Basic; use async_std::task;
use tracing::{error, info}; use tracing::{error, info};
use ofborg::checkout;
use ofborg::config;
use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
use ofborg::easylapin;
use ofborg::stats;
use ofborg::tasks;
// FIXME: remove with rust/cargo update // FIXME: remove with rust/cargo update
#[allow(clippy::cognitive_complexity)] #[allow(clippy::cognitive_complexity)]
fn main() { fn main() -> Result<(), Box<dyn Error>> {
ofborg::setup_log();
let arg = env::args().nth(1).expect("usage: mass-rebuilder <config>");
let cfg = config::load(arg.as_ref());
let memory_info = sys_info::mem_info().expect("Unable to get memory information from OS"); let memory_info = sys_info::mem_info().expect("Unable to get memory information from OS");
if memory_info.avail < 8 * 1024 * 1024 { if memory_info.avail < 8 * 1024 * 1024 {
@ -26,68 +32,52 @@ fn main() {
process::exit(1); process::exit(1);
}; };
let cfg = config::load(env::args().nth(1).unwrap().as_ref()); let conn = easylapin::from_config(&cfg.rabbitmq)?;
let mut chan = task::block_on(conn.create_channel())?;
ofborg::setup_log();
info!("Hello, world!");
let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
info!("Connected to rabbitmq");
let mut channel = session.open_channel(1).unwrap();
let cloner = checkout::cached_cloner(Path::new(&cfg.checkout.root)); let cloner = checkout::cached_cloner(Path::new(&cfg.checkout.root));
let nix = cfg.nix(); let nix = cfg.nix();
let events = stats::RabbitMQ::from_amqp( let events = stats::RabbitMQ::from_lapin(
&format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()), &format!("{}-{}", cfg.runner.identity.clone(), cfg.nix.system.clone()),
session.open_channel(3).unwrap(), task::block_on(conn.create_channel())?,
); );
let mrw = tasks::evaluate::EvaluationWorker::new( let queue_name = String::from("mass-rebuild-check-jobs");
cloner, chan.declare_queue(easyamqp::QueueConfig {
&nix, queue: queue_name.clone(),
cfg.github(), passive: false,
cfg.github_app_vendingmachine(), durable: true,
cfg.acl(), exclusive: false,
cfg.runner.identity.clone(), auto_delete: false,
events, no_wait: false,
cfg.tag_paths.clone().unwrap(), })?;
);
channel let handle = chan.consume(
.declare_queue(easyamqp::QueueConfig { tasks::evaluate::EvaluationWorker::new(
queue: "mass-rebuild-check-jobs".to_owned(), cloner,
passive: false, &nix,
durable: true, cfg.github(),
exclusive: false, cfg.github_app_vendingmachine(),
auto_delete: false, cfg.acl(),
cfg.runner.identity.clone(),
events,
cfg.tag_paths.clone().unwrap(),
),
easyamqp::ConsumeConfig {
queue: queue_name.clone(),
consumer_tag: format!("{}-mass-rebuild-checker", cfg.whoami()),
no_local: false,
no_ack: false,
no_wait: false, no_wait: false,
}) exclusive: false,
.unwrap(); },
)?;
channel.basic_prefetch(1).unwrap(); info!("Fetching jobs from {}", queue_name);
let mut channel = channel task::block_on(handle);
.consume(
worker::new(mrw),
easyamqp::ConsumeConfig {
queue: "mass-rebuild-check-jobs".to_owned(),
consumer_tag: format!("{}-mass-rebuild-checker", cfg.whoami()),
no_local: false,
no_ack: false,
no_wait: false,
exclusive: false,
},
)
.unwrap();
channel.start_consuming(); drop(conn); // Close connection.
info!("Finished consuming?");
channel.close(200, "Bye").unwrap();
info!("Closed the channel");
session.close(200, "Good Bye");
info!("Closed the session... EOF"); info!("Closed the session... EOF");
Ok(())
} }