13833975459

Committed 13 Mar 2025 11:54AM UTC coverage: 88.297% (-4.0%) from 92.319%

Build # 13833975459

Build Type

push

github

Committed by

jell-o-fishi

Commit Message

fixed starlette/fastapi example.

Coverage Stats

781 of 959 branches covered (81.44%)

Branch coverage included in aggregate %.

0 of 2 new or added lines in 1 file covered. (0.0%)

172 existing lines in 9 files now uncovered.

3844 of 4279 relevant lines covered (89.83%)

2.68 hits per line

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

56.76

/rsocket/queue_peekable.py

import sys
from asyncio import Queue, QueueEmpty


class QueuePeekable(Queue):

    async def peek(self):
        """Peek the next item in the queue.

        If queue is empty, wait until an item is available.
        """
        while self.empty():
            getter = self._get_loop().create_future()
            self._getters.append(getter)
            try:
                await getter
            except Exception:
                getter.cancel()  # Just in case getter is not done yet.
                try:
                    # Clean self._getters from canceled getters.
                    self._getters.remove(getter)
                except ValueError:
                    # The getter could be removed from self._getters by a
                    # previous put_nowait call.
                    pass
                if not self.empty() and not getter.cancelled():
                    # We were woken up by put_nowait(), but can't take
                    # the call.  Wake up the next in line.
                    self._wakeup_next(self._getters)
                raise
        return self.peek_nowait()

    def peek_nowait(self):
        """Peek the next item in the queue.

        Return an item if one is immediately available, else raise QueueEmpty.
        """
        if self.empty():
            raise QueueEmpty

        item = self._queue[0]
        self._wakeup_next(self._putters)
        return item


if sys.version_info < (3, 10):
    class QueuePeekableBackwardCompatible(QueuePeekable):
        def _get_loop(self):
            return self._loop


    QueuePeekable = QueuePeekableBackwardCompatible

1	import sys	3✔
2	from asyncio import Queue, QueueEmpty	3✔
3
4
5	class QueuePeekable(Queue):	3✔
6
7	async def peek(self):	3✔
8	"""Peek the next item in the queue.
9
10	If queue is empty, wait until an item is available.
11	"""
12	while self.empty():	3✔
13	getter = self._get_loop().create_future()	3✔
14	self._getters.append(getter)	3✔
15	try:	3✔
16	await getter	3✔
17	except Exception:	3✔
18	getter.cancel() # Just in case getter is not done yet.	×
19	try:	×
20	# Clean self._getters from canceled getters.
21	self._getters.remove(getter)	×
22	except ValueError:	×
23	# The getter could be removed from self._getters by a
24	# previous put_nowait call.
25	pass
26	if not self.empty() and not getter.cancelled():	×
27	# We were woken up by put_nowait(), but can't take
28	# the call. Wake up the next in line.
29	self._wakeup_next(self._getters)	×
30	raise	×
31	return self.peek_nowait()	3✔
32
33	def peek_nowait(self):	3✔
34	"""Peek the next item in the queue.
35
36	Return an item if one is immediately available, else raise QueueEmpty.
37	"""
38	if self.empty():	3!
39	raise QueueEmpty	×
40
41	item = self._queue[0]	3✔
42	self._wakeup_next(self._putters)	3✔
43	return item	3✔
44
45
46	if sys.version_info < (3, 10):	3!
UNCOV 47	class QueuePeekableBackwardCompatible(QueuePeekable):	×
UNCOV 48	def _get_loop(self):	×
UNCOV 49	return self._loop	×
50
51
UNCOV 52	QueuePeekable = QueuePeekableBackwardCompatible	×

rsocket / rsocket-py / 13833975459

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