generalize SimpleWorker trait

ofborg/src/bin/builder.rs

@@ -81,7 +81,7 @@ fn main() {
         })
         .unwrap();
 
-    channel
+    let mut channel = channel
         .consume(
             notifyworker::new(tasks::build::BuildWorker::new(
                 cloner,

ofborg/src/bin/evaluation-filter.rs

@@ -62,7 +62,7 @@ fn main() {
         .unwrap();
 
     channel.basic_prefetch(1).unwrap();
-    channel
+    let mut channel = channel
         .consume(
             worker::new(tasks::evaluationfilter::EvaluationFilterWorker::new(
                 cfg.acl(),

ofborg/src/bin/github-comment-filter.rs

@@ -62,7 +62,7 @@ fn main() {
         .unwrap();
 
     channel.basic_prefetch(1).unwrap();
-    channel
+    let mut channel = channel
         .consume(
             worker::new(tasks::githubcommentfilter::GitHubCommentWorker::new(
                 cfg.acl(),

ofborg/src/bin/github-comment-poster.rs

@@ -47,7 +47,7 @@ fn main() {
         .unwrap();
 
     channel.basic_prefetch(1).unwrap();
-    channel
+    let mut channel = channel
         .consume(
             worker::new(tasks::githubcommentposter::GitHubCommentPoster::new(
                 cfg.github_app_vendingmachine(),

ofborg/src/bin/log-message-collector.rs

@@ -49,7 +49,7 @@ fn main() {
         })
         .unwrap();
 
-    channel
+    let mut channel = channel
         .consume(
             worker::new(tasks::log_message_collector::LogMessageCollector::new(
                 PathBuf::from(cfg.log_storage.clone().unwrap().path),

ofborg/src/bin/mass-rebuilder.rs

@@ -65,7 +65,7 @@ fn main() {
         .unwrap();
 
     channel.basic_prefetch(1).unwrap();
-    channel
+    let mut channel = channel
         .consume(
             worker::new(mrw),
             easyamqp::ConsumeConfig {

ofborg/src/bin/stats.rs

@@ -59,7 +59,7 @@ fn main() {
         .unwrap();
 
     channel.basic_prefetch(1).unwrap();
-    channel
+    let mut channel = channel
         .consume(
             worker::new(collector),
             easyamqp::ConsumeConfig {

ofborg/src/easyamqp.rs

@@ -303,9 +303,10 @@ pub trait ChannelExt {
     fn bind_queue(&mut self, config: BindQueueConfig) -> Result<(), Self::Error>;
 }
 
-pub trait ConsumerExt<T> {
+pub trait ConsumerExt<C> {
     type Error;
-    fn consume(&mut self, callback: T, config: ConsumeConfig) -> Result<(), Self::Error>;
+    type Handle;
+    fn consume(self, callback: C, config: ConsumeConfig) -> Result<Self::Handle, Self::Error>;
 }
 
 impl ChannelExt for amqp::Channel {
@@ -350,10 +351,11 @@ impl ChannelExt for amqp::Channel {
     }
 }
 
-impl<T: amqp::Consumer + 'static> ConsumerExt<T> for amqp::Channel {
+impl<C: amqp::Consumer + 'static> ConsumerExt<C> for amqp::Channel {
     type Error = amqp::AMQPError;
+    type Handle = Self;
 
-    fn consume(&mut self, callback: T, config: ConsumeConfig) -> Result<(), Self::Error> {
+    fn consume(mut self, callback: C, config: ConsumeConfig) -> Result<Self::Handle, Self::Error> {
         self.basic_consume(
             callback,
             config.queue,
@@ -364,6 +366,6 @@ impl<T: amqp::Consumer + 'static> ConsumerExt<T> for amqp::Channel {
             config.no_wait,
             amqp::Table::new(),
         )?;
-        Ok(())
+        Ok(self)
     }
 }

ofborg/src/tasks/evaluate.rs

@@ -11,7 +11,6 @@ use crate::systems;
 use crate::tasks::eval;
 use crate::worker;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
 use hubcaps::checks::CheckRunOptions;
 use hubcaps::gists::Gists;
 use hubcaps::issues::Issue;
@@ -60,12 +59,7 @@ impl<E: stats::SysEvents> EvaluationWorker<E> {
 impl<E: stats::SysEvents + 'static> worker::SimpleWorker for EvaluationWorker<E> {
     type J = evaluationjob::EvaluationJob;
 
-    fn msg_to_job(
+    fn msg_to_job(&mut self, _: &str, _: &Option<String>, body: &[u8]) -> Result<Self::J, String> {
-        &mut self,
-        _: &Deliver,
-        _: &BasicProperties,
-        body: &[u8],
-    ) -> Result<Self::J, String> {
         self.events.notify(Event::JobReceived);
         match evaluationjob::from(body) {
             Ok(e) => {

ofborg/src/tasks/evaluationfilter.rs

@@ -3,8 +3,6 @@ use crate::ghevent;
 use crate::message::{evaluationjob, Pr, Repo};
 use crate::worker;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
-
 pub struct EvaluationFilterWorker {
     acl: acl::ACL,
 }
@@ -20,8 +18,8 @@ impl worker::SimpleWorker for EvaluationFilterWorker {
 
     fn msg_to_job(
         &mut self,
-        _: &Deliver,
+        _: &str,
-        _: &BasicProperties,
+        _: &Option<String>,
         body: &[u8],
     ) -> Result<Self::J, String> {
         match serde_json::from_slice(body) {

ofborg/src/tasks/githubcommentfilter.rs

@@ -4,7 +4,6 @@ use crate::ghevent;
 use crate::message::{buildjob, evaluationjob, Pr, Repo};
 use crate::worker;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
 use uuid::Uuid;
 
 pub struct GitHubCommentWorker {
@@ -21,12 +20,7 @@ impl GitHubCommentWorker {
 impl worker::SimpleWorker for GitHubCommentWorker {
     type J = ghevent::IssueComment;
 
-    fn msg_to_job(
+    fn msg_to_job(&mut self, _: &str, _: &Option<String>, body: &[u8]) -> Result<Self::J, String> {
-        &mut self,
-        _: &Deliver,
-        _: &BasicProperties,
-        body: &[u8],
-    ) -> Result<Self::J, String> {
         match serde_json::from_slice(body) {
             Ok(e) => Ok(e),
             Err(e) => {

ofborg/src/tasks/githubcommentposter.rs

@@ -4,7 +4,6 @@ use crate::message::buildresult::{BuildResult, BuildStatus, LegacyBuildResult};
 use crate::message::Repo;
 use crate::worker;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
 use chrono::{DateTime, Utc};
 use hubcaps::checks::{CheckRunOptions, CheckRunState, Conclusion, Output};
 
@@ -42,12 +41,7 @@ impl PostableEvent {
 impl worker::SimpleWorker for GitHubCommentPoster {
     type J = PostableEvent;
 
-    fn msg_to_job(
+    fn msg_to_job(&mut self, _: &str, _: &Option<String>, body: &[u8]) -> Result<Self::J, String> {
-        &mut self,
-        _: &Deliver,
-        _: &BasicProperties,
-        body: &[u8],
-    ) -> Result<Self::J, String> {
         PostableEvent::from(body)
     }
 

ofborg/src/tasks/log_message_collector.rs

@@ -3,7 +3,6 @@ use crate::message::buildresult::BuildResult;
 use crate::worker;
 use crate::writetoline::LineWriter;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
 use lru_cache::LruCache;
 
 use std::fs::{self, File, OpenOptions};
@@ -174,8 +173,8 @@ impl worker::SimpleWorker for LogMessageCollector {
 
     fn msg_to_job(
         &mut self,
-        deliver: &Deliver,
+        routing_key: &str,
-        _: &BasicProperties,
+        _: &Option<String>,
         body: &[u8],
     ) -> Result<Self::J, String> {
         let message: MsgType;
@@ -203,7 +202,7 @@ impl worker::SimpleWorker for LogMessageCollector {
 
         Ok(LogMessage {
             from: LogFrom {
-                routing_key: deliver.routing_key.clone(),
+                routing_key: routing_key.to_string(),
                 attempt_id,
             },
             message,

ofborg/src/tasks/statscollector.rs

@@ -1,8 +1,6 @@
 use crate::stats;
 use crate::worker;
 
-use amqp::protocol::basic::{BasicProperties, Deliver};
-
 pub struct StatCollectorWorker<E> {
     events: E,
     collector: stats::MetricCollector,
@@ -17,12 +15,7 @@ impl<E: stats::SysEvents + 'static> StatCollectorWorker<E> {
 impl<E: stats::SysEvents + 'static> worker::SimpleWorker for StatCollectorWorker<E> {
     type J = stats::EventMessage;
 
-    fn msg_to_job(
+    fn msg_to_job(&mut self, _: &str, _: &Option<String>, body: &[u8]) -> Result<Self::J, String> {
-        &mut self,
-        _: &Deliver,
-        _: &BasicProperties,
-        body: &[u8],
-    ) -> Result<Self::J, String> {
         match serde_json::from_slice(body) {
             Ok(e) => Ok(e),
             Err(_) => {

ofborg/src/worker.rs

@@ -1,5 +1,5 @@
 use amqp::protocol::basic::{BasicProperties, Deliver};
-use amqp::{Basic, Channel, Consumer};
+use amqp::Basic;
 use serde::Serialize;
 
 use std::marker::Send;
@@ -53,15 +53,15 @@ where
     }))
 }
 
-pub trait SimpleWorker: Send + 'static {
+pub trait SimpleWorker: Send {
     type J: Send;
 
     fn consumer(&mut self, job: &Self::J) -> Actions;
 
     fn msg_to_job(
         &mut self,
-        method: &Deliver,
+        method: &str,
-        headers: &BasicProperties,
+        headers: &Option<String>,
         body: &[u8],
     ) -> Result<Self::J, String>;
 }
@@ -70,15 +70,17 @@ pub fn new<T: SimpleWorker>(worker: T) -> Worker<T> {
     Worker { internal: worker }
 }
 
-impl<T: SimpleWorker + Send> Consumer for Worker<T> {
+impl<T: SimpleWorker + Send> amqp::Consumer for Worker<T> {
     fn handle_delivery(
         &mut self,
-        channel: &mut Channel,
+        channel: &mut amqp::Channel,
         method: Deliver,
         headers: BasicProperties,
         body: Vec<u8>,
     ) {
-        let job = self.internal.msg_to_job(&method, &headers, &body);
+        let job = self
+            .internal
+            .msg_to_job(&method.routing_key, &headers.content_type, &body);
 
         if let Err(e) = job {
             error!("Error decoding job: {:?}", e);