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

Add publish support, and nack is now nackdump / nackrequeue

Browse source
This commit is contained in:
Graham Christensen 2017-11-09 13:24:24 -05:00
parent 3d0f728d92
commit 26288b459e
No known key found for this signature in database
GPG key ID: ACA1C1D120C83D5C
2 changed files with 26 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
ofborg/src/message/buildjob.rs
View file

@ -37,14 +37,14 @@ impl Actions {
return vec![
worker::Action::Publish(worker::QueueMsg{
exchange: Some("build-results".to_owned()),
exchange: Some("build-results-x".to_owned()),
routing_key: None,
mandatory: true,
immediate: true,
properties: Some(props),
content: serde_json::to_string(&msg).unwrap().into_bytes()
}),
worker::Action::Ack
worker::Action::NackRequeue
];
}
}

30
ofborg/src/worker.rs
View file

@ -1,4 +1,4 @@
use amqp::Basic;
use amqp::{Consumer, Channel};
use amqp::protocol::basic::{Deliver,BasicProperties};
use std::marker::Send;
@ -15,7 +15,8 @@ pub type Actions = Vec<Action>;
pub enum Action {
Ack,
Nack,
NackRequeue,
NackDump,
Publish(QueueMsg),
}
@ -48,7 +49,7 @@ pub fn new<T: SimpleWorker>(worker: T) -> Worker<T> {
impl <T: SimpleWorker + Send> Consumer for Worker<T> {
fn handle_delivery(&mut self,
_: &mut Channel,
channel: &mut Channel,
method: Deliver,
headers: BasicProperties,
body: Vec<u8>) {
@ -56,9 +57,26 @@ impl <T: SimpleWorker + Send> Consumer for Worker<T> {
let job = self.internal.msg_to_job(&method, &headers, &body).unwrap();
for action in self.internal.consumer(&job) {
match action {
Action::Ack => {}
Action::Nack => {}
Action::Publish(_) => {}
Action::Ack => {
channel.basic_ack(method.delivery_tag, false).unwrap();
}
Action::NackRequeue => {
channel.basic_nack(method.delivery_tag, false, true).unwrap();
}
Action::NackDump => {
channel.basic_nack(method.delivery_tag, false, false).unwrap();
}
Action::Publish(msg) => {
let props = msg.properties.unwrap_or(BasicProperties{ ..Default::default()});
channel.basic_publish(
msg.exchange.unwrap_or("".to_owned()),
msg.routing_key.unwrap_or("".to_owned()),
msg.mandatory,
msg.immediate,
props,
msg.content
).unwrap();
}
}
}
}