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

convert evaluation filter to easylapin

Browse source
This commit is contained in:
Daiderd Jordan 2020-05-21 20:33:33 +02:00
parent f1d34ee242
commit bd8bed6ad3
No known key found for this signature in database
GPG key ID: D02435D05B810C96
1 changed files with 62 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

79
ofborg/src/bin/evaluation-filter.rs
View file

@ -1,26 +1,26 @@
use amqp::Basic;
use ofborg::config;
use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
use ofborg::tasks;
use ofborg::worker;
use std::env; use std::env;
use std::error::Error;
use async_std::task;
use tracing::info; 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(); ofborg::setup_log();
info!("Hello, world!"); let arg = env::args()
.nth(1)
.expect("usage: evaluation-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: "github-events".to_owned(),
exchange_type: easyamqp::ExchangeType::Topic, exchange_type: easyamqp::ExchangeType::Topic,
passive: false, passive: false,
@ -28,63 +28,50 @@ fn main() {
auto_delete: false, auto_delete: false,
no_wait: false, no_wait: false,
internal: false, internal: false,
}) })?;
.unwrap();
channel chan.declare_queue(easyamqp::QueueConfig {
.declare_queue(easyamqp::QueueConfig {
queue: "mass-rebuild-check-jobs".to_owned(), queue: "mass-rebuild-check-jobs".to_owned(),
passive: false, passive: false,
durable: true, durable: true,
exclusive: false, exclusive: false,
auto_delete: false, auto_delete: false,
no_wait: false, no_wait: false,
}) })?;
.unwrap();
channel let queue_name = String::from("mass-rebuild-check-jobs");
.declare_queue(easyamqp::QueueConfig { chan.declare_queue(easyamqp::QueueConfig {
queue: "mass-rebuild-check-inputs".to_owned(), queue: queue_name.clone(),
passive: false, passive: false,
durable: true, durable: true,
exclusive: false, exclusive: false,
auto_delete: false, auto_delete: false,
no_wait: false, no_wait: false,
}) })?;
.unwrap();
channel chan.bind_queue(easyamqp::BindQueueConfig {
.bind_queue(easyamqp::BindQueueConfig { queue: queue_name.clone(),
queue: "mass-rebuild-check-inputs".to_owned(),
exchange: "github-events".to_owned(), exchange: "github-events".to_owned(),
routing_key: Some("pull_request.nixos/nixpkgs".to_owned()), routing_key: Some("pull_request.nixos/nixpkgs".to_owned()),
no_wait: false, no_wait: false,
}) })?;
.unwrap();
channel.basic_prefetch(1).unwrap(); let handle = chan.consume(
let mut channel = channel tasks::evaluationfilter::EvaluationFilterWorker::new(cfg.acl()),
.consume(
worker::new(tasks::evaluationfilter::EvaluationFilterWorker::new(
cfg.acl(),
)),
easyamqp::ConsumeConfig { easyamqp::ConsumeConfig {
queue: "mass-rebuild-check-inputs".to_owned(), queue: queue_name.clone(),
consumer_tag: format!("{}-evaluation-filter", cfg.whoami()), consumer_tag: format!("{}-evaluation-filter", cfg.whoami()),
no_local: false, no_local: false,
no_ack: false, no_ack: false,
no_wait: false, no_wait: false,
exclusive: 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"); info!("Closed the session... EOF");
Ok(())
} }