from __future__ import annotations
|
|
from dataclasses import dataclass
|
import time
|
from typing import Callable, Iterable
|
|
from .adb import RemoteFile
|
from .models import ContractError, ErrorCode
|
|
|
@dataclass(frozen=True)
|
class CacheMatch:
|
status: str
|
remote: RemoteFile | None
|
poll_count: int
|
ambiguous_paths: tuple[str, ...]
|
|
|
class CacheWatcher:
|
def __init__(self, list_files: Callable[[], tuple[RemoteFile, ...]], *, clock=time.monotonic,
|
sleep=time.sleep):
|
self.list_files = list_files
|
self.clock = clock
|
self.sleep = sleep
|
|
def baseline(self) -> dict[str, tuple[int, str]]:
|
return {f.path: (f.bytes, f.mtime_token) for f in self.list_files()}
|
|
def wait_for_unique_stable(self, baseline: dict[str, tuple[int, str]], *, timeout_ms: int,
|
poll_ms: int = 500, stable_polls: int = 2) -> CacheMatch:
|
deadline = self.clock() + timeout_ms / 1000
|
last: dict[str, tuple[int, str]] = {}
|
stable: dict[str, int] = {}
|
polls = 0
|
while self.clock() < deadline:
|
polls += 1
|
files = self.list_files()
|
changed = [f for f in files if baseline.get(f.path) != (f.bytes, f.mtime_token)]
|
for item in changed:
|
signature = (item.bytes, item.mtime_token)
|
stable[item.path] = stable.get(item.path, 0) + 1 if last.get(item.path) == signature else 1
|
last[item.path] = signature
|
ready = [f for f in changed if stable.get(f.path, 0) >= stable_polls and f.bytes > 0]
|
if len(ready) == 1:
|
return CacheMatch("UNIQUE_STABLE", ready[0], polls, ())
|
if len(ready) > 1:
|
return CacheMatch("AMBIGUOUS", None, polls, tuple(sorted(f.path for f in ready)))
|
self.sleep(min(poll_ms / 1000, max(0, deadline - self.clock())))
|
return CacheMatch("NONE_TIMEOUT", None, polls, ())
|
