danderson/ofborg
0
0
Fork 0
forked from the-distro/ofborg
Code Pull requests Activity

remove amqp return types

Browse source
This commit is contained in:
Daiderd Jordan 2020-04-27 22:21:25 +02:00
parent a34a4accaf
commit 98723462a0
No known key found for this signature in database
GPG key ID: D02435D05B810C96
3 changed files with 29 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
ofborg/src/bin/builder.rs
View file

@ -42,32 +42,34 @@ fn main() {
}) })
.unwrap(); .unwrap();
let queue_name: String = if cfg.runner.build_all_jobs != Some(true) { let queue_name = if cfg.runner.build_all_jobs != Some(true) {
let queue_name = format!("build-inputs-{}", cfg.nix.system.clone());
channel channel
.declare_queue(easyamqp::QueueConfig { .declare_queue(easyamqp::QueueConfig {
queue: format!("build-inputs-{}", cfg.nix.system.clone()), 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() .unwrap();
.queue queue_name
} else { } else {
warn!("Building all jobs, please don't use this unless you're"); warn!("Building all jobs, please don't use this unless you're");
warn!("developing and have Graham's permission!"); warn!("developing and have Graham's permission!");
let queue_name = "".to_owned();
channel channel
.declare_queue(easyamqp::QueueConfig { .declare_queue(easyamqp::QueueConfig {
queue: "".to_owned(), queue: queue_name.clone(),
passive: false, passive: false,
durable: false, durable: false,
exclusive: true, exclusive: true,
auto_delete: true, auto_delete: true,
no_wait: false, no_wait: false,
}) })
.unwrap() .unwrap();
.queue queue_name
}; };
channel channel

8
ofborg/src/bin/log-message-collector.rs
View file

@ -28,17 +28,17 @@ fn main() {
}) })
.unwrap(); .unwrap();
let queue_name = channel let queue_name = "".to_owned();
channel
.declare_queue(easyamqp::QueueConfig { .declare_queue(easyamqp::QueueConfig {
queue: "".to_owned(), queue: queue_name.clone(),
passive: false, passive: false,
durable: false, durable: false,
exclusive: true, exclusive: true,
auto_delete: true, auto_delete: true,
no_wait: false, no_wait: false,
}) })
.unwrap() .unwrap();
.queue;
channel channel
.bind_queue(easyamqp::BindQueueConfig { .bind_queue(easyamqp::BindQueueConfig {

28
ofborg/src/easyamqp.rs
View file

@ -299,30 +299,30 @@ pub fn session_from_config(config: &RabbitMQConfig) -> Result<amqp::Session, amq
pub trait TypedWrappers { pub trait TypedWrappers {
type Error; type Error;
fn consume<T>(&mut self, callback: T, config: ConsumeConfig) -> Result<String, Self::Error> fn consume<T>(&mut self, callback: T, config: ConsumeConfig) -> Result<(), Self::Error>
where where
T: amqp::Consumer + 'static; T: amqp::Consumer + 'static;
fn declare_exchange( fn declare_exchange(
&mut self, &mut self,
config: ExchangeConfig, config: ExchangeConfig,
) -> Result<amqp::protocol::exchange::DeclareOk, Self::Error>; ) -> Result<(), Self::Error>;
fn declare_queue( fn declare_queue(
&mut self, &mut self,
config: QueueConfig, config: QueueConfig,
) -> Result<amqp::protocol::queue::DeclareOk, Self::Error>; ) -> Result<(), Self::Error>;
fn bind_queue( fn bind_queue(
&mut self, &mut self,
config: BindQueueConfig, config: BindQueueConfig,
) -> Result<amqp::protocol::queue::BindOk, Self::Error>; ) -> Result<(), Self::Error>;
} }
impl TypedWrappers for amqp::Channel { impl TypedWrappers for amqp::Channel {
type Error = amqp::AMQPError; type Error = amqp::AMQPError;
fn consume<T>(&mut self, callback: T, config: ConsumeConfig) -> Result<String, Self::Error> fn consume<T>(&mut self, callback: T, config: ConsumeConfig) -> Result<(), Self::Error>
where where
T: amqp::Consumer + 'static, T: amqp::Consumer + 'static,
{ {
@ -335,13 +335,14 @@ impl TypedWrappers for amqp::Channel {
config.exclusive, config.exclusive,
config.no_wait, config.no_wait,
amqp::Table::new(), amqp::Table::new(),
) )?;
Ok(())
} }
fn declare_exchange( fn declare_exchange(
&mut self, &mut self,
config: ExchangeConfig, config: ExchangeConfig,
) -> Result<amqp::protocol::exchange::DeclareOk, Self::Error> { ) -> Result<(), Self::Error> {
self.exchange_declare( self.exchange_declare(
config.exchange, config.exchange,
config.exchange_type.into(), config.exchange_type.into(),
@ -351,13 +352,14 @@ impl TypedWrappers for amqp::Channel {
config.internal, config.internal,
config.no_wait, config.no_wait,
amqp::Table::new(), amqp::Table::new(),
) )?;
Ok(())
} }
fn declare_queue( fn declare_queue(
&mut self, &mut self,
config: QueueConfig, config: QueueConfig,
) -> Result<amqp::protocol::queue::DeclareOk, Self::Error> { ) -> Result<(), Self::Error> {
self.queue_declare( self.queue_declare(
config.queue, config.queue,
config.passive, config.passive,
@ -366,19 +368,21 @@ impl TypedWrappers for amqp::Channel {
config.auto_delete, config.auto_delete,
config.no_wait, config.no_wait,
amqp::Table::new(), amqp::Table::new(),
) )?;
Ok(())
} }
fn bind_queue( fn bind_queue(
&mut self, &mut self,
config: BindQueueConfig, config: BindQueueConfig,
) -> Result<amqp::protocol::queue::BindOk, Self::Error> { ) -> Result<(), Self::Error> {
self.queue_bind( self.queue_bind(
config.queue, config.queue,
config.exchange, config.exchange,
config.routing_key.unwrap_or_else(|| "".to_owned()), config.routing_key.unwrap_or_else(|| "".to_owned()),
config.no_wait, config.no_wait,
amqp::Table::new(), amqp::Table::new(),
) )?;
Ok(())
} }
} }