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

move prefetch qos to WorkerChannel

Browse source 
This way the log collecter can use a channel with prefetching while the
other workers still consume items one by one.
This commit is contained in:
Daiderd Jordan 2020-05-22 21:55:18 +02:00
parent 51b4c3dbc9
commit f652901df2
No known key found for this signature in database
GPG key ID: D02435D05B810C96
6 changed files with 19 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -56,7 +56,7 @@ fn main() -> Result<(), Box<dyn Error>> {
no_wait: false,
})?;
let handle = chan.consume(
let handle = easylapin::WorkerChannel(chan).consume(
tasks::evaluationfilter::EvaluationFilterWorker::new(cfg.acl()),
easyamqp::ConsumeConfig {
queue: queue_name.clone(),

2
ofborg/src/bin/github-comment-filter.rs
View file

@ -57,7 +57,7 @@ fn main() -> Result<(), Box<dyn Error>> {
no_wait: false,
})?;
let handle = chan.consume(
let handle = easylapin::WorkerChannel(chan).consume(
tasks::githubcommentfilter::GitHubCommentWorker::new(cfg.acl(), cfg.github()),
easyamqp::ConsumeConfig {
queue: "build-inputs".to_owned(),

2
ofborg/src/bin/github-comment-poster.rs
View file

@ -46,7 +46,7 @@ fn main() -> Result<(), Box<dyn Error>> {
no_wait: false,
})?;
let handle = chan.consume(
let handle = easylapin::WorkerChannel(chan).consume(
tasks::githubcommentposter::GitHubCommentPoster::new(cfg.github_app_vendingmachine()),
easyamqp::ConsumeConfig {
queue: "build-results".to_owned(),

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

@ -48,6 +48,7 @@ fn main() -> Result<(), Box<dyn Error>> {
no_wait: false,
})?;
// Regular channel, we want prefetching here.
let handle = chan.consume(
tasks::log_message_collector::LogMessageCollector::new(
PathBuf::from(cfg.log_storage.clone().unwrap().path),

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

@ -53,7 +53,7 @@ fn main() -> Result<(), Box<dyn Error>> {
no_wait: false,
})?;
let handle = chan.consume(
let handle = easylapin::WorkerChannel(chan).consume(
tasks::evaluate::EvaluationWorker::new(
cloner,
&nix,

16
ofborg/src/easylapin.rs
View file

@ -86,8 +86,6 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop<Channel> {
type Handle = Pin<Box<dyn Future<Output = ()> + 'a>>;
fn consume(self, mut worker: W, config: ConsumeConfig) -> Result<Self::Handle, Self::Error> {
task::block_on(self.basic_qos(1, BasicQosOptions::default()))?;
let mut consumer = task::block_on(self.basic_consume(
&config.queue,
&config.consumer_tag,
@ -117,6 +115,20 @@ impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for CloseOnDrop<Channel> {
}
}
// Same as a regular channel, but without prefetching,
// used for services with multiple instances.
pub struct WorkerChannel(pub CloseOnDrop<Channel>);
impl<'a, W: SimpleWorker + 'a> ConsumerExt<'a, W> for WorkerChannel {
type Error = lapin::Error;
type Handle = Pin<Box<dyn Future<Output = ()> + 'a>>;
fn consume(self, worker: W, config: ConsumeConfig) -> Result<Self::Handle, Self::Error> {
task::block_on(self.0.basic_qos(1, BasicQosOptions::default()))?;
self.0.consume(worker, config)
}
}
struct ChannelNotificationReceiver<'a> {
channel: &'a mut CloseOnDrop<lapin::Channel>,
deliver: &'a Delivery,