add a typed wrapper to amqp to avoid false, false, false, false

This commit is contained in:
Graham Christensen 2018-01-30 17:25:56 -05:00
parent 60a8081daa
commit cb18c0d5b1
No known key found for this signature in database
GPG key ID: ACA1C1D120C83D5C
4 changed files with 90 additions and 32 deletions

View file

@ -9,13 +9,12 @@ use std::env;
use std::path::Path; use std::path::Path;
use amqp::Basic; use amqp::Basic;
use amqp::Session;
use amqp::Table;
use ofborg::config; use ofborg::config;
use ofborg::checkout; use ofborg::checkout;
use ofborg::notifyworker; use ofborg::notifyworker;
use ofborg::tasks; use ofborg::tasks;
use ofborg::easyamqp;
use ofborg::easyamqp::TypedWrappers;
fn main() { fn main() {
@ -23,36 +22,29 @@ fn main() {
ofborg::setup_log(); ofborg::setup_log();
println!("Hello, world!");
let mut session = Session::open_url(&cfg.rabbitmq.as_uri()).unwrap();
println!("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 full_logs: bool = match &cfg.feedback {
let full_logs: bool;
match &cfg.feedback {
&Some(ref feedback) => { &Some(ref feedback) => {
full_logs = feedback.full_logs; feedback.full_logs
} }
&None => { &None => {
warn!("Please define feedback.full_logs in your configuration to true or false!"); warn!("Please define feedback.full_logs in your configuration to true or false!");
warn!("feedback.full_logs when true will cause the full build log to be sent back"); warn!("feedback.full_logs when true will cause the full build log to be sent back");
warn!("to the server, and be viewable by everyone."); warn!("to the server, and be viewable by everyone.");
warn!("I strongly encourage everybody turn this on!"); warn!("I strongly encourage everybody turn this on!");
full_logs = false; false
} }
} };
let mut session = easyamqp::session_from_config(&cfg.rabbitmq).unwrap();
let mut channel = session.open_channel(1).unwrap();
channel.basic_prefetch(1).unwrap(); channel.basic_prefetch(1).unwrap();
channel channel
.basic_consume( .consume(
notifyworker::new(tasks::build::BuildWorker::new( notifyworker::new(tasks::build::BuildWorker::new(
cloner, cloner,
nix, nix,
@ -60,20 +52,19 @@ fn main() {
cfg.runner.identity.clone(), cfg.runner.identity.clone(),
full_logs, full_logs,
)), )),
format!("build-inputs-{}", cfg.nix.system.clone()).as_ref(), easyamqp::ConsumeConfig {
format!("{}-builder", cfg.whoami()).as_ref(), queue: format!("build-inputs-{}", cfg.nix.system.clone()),
false, consumer_tag: format!("{}-builder", cfg.whoami()),
false, no_local: false,
false, no_ack: false,
false, exclusive: false,
Table::new(), nowait: false,
arguments: None
},
) )
.unwrap(); .unwrap();
channel.start_consuming(); channel.start_consuming();
println!("Finished consuming?");
channel.close(200, "Bye").unwrap(); channel.close(200, "Bye").unwrap();
println!("Closed the channel"); println!("Closed the channel");
session.close(200, "Good Bye"); session.close(200, "Good Bye");

View file

@ -27,7 +27,7 @@ pub struct FeedbackConfig {
pub full_logs: bool, pub full_logs: bool,
} }
#[derive(Serialize, Deserialize, Debug)] #[derive(Serialize, Deserialize, Debug, Clone)]
pub struct RabbitMQConfig { pub struct RabbitMQConfig {
pub ssl: bool, pub ssl: bool,
pub host: String, pub host: String,
@ -107,7 +107,6 @@ impl Config {
} }
} }
impl RabbitMQConfig { impl RabbitMQConfig {
pub fn as_uri(&self) -> String { pub fn as_uri(&self) -> String {
return format!( return format!(

66
ofborg/src/easyamqp.rs Normal file
View file

@ -0,0 +1,66 @@
use ofborg;
use ofborg::config::RabbitMQConfig;
use amqp;
use amqp::Basic;
pub struct ConsumeConfig {
pub queue: String,
pub consumer_tag: String,
pub no_local: bool,
pub no_ack: bool,
pub exclusive: bool,
pub nowait: bool,
pub arguments: Option<amqp::Table>,
}
pub fn session_from_config(config: &RabbitMQConfig)
-> Result<amqp::Session, amqp::AMQPError> {
let scheme = if config.ssl {
amqp::AMQPScheme::AMQPS
} else {
amqp::AMQPScheme::AMQP
};
let mut properties = amqp::Table::new();
// properties.insert("identity".to_owned(), amqp::TableEntry::LongString(identity.to_owned()));
properties.insert(
"ofborg_version".to_owned(),
amqp::TableEntry::LongString(ofborg::VERSION.to_owned())
);
amqp::Session::new(
amqp::Options{
host: config.host.clone(),
login: config.username.clone(),
password: config.password.clone(),
scheme: scheme,
properties: properties,
.. amqp::Options::default()
}
)
}
pub trait TypedWrappers {
fn consume<T>(&mut self, callback: T, config: ConsumeConfig)
-> Result<String, amqp::AMQPError>
where T: amqp::Consumer + 'static;
}
impl TypedWrappers for amqp::Channel {
fn consume<T>(&mut self, callback: T, config: ConsumeConfig)
-> Result<String, amqp::AMQPError>
where T: amqp::Consumer + 'static
{
self.basic_consume(
callback,
config.queue,
config.consumer_tag,
config.no_local,
config.no_ack,
config.exclusive,
config.nowait,
config.arguments.unwrap_or(amqp::Table::new()),
)
}
}

View file

@ -40,6 +40,7 @@ pub mod asynccmd;
pub mod notifyworker; pub mod notifyworker;
pub mod writetoline; pub mod writetoline;
pub mod test_scratch; pub mod test_scratch;
pub mod easyamqp;
pub mod ofborg { pub mod ofborg {
pub use asynccmd; pub use asynccmd;
@ -62,8 +63,9 @@ pub mod ofborg {
pub use tagger; pub use tagger;
pub use writetoline; pub use writetoline;
pub use test_scratch; pub use test_scratch;
pub use easyamqp;
pub const VERSION: &'static str = env!("CARGO_PKG_VERSION");
} }
pub fn setup_log() { pub fn setup_log() {