22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

86.11

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

import time
from queue import Empty, PriorityQueue, Queue


class InterruptibleQueue[T](Queue):
    # is_shutdown is used to check whether we have triggered a shutdown of the Queue
    is_shutdown: bool

    def __init__(self, maxsize: int = 0):
        super().__init__(maxsize)
        self.is_shutdown = False

    def get(self, block: bool = True, timeout: float | None = None) -> T:
        with self.not_empty:
            if self.is_shutdown:
                raise 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) -> None:
        """
        `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[T](PriorityQueue[T], InterruptibleQueue[T]):
    pass

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

localstack / localstack / 22709357475

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