Broker trait: Pop message returns Option

Author: magnunm

src/broker.rs

@@ -4,7 +4,7 @@ use anyhow::Result;
 pub trait Broker {
     fn push_message(&self, message: &Message) -> Result<()>;
 
-    fn pop_message(&self) -> Result<Message>;
+    fn pop_message(&self) -> Result<Option<Message>>;
 
     fn store_result(&self, result_message: ResultMessage) -> Result<()>;
 }

src/brokers/redis.rs

@@ -29,12 +29,12 @@ impl Broker for RedisBroker {
         Ok(())
     }
 
-    fn pop_message(&self) -> Result<crate::messages::Message> {
+    fn pop_message(&self) -> Result<Option<crate::messages::Message>> {
         let mut con = self.redis_client.get_connection()?;
         let serialized_message: Option<String> = con.rpop(&self.queue, None)?;
         match serialized_message {
-            Some(v) => Ok(serde_json::from_str(&v)?),
-            None => Err(anyhow::anyhow!("Task queue is empty")),
+            Some(v) => Ok(Some(serde_json::from_str(&v)?)),
+            None => Ok(None),
         }
     }
 

src/main.rs

@@ -29,15 +29,15 @@ impl Broker for InMemoryTestBroker {
         Ok(())
     }
 
-    fn pop_message(&self) -> anyhow::Result<Message> {
+    fn pop_message(&self) -> anyhow::Result<Option<Message>> {
         match self
             .queue
             .write()
             .expect("Failed to aquire lock")
             .pop_front()
         {
-            Some(message) => Ok(message),
-            None => Err(anyhow::anyhow!("No messages in queue!")),
+            Some(message) => Ok(Some(message)),
+            None => Ok(None),
         }
     }
 

src/worker.rs

@@ -19,6 +19,9 @@ impl<'a, B: Broker + 'static> Worker<'a, B> {
 
     pub fn take_first_task_in_queue(&self) -> Result<(), Error> {
         let message = self.app.broker.pop_message()?;
-        self.app.handle_message(&message)
+        match message {
+            Some(m) => self.app.handle_message(&m),
+            None => Err(anyhow::anyhow!("No messages in queue")),
+        }
     }
 }

tests/running_task_from_message_test.rs

@@ -30,15 +30,15 @@ impl Broker for InMemoryTestBroker {
         Ok(())
     }
 
-    fn pop_message(&self) -> anyhow::Result<Message> {
+    fn pop_message(&self) -> anyhow::Result<Option<Message>> {
         match self
             .queue
             .write()
             .expect("Failed to aquire lock")
             .pop_front()
         {
-            Some(message) => Ok(message),
-            None => Err(anyhow::anyhow!("No messages in queue!")),
+            Some(message) => Ok(Some(message)),
+            None => Ok(None),
         }
     }