refactor: update DAQJob consume method to accept nowait parameter and…

… adjust sleep intervals

src/daq/alert/base.py

@@ -1,4 +1,3 @@
-import time
 from datetime import datetime
 from enum import Enum
 from typing import Any
@@ -36,10 +35,9 @@ def __init__(self, config: Any):
 
     def start(self):
         while True:
-            self.consume()
+            self.consume(nowait=False)
             self.alert_loop()
             self._alerts.clear()
-            time.sleep(0.5)
 
     def handle_message(self, message: DAQJobMessageAlert) -> bool:
         self._alerts.append(message)

src/daq/base.py

@@ -39,13 +39,16 @@ def __init__(self, config: Any):
         self._should_stop = False
         self.unique_id = str(uuid.uuid4())
 
-    def consume(self):
+    def consume(self, nowait=True):
         # consume messages from the queue
         while True:
             try:
-                message = self.message_in.get_nowait()
+                if nowait:
+                    message = self.message_in.get_nowait()
+                else:
+                    message = self.message_in.get()
                 if not self.handle_message(message):
-                    self.message_in.put_nowait(message)
+                    self.message_in.put(message)
             except Empty:
                 break
 

src/daq/jobs/handle_alerts.py

@@ -1,5 +1,3 @@
-import time
-
 from daq.alert.base import DAQJobMessageAlert
 from daq.base import DAQJob
 from daq.store.models import DAQJobMessageStore, StorableDAQJobConfig
@@ -17,8 +15,7 @@ class DAQJobHandleAlerts(DAQJob):
 
     def start(self):
         while True:
-            self.consume()
-            time.sleep(0.5)
+            self.consume(nowait=False)
 
     def handle_message(self, message: DAQJobMessageAlert) -> bool:
         super().handle_message(message)

src/daq/jobs/handle_stats.py

@@ -1,4 +1,3 @@
-import time
 from datetime import datetime
 from typing import Dict, Optional
 
@@ -25,8 +24,7 @@ class DAQJobHandleStats(DAQJob):
 
     def start(self):
         while True:
-            self.consume()
-            time.sleep(1)
+            self.consume(nowait=False)
 
     def handle_message(self, message: DAQJobMessageStats) -> bool:
         if not super().handle_message(message):

src/daq/jobs/healthcheck.py

@@ -11,6 +11,8 @@
 from daq.jobs.handle_stats import DAQJobMessageStats, DAQJobStatsDict
 from daq.models import DAQJobConfig, DAQJobStats
 
+HEALTHCHECK_LOOP_INTERVAL_SECONDS = 0.1
+
 
 class AlertCondition(str, Enum):
     SATISFIED = "satisfied"
@@ -115,7 +117,7 @@ def start(self):
         while True:
             self.consume()
             self.handle_checks()
-            time.sleep(0.5)
+            time.sleep(HEALTHCHECK_LOOP_INTERVAL_SECONDS)
 
     def handle_message(self, message: DAQJobMessageStats) -> bool:
         if not super().handle_message(message):

src/daq/store/base.py

@@ -4,6 +4,8 @@
 from daq.models import DAQJobMessage
 from daq.store.models import DAQJobMessageStore
 
+STORE_LOOP_INTERVAL_SECONDS = 0.1
+
 
 class DAQJobStore(DAQJob):
     allowed_store_config_types: list
@@ -12,7 +14,7 @@ def start(self):
         while True:
             self.consume()
             self.store_loop()
-            time.sleep(0.5)
+            time.sleep(STORE_LOOP_INTERVAL_SECONDS)
 
     def store_loop(self):
         raise NotImplementedError

src/tests/test_handle_alerts.py

@@ -62,16 +62,14 @@ def test_handle_message_invalid(self):
         self.assertFalse(result)
         self.daq_job.message_out.put.assert_not_called()
 
-    @patch("time.sleep", return_value=None)
-    def test_start(self, mock_sleep):
+    def test_start(self):
         self.daq_job.consume = MagicMock(side_effect=[None, Exception("Stop")])
 
         with self.assertRaises(Exception) as context:
             self.daq_job.start()
 
         self.assertEqual(str(context.exception), "Stop")
         self.daq_job.consume.assert_called()
-        mock_sleep.assert_called()
 
 
 if __name__ == "__main__":