Coverage for kv4p/flow_control.py: 100%

1"""HTTP/2-like flow-control window for outgoing vendor frames.""" 

2 

3from __future__ import annotations 

4 

5import threading 

6import time 

7 

8 

9class FlowControlWindow: 

10 """Tracks how many bytes may still be sent before a WINDOW_UPDATE is required.""" 

11 

12 def __init__(self, initial_size: int = 2048) -> None: 

13 self._lock = threading.Condition() 

14 self._size = initial_size 

15 

16 def reset(self, size: int) -> None: 

17 """Set the window to a new value (e.g. after HELLO) and wake waiters.""" 

18 with self._lock: 

19 self._size = size 

20 self._lock.notify_all() 

21 

22 def add(self, size: int) -> None: 

23 """Add bytes freed by a WINDOW_UPDATE and wake waiters.""" 

24 with self._lock: 

25 self._size += size 

26 self._lock.notify_all() 

27 

28 def claim(self, size: int, timeout: float = 1.0) -> bool: 

29 """Block until `size` bytes are available and deduct them; False on timeout.""" 

30 deadline = time.monotonic() + timeout 

31 with self._lock: 

32 while self._size < size: 

33 remaining = deadline - time.monotonic() 

34 if remaining <= 0: 

35 return False 

36 self._lock.wait(timeout=remaining) 

37 self._size -= size 

38 return True