Rename shutdown method and highlight custom implementation

gregfurman · gregfurman · commit 3b034a32314a · 2025-02-03T16:44:31.000+02:00
diff --git a/localstack-core/localstack/services/sqs/models.py b/localstack-core/localstack/services/sqs/models.py
@@ -301,7 +301,7 @@ def __init__(self, name: str, region: str, account_id: str, attributes=None, tag
         self.permissions = set()
         self.mutex = threading.RLock()
 
-    def do_shutdown(self):
+    def shutdown(self):
         pass
 
     def default_attributes(self) -> QueueAttributeMap:
@@ -739,7 +739,7 @@ def clear(self):
     def approx_number_of_messages(self):
         return self.visible.qsize()
 
-    def do_shutdown(self):
+    def shutdown(self):
         self.visible.shutdown()
 
     def put(
@@ -966,7 +966,7 @@ def approx_number_of_messages(self):
             n += len(message_group.messages)
         return n
 
-    def do_shutdown(self):
+    def shutdown(self):
         self.message_group_queue.shutdown()
 
     def get_message_group(self, message_group_id: str) -> MessageGroup:
diff --git a/localstack-core/localstack/services/sqs/provider.py b/localstack-core/localstack/services/sqs/provider.py
@@ -820,7 +820,7 @@ def on_before_stop(self):
         self._message_move_task_manager.close()
         for _, _, store in sqs_stores.iter_stores():
             for queue in store.queues.values():
-                queue.do_shutdown()
+                queue.shutdown()
 
         self._stop_cloudwatch_metrics_reporting()
 
diff --git a/localstack-core/localstack/services/sqs/queue.py b/localstack-core/localstack/services/sqs/queue.py
@@ -1,41 +1,46 @@
-import threading
 import time
 from queue import Empty, PriorityQueue, Queue
 
 
 class InterruptibleQueue(Queue):
-    shutdown_event: threading.Event
+    # is_shutdown is used to check whether we have triggered a shutdown of the Queue
+    is_shutdown: bool
 
     def __init__(self, maxsize=0):
         super().__init__(maxsize)
-        self.shutdown_event = threading.Event()
+        self.is_shutdown = False
 
     def get(self, block=True, timeout=None):
         with self.not_empty:
             if not block:
                 if not self._qsize():
                     raise Empty
             elif timeout is None:
-                while not self._qsize() and not self.shutdown_event.is_set():
+                while not self._qsize() and not self.is_shutdown:  # additional shutdown check
                     self.not_empty.wait()
             elif timeout < 0:
                 raise ValueError("'timeout' must be a non-negative number")
             else:
                 endtime = time.time() + timeout
-                while not self._qsize() and not self.shutdown_event.is_set():
+                while not self._qsize() and not self.is_shutdown:  # additional shutdown check
                     remaining = endtime - time.time()
                     if remaining <= 0.0:
                         raise Empty
                     self.not_empty.wait(remaining)
-            if self.shutdown_event.is_set():
+            if self.is_shutdown:  # additional shutdown check
                 raise Empty
             item = self._get()
             self.not_full.notify()
             return item
 
     def shutdown(self):
-        self.shutdown_event.set()
+        """
+        `shutdown` signals to stop all current and future `Queue.get` calls from executing.
+
+        This is helpful for exiting otherwise blocking calls early.
+        """
         with self.not_empty:
+            self.is_shutdown = True
             self.not_empty.notify_all()
 
 
