feat: made generic delegate for easy implementation of new handlers

Author: raphael-goetz

src/flow_queue/delegate.rs

@@ -1,32 +1,68 @@
-use lapin::message::{DeliveryResult};
+use lapin::message::{Delivery, DeliveryResult};
 use lapin::ConsumerDelegate;
+use log::debug;
 use std::future::Future;
 use std::pin::Pin;
 
-struct QueueDelegate;
+/// Delegate trait to implement.
+///
+/// Use as delegate for RabbitMQ
+///
+/// # Example
+/// ```
+/// use lapin::message::Delivery;
+/// use code0_flow::flow_queue::delegate::Delegate;
+///
+/// struct HttpDelegate;
+///
+/// impl Delegate for HttpDelegate {
+///     fn handle_delivery(&self, delivery: Delivery) {
+///         todo!("Handle delivery!")
+///     }
+/// }
+/// ```
+pub trait Delegate {
+    fn handle_delivery(&self, delivery: Delivery);
+}
+
+pub struct QueueDelegate<T: Delegate> {
+    pub delegate: T,
+}
+
+impl<T: Delegate> QueueDelegate<T> {
+    pub fn new(delegate: T) -> Self {
+        QueueDelegate { delegate }
+    }
+
+    pub fn deliver(&self, delivery: Delivery) {
+        self.delegate.handle_delivery(delivery);
+    }
+}
 
-impl ConsumerDelegate for QueueDelegate {
+impl<T: Delegate> ConsumerDelegate for QueueDelegate<T> {
     fn on_new_delivery(
         &self,
         delivery: DeliveryResult,
     ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-        let optional_delivery = match delivery {
-            Ok(option) => option,
-            Err(error) => {
-                todo!("error handling")
-            }
-        };
+        async move {
+            let optional_delivery = match delivery {
+                Ok(option) => option,
+                Err(_) => return,
+            };
+            let delivery = match optional_delivery {
+                Some(del) => del,
+                None => return,
+            };
 
-        let delivery = match optional_delivery {
-            Some(del) => del,
-            None => {
-                todo!("error handling")
-            }
-        };
-        todo!("consumer shoud consume the data of delivy as &Vec<u8>")
+            self.delegate.handle_delivery(delivery);
+        }
     }
 
     fn drop_prefetched_messages(&self) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-        todo!("")
+        let future = async move {
+            debug!("Dropping prefetched messages...");
+        };
+
+        Box::pin(future)
     }
 }

src/flow_queue/handler.rs

@@ -1,4 +1,5 @@
 use crate::flow_queue::connection::{get_flow_channel, FlowChannel, FlowQueue};
+use crate::flow_queue::delegate::{Delegate, QueueDelegate};
 use crate::flow_queue::name::{QueueName, QueuePrefix, QueueProtocol};
 use lapin::message::{Delivery, DeliveryResult};
 use lapin::options::{BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions};
@@ -36,6 +37,7 @@ pub async fn declare_queues(flow_channel: FlowChannel, names: Vec<QueueName>) {
     }
 }
 
+/// Sends a message into a queue
 pub async fn send_message(
     flow_channel: FlowChannel,
     queue_name: QueueName,
@@ -60,19 +62,59 @@ pub async fn send_message(
     }
 }
 
-pub async fn consume_message(channel: FlowChannel, queue_protocol: QueueProtocol) {
-    let name = QueuePrefix::Send + queue_protocol;
+/// Consumes a message
+///
+/// Creates a delegate that waits on messages and consumes them.
+///
+/// # Params
+/// - channel: FlowChannel of the send message
+/// - queue_name: Name of the Queue that should be listened to
+/// - delegate: Consumer delegate of the message
+///
+/// # Example
+/// ```
+/// use lapin::message::Delivery;
+/// use code0_flow::flow_queue::delegate::Delegate;
+/// use code0_flow::flow_queue::connection::get_flow_channel;
+/// use code0_flow::flow_queue::name::{QueueName, QueuePrefix, QueueProtocol};
+/// use code0_flow::flow_queue::handler::consume_message;
+///
+/// struct HttpDelegate;
+///
+/// impl Delegate for HttpDelegate {
+///     fn handle_delivery(&self, delivery: Delivery) {
+///         todo!("Handle delivery!")
+///     }
+/// }
+///
+/// async fn main() {
+///     let uri = "abc";
+///     let channel = get_flow_channel(uri).await;
+///     let queue_name = QueueName {
+///         prefix: QueuePrefix::Send,
+///         protocol: QueueProtocol::Rest,
+///     };
+///
+///     consume_message(channel, queue_name, HttpDelegate).await;
+/// }
+/// ```
+pub async fn consume_message<T: Delegate>(
+    channel: FlowChannel,
+    queue_name: QueueName,
+    delegate: T,
+) {
+    let name = queue_name.prefix + queue_name.protocol;
     let channel_arc = channel.lock().await;
 
     let mut consumer = channel_arc
         .basic_consume(
             &*name,
-            "my_consumer",
+            "",
             BasicConsumeOptions::default(),
             FieldTable::default(),
         )
         .await
         .unwrap();
 
-    consumer.set_delegate(SendQueueDelegate);
-}
+    consumer.set_delegate(QueueDelegate { delegate });
+}