Coverage for aiocoap/util/asyncio/timeoutdict.py: 100%
31 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-16 16:09 +0000
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-16 16:09 +0000
1# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
2#
3# SPDX-License-Identifier: MIT
5import asyncio
7class TimeoutDict:
8 """A dict-ish type whose entries live on a timeout; adding and accessing an
9 item each refreshes the timeout.
11 The timeout is a lower bound; items may live up to twice as long.
13 The container is implemented incompletely, with additions made on demand.
15 This is not thread safe.
16 """
18 def __init__(self, timeout: float):
19 self.timeout = timeout
20 """Timeout set on any access
22 This can be changed at runtime, but changes only take effect """
24 self._items = {}
25 """The actual dictionary"""
26 self._recently_accessed = None
27 """Items accessed since the timeout last fired"""
28 self._timeout = None
29 """Canceler for the timeout function"""
30 # Note: Without a __del__ implementation that even cancels, the object
31 # will be kept alive by the main loop for a timeout
33 def __getitem__(self, key):
34 result = self._items[key]
35 self._accessed(key)
36 return result
38 def __setitem__(self, key, value):
39 self._items[key] = value
40 self._accessed(key)
42 def _start_over(self):
43 """Clear _recently_accessed, set the timeout"""
44 self._timeout = asyncio.get_running_loop().call_later(self.timeout, self._tick)
45 self._recently_accessed = set()
47 def _accessed(self, key):
48 """Mark a key as recently accessed"""
49 if self._timeout is None:
50 self._start_over()
51 # No need to add the key, it'll live for this duration anyway
52 else:
53 self._recently_accessed.add(key)
55 def _tick(self):
56 self._items = {k: v for (k, v) in self._items.items() if k in self._recently_accessed}
57 if self._items:
58 self._start_over()
59 else:
60 self._timeout = None
61 self._recently_accessed = None