daniel156161
d3c914285c
Build and Push Docker Container / build-and-push (push) Successful in 2m3s
- Add a private SSE endpoint that emits file change events for authenticated clients. - Replace the watch sleep loop with watchdog local file events and remote SSE triggers. - Debounce local event bursts and keep watch running after transient sync failures. - Add watchdog as a standalone CLI dependency and cover watch behavior in tests. - Bump NanoShare to 1.25.0 and the standalone CLI to 0.4.0.
128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
import httpx
|
|
|
|
from .client import events_url
|
|
from .ignore import is_ignored, load_ignore_patterns
|
|
from .sync import relative, state_path, SYNC_DIR_NAME
|
|
|
|
SyncFunc = Callable[[object], int]
|
|
|
|
class _ChangeFlag:
|
|
def __init__(self) -> None:
|
|
self.event = threading.Event()
|
|
|
|
def notify(self) -> None:
|
|
self.event.set()
|
|
|
|
def wait(self, timeout: float) -> bool:
|
|
return self.event.wait(timeout)
|
|
|
|
def clear(self) -> None:
|
|
self.event.clear()
|
|
|
|
def _is_relevant_path(root: Path, path: str, ignore_patterns: list[str]) -> bool:
|
|
try:
|
|
candidate = Path(path).expanduser().resolve()
|
|
rel = relative(root, candidate)
|
|
except ValueError:
|
|
return False
|
|
if candidate == state_path(root) or root / SYNC_DIR_NAME in candidate.parents:
|
|
return False
|
|
return not is_ignored(rel, ignore_patterns)
|
|
|
|
def _remote_events_thread(args, flag: _ChangeFlag, stop: threading.Event) -> threading.Thread:
|
|
def run() -> None:
|
|
while not stop.is_set():
|
|
client = None
|
|
try:
|
|
from .app import make_client
|
|
client = make_client(args)
|
|
headers = {'Accept': 'text/event-stream', **client.auth_headers()}
|
|
with httpx.Client(timeout=None) as http:
|
|
with http.stream('GET', events_url(client, args.node), headers=headers) as response:
|
|
response.raise_for_status()
|
|
for line in response.iter_lines():
|
|
if stop.is_set():
|
|
return
|
|
if line.startswith('event: files.changed'):
|
|
flag.notify()
|
|
except Exception as exc:
|
|
if not stop.is_set():
|
|
print(f'remote event stream unavailable; retrying in 30s: {exc}', file=sys.stderr)
|
|
stop.wait(30)
|
|
finally:
|
|
if client:
|
|
client.close()
|
|
|
|
thread = threading.Thread(target=run, daemon=True)
|
|
thread.start()
|
|
return thread
|
|
|
|
def _watchdog_observer(root: Path, ignore_patterns: list[str], flag: _ChangeFlag):
|
|
from watchdog.events import FileSystemEventHandler
|
|
from watchdog.observers import Observer
|
|
|
|
class Handler(FileSystemEventHandler):
|
|
def on_any_event(self, event):
|
|
paths = [event.src_path]
|
|
dest_path = getattr(event, 'dest_path', None)
|
|
if dest_path:
|
|
paths.append(dest_path)
|
|
if any(_is_relevant_path(root, path, ignore_patterns) for path in paths):
|
|
flag.notify()
|
|
|
|
observer = Observer()
|
|
observer.schedule(Handler(), str(root), recursive=True)
|
|
observer.start()
|
|
return observer
|
|
|
|
def watch_loop(args, sync_func: SyncFunc) -> int:
|
|
root = Path(args.folder).expanduser().resolve()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
ignore_patterns = load_ignore_patterns(root, getattr(args, 'ignore', None))
|
|
interval = max(float(getattr(args, 'interval', 10.0)), 0.1)
|
|
flag = _ChangeFlag()
|
|
observer = None
|
|
stop_events = threading.Event()
|
|
events_thread = None
|
|
|
|
try:
|
|
try:
|
|
observer = _watchdog_observer(root, ignore_patterns, flag)
|
|
print(f'watching {args.folder} for local changes; remote events with {interval}s fallback')
|
|
except Exception as exc:
|
|
print(f'watchdog unavailable, polling every {interval}s: {exc}', file=sys.stderr)
|
|
print(f'watching {args.folder} every {interval}s')
|
|
|
|
events_thread = _remote_events_thread(args, flag, stop_events)
|
|
|
|
while True:
|
|
flag.clear()
|
|
code = sync_func(args)
|
|
if code:
|
|
print(f'watch sync failed; retrying in {interval}s', file=sys.stderr)
|
|
flag.wait(interval)
|
|
continue
|
|
|
|
if flag.wait(interval):
|
|
flag.clear()
|
|
time.sleep(1.0)
|
|
while flag.wait(1.0):
|
|
flag.clear()
|
|
except KeyboardInterrupt:
|
|
return 0
|
|
finally:
|
|
stop_events.set()
|
|
if observer:
|
|
observer.stop()
|
|
observer.join(timeout=5)
|
|
if events_thread:
|
|
events_thread.join(timeout=5)
|