""" One connection per group for the whole run, and a ceiling on how many are being negotiated at once — the application's `ConnectionPool` rules (docs/MESHBAY_DESIGN.md §9.11). The ceiling is on concurrency, never a batch: a group whose node does not answer holds one place for its own deadline and delays nobody else. Six stays far below what the hub admits per account (32 pending offers), even with the person's application and a browser open at the same time. """ import asyncio import contextlib from .account import Account from .groups import open_group from .session import GroupSession MAX_IN_FLIGHT = 6 class Pool: def __init__(self, account: Account, max_in_flight: int = MAX_IN_FLIGHT): self.account = account self._gate = asyncio.Semaphore(max_in_flight) self._sessions: dict[str, asyncio.Task] = {} async def session(self, group: dict) -> GroupSession: """The group's session, negotiated once and shared by every caller.""" task = self._sessions.get(group["id"]) if task is None: task = asyncio.ensure_future(self._open(group)) self._sessions[group["id"]] = task return await asyncio.shield(task) async def _open(self, group: dict) -> GroupSession: async with self._gate: return await open_group(self.account, group) async def close(self) -> None: for task in self._sessions.values(): if task.done() and not task.cancelled() and task.exception() is None: await task.result().close() else: task.cancel() with contextlib.suppress(BaseException): await task self._sessions.clear()