2ff6ab1f-75d0-4794-ad18-1c11b64ed36a

Committed 04 Feb 2025 08:50AM UTC coverage: 86.912% (+0.007%) from 86.905%

Build # 2ff6ab1f-75d0-4794-ad18-1c11b64ed36a

Build Type

push

circleci

Committed by

web-flow

Commit Message

[SQS] Stop blocking Queue.get() call on LocalStack shutdown (#12214)

Coverage Stats

44 of 49 new or added lines in 3 files covered. (89.8%)

66 existing lines in 8 files now uncovered.

61454 of 70708 relevant lines covered (86.91%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.24

/localstack-core/localstack/services/sqs/queue.py

import time
from queue import Empty, PriorityQueue, Queue


class InterruptibleQueue(Queue):
    # 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.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.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.is_shutdown:  # additional shutdown check
                    remaining = endtime - time.time()
                    if remaining <= 0.0:
                        raise Empty
                    self.not_empty.wait(remaining)
            if self.is_shutdown:  # additional shutdown check
                raise Empty
            item = self._get()
            self.not_full.notify()
            return item

    def shutdown(self):
        """
        `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()


class InterruptiblePriorityQueue(PriorityQueue, InterruptibleQueue):
    pass

1	import time	1✔
2	from queue import Empty, PriorityQueue, Queue	1✔
3
4
5	class InterruptibleQueue(Queue):	1✔
6	# is_shutdown is used to check whether we have triggered a shutdown of the Queue
7	is_shutdown: bool	1✔
8
9	def __init__(self, maxsize=0):	1✔
10	super().__init__(maxsize)	1✔
11	self.is_shutdown = False	1✔
12
13	def get(self, block=True, timeout=None):	1✔
14	with self.not_empty:	1✔
15	if not block:	1✔
16	if not self._qsize():	1✔
17	raise Empty	1✔
18	elif timeout is None:	1✔
NEW 19	while not self._qsize() and not self.is_shutdown: # additional shutdown check	×
NEW 20	self.not_empty.wait()	×
21	elif timeout < 0:	1✔
NEW 22	raise ValueError("'timeout' must be a non-negative number")	×
23	else:
24	endtime = time.time() + timeout	1✔
25	while not self._qsize() and not self.is_shutdown: # additional shutdown check	1✔
26	remaining = endtime - time.time()	1✔
27	if remaining <= 0.0:	1✔
28	raise Empty	1✔
29	self.not_empty.wait(remaining)	1✔
30	if self.is_shutdown: # additional shutdown check	1✔
NEW 31	raise Empty	×
32	item = self._get()	1✔
33	self.not_full.notify()	1✔
34	return item	1✔
35
36	def shutdown(self):	1✔
37	"""
38	`shutdown` signals to stop all current and future `Queue.get` calls from executing.
39
40	This is helpful for exiting otherwise blocking calls early.
41	"""
42	with self.not_empty:	1✔
43	self.is_shutdown = True	1✔
44	self.not_empty.notify_all()	1✔
45
46
47	class InterruptiblePriorityQueue(PriorityQueue, InterruptibleQueue):	1✔
48	pass	1✔

localstack / localstack / 2ff6ab1f-75d0-4794-ad18-1c11b64ed36a

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous