fix: lifetime identifier was missing

Author: raphael-goetz

src/flow_queue/delegate.rs

@@ -3,48 +3,37 @@ use lapin::ConsumerDelegate;
 use log::debug;
 use std::future::Future;
 use std::pin::Pin;
+use std::sync::Arc;
 
 /// 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 {
+pub trait Delegate: Sync + Send {
     fn handle_delivery(&self, delivery: Delivery);
 }
 
 pub struct QueueDelegate<T: Delegate> {
-    pub delegate: T,
+    pub delegate: Arc<T>, // Use Arc for safe ownership transfer
 }
 
 impl<T: Delegate> QueueDelegate<T> {
     pub fn new(delegate: T) -> Self {
-        QueueDelegate { delegate }
+        QueueDelegate {
+            delegate: Arc::new(delegate),
+        }
     }
 
     pub fn deliver(&self, delivery: Delivery) {
         self.delegate.handle_delivery(delivery);
     }
 }
 
-impl<T: Delegate> ConsumerDelegate for QueueDelegate<T> {
+impl<T: Delegate + Send + Sync + 'static> ConsumerDelegate for QueueDelegate<T> {
     fn on_new_delivery(
         &self,
         delivery: DeliveryResult,
-    ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-        async move {
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+        let delegate = Arc::clone(&self.delegate);
+
+        Box::pin(async move {
             let optional_delivery = match delivery {
                 Ok(option) => option,
                 Err(_) => return,
@@ -54,15 +43,13 @@ impl<T: Delegate> ConsumerDelegate for QueueDelegate<T> {
                 None => return,
             };
 
-            self.delegate.handle_delivery(delivery);
-        }
+            delegate.handle_delivery(delivery); // Use cloned delegate
+        })
     }
 
-    fn drop_prefetched_messages(&self) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-        let future = async move {
+    fn drop_prefetched_messages(&self) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>> {
+        Box::pin(async move {
             debug!("Dropping prefetched messages...");
-        };
-
-        Box::pin(future)
+        })
     }
 }

src/flow_queue/handler.rs

@@ -1,14 +1,11 @@
-use crate::flow_queue::connection::{get_flow_channel, FlowChannel, FlowQueue};
+use crate::flow_queue::connection::FlowChannel;
 use crate::flow_queue::delegate::{Delegate, QueueDelegate};
-use crate::flow_queue::name::{QueueName, QueuePrefix, QueueProtocol};
-use lapin::message::{Delivery, DeliveryResult};
+use crate::flow_queue::name::QueueName;
 use lapin::options::{BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions};
-use lapin::protocol::basic::gen_return;
 use lapin::types::FieldTable;
-use lapin::{ConsumerDelegate, Error};
-use log::{debug, error, info};
-use std::future::Future;
-use std::pin::Pin;
+use lapin::Error;
+use log::info;
+use std::sync::Arc;
 
 /// # Declares all given queues
 ///
@@ -72,7 +69,7 @@ pub async fn send_message(
 /// - delegate: Consumer delegate of the message
 ///
 /// # Example
-/// ```
+/// ``` ignore
 /// use lapin::message::Delivery;
 /// use code0_flow::flow_queue::delegate::Delegate;
 /// use code0_flow::flow_queue::connection::get_flow_channel;
@@ -98,15 +95,15 @@ pub async fn send_message(
 ///     consume_message(channel, queue_name, HttpDelegate).await;
 /// }
 /// ```
-pub async fn consume_message<T: Delegate>(
+pub async fn consume_message<T: Delegate + 'static>(
     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
+    let consumer = channel_arc
         .basic_consume(
             &*name,
             "",
@@ -116,5 +113,7 @@ pub async fn consume_message<T: Delegate>(
         .await
         .unwrap();
 
-    consumer.set_delegate(QueueDelegate { delegate });
+    consumer.set_delegate(QueueDelegate {
+        delegate: Arc::new(delegate),
+    });
 }