Coverage for kilter/service/util.py: 94.03%

1# Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk> 

2# 

3# This Source Code Form is subject to the terms of the Mozilla Public 

4# License, v. 2.0. If a copy of the MPL was not distributed with this 

5# file, You can obtain one at http://mozilla.org/MPL/2.0/. 

6 

7""" 

8Common helper utilities 

9""" 

10 

11# mypy: disallow-any-explicit=False 

12 

13from __future__ import annotations (empty)

14 

15from collections.abc import AsyncIterator (empty)

16from collections.abc import Callable (empty)

17from contextlib import asynccontextmanager (empty)

18from typing import Generic (empty)

19from typing import Optional (empty)

20from typing import TypeVar (empty)

21 

22import anyio (empty)

23 

24T = TypeVar("T") (empty)

25Fn = TypeVar("Fn", bound=Callable[..., object]) (empty)

26 

27 

28class Broadcast(anyio.Condition, Generic[T]): (empty)

29 """ 

30 A reliable, blocking message queue for delivering to multiple listening tasks 

31 

32 Listeners must acquire the lock (by using the `Broadcast` instance as a context manager) 

33 before calling `Broadcast.receive()` or it will fail. If a listener is repeatedly 

34 awaiting messages in a loop, the loop should be inside the locked context or messages 

35 may be lost to race conditions. 

36 """ 

37 

38 def __init__(self) -> None: (empty)

39 super().__init__() (empty)

40 self.obj: Optional[T] = None (empty)

41 self.exc: Optional[BaseException|type[BaseException]] = None (empty)

42 

43 async def pre_receive_hook(self) -> None: (empty)

44 """ 

45 A hook for subclasses to inject synchronisation instructions before awaiting objects 

46 """ # noqa: D401 

47 

48 async def post_send_hook(self) -> None: (empty)

49 """ 

50 A hook for subclasses to inject synchronisation instructions after sending objects 

51 """ # noqa: D401 

52 

53 async def shutdown_hook(self) -> None: (empty)

54 """ 

55 A hook for subclasses to inject cleanup or synchronisation instructions on close 

56 

57 Users must ensure this method is called, especially if using a subclass which 

58 implements it. 

59 """ # noqa: D401 

60 

61 async def abort(self, exc: BaseException|type[BaseException]) -> None: (empty)

62 """ 

63 Send a notification to all listeners to abort by raising an exception 

64 """ 

65 async with self._ready(): (empty)

66 assert self.exc is None and self.obj is None (empty)

67 self.exc = exc (empty)

68 self.notify_all() (empty)

69 await self._post() (empty)

70 

71 async def send(self, obj: T) -> None: (empty)

72 """ 

73 Send a message object and block until all listeners have received it 

74 """ 

75 async with self._ready(): (empty)

76 assert self.exc is None and self.obj is None (empty)

77 self.obj = obj (empty)

78 self.notify_all() (empty)

79 await self._post() (empty)

80 

81 @asynccontextmanager (empty)

82 async def _ready(self) -> AsyncIterator[None]: (empty)

83 while 1: (empty)

84 await anyio.sleep(0.0) (empty)

85 async with self: (empty)

86 if self.obj is not None or self.exc is not None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true(empty)

87 continue 

88 yield (empty)

89 return (empty)

90 

91 async def _post(self) -> None: (empty)

92 await anyio.sleep(0.0) # ensure listeners have opportunity to wait for locks (empty)

93 await self.post_send_hook() (empty)

94 

95 # Ensure all listeners have had a chance to lock and process self.obj 

96 while 1: (empty)

97 async with self: (empty)

98 if self.statistics().lock_statistics.tasks_waiting: # pragma: no-branch (empty)

99 continue (empty)

100 self.obj = self.exc = None (empty)

101 break (empty)

102 

103 async def receive(self) -> T: (empty)

104 """ 

105 Listen for a single message and return it once it arrives 

106 """ 

107 await self.pre_receive_hook() (empty)

108 await self.wait() (empty)

109 if self.exc is not None: (empty)

110 raise self.exc (empty)

111 assert self.obj is not None (empty)

112 return self.obj (empty)

113 

114 

115def qualname(func: Fn) -> str: (empty)

116 """ 

117 Return a qualified name for a callable 

118 """ 

119 if func.__module__ == "__main__": 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true(empty)

120 return func.__qualname__ 

121 return f"{func.__module__}.{func.__qualname__}" (empty)