| from __future__ import annotations # defer annotation evaluation (PEP 563) | |
| import threading # provides the Lock we use for thread-safe rotation | |
| from typing import List, Optional # type aliases for the signatures | |
| class RoundRobin: # cycles through endpoints so load is spread evenly | |
| def __init__(self, items: Optional[List] = None): # construct with optional seed list | |
| self._lock = threading.Lock() # guards _items and _index across request threads | |
| self._items: List = list(items) if items else [] # defensive copy of the seed list | |
| self._index = 0 # next slot to hand out | |
| def set_items(self, items: List) -> None: # swap in a fresh endpoint pool (e.g. after reload) | |
| with self._lock: # take the lock so concurrent next() callers don't see half-swapped state | |
| self._items = list(items) # store a copy so external mutation can't affect us | |
| if self._index >= len(self._items): # if the cursor now points past the end | |
| self._index = 0 # wrap it back to the start | |
| def next(self) -> Optional[object]: # return the next item, or None if pool empty | |
| with self._lock: # lock so two threads never get the same index | |
| if not self._items: # empty pool guard | |
| return None # caller decides how to handle no-endpoints | |
| item = self._items[self._index % len(self._items)] # modulo keeps index in bounds | |
| self._index = (self._index + 1) % len(self._items) # advance and wrap the cursor | |
| return item # hand back the selected item | |
| def __len__(self) -> int: # lets `len(RR)` report current pool size | |
| with self._lock: # lock because _items can be replaced concurrently | |
| return len(self._items) # current count of items in the pool | |