wancat/vlchan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add retry to synchan

Browse Source
This commit is contained in:
Justin Lin
2025-10-11 16:39:13 +11:00
parent eb536f385a
commit 9ccc0cd2b5
1 changed files with 11 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
vlchan/synchan.py
View File

@@ -16,6 +16,7 @@ from concurrent.futures import ThreadPoolExecutor
@dataclass @dataclass
class SynchanState: class SynchanState:
"""State information from synchan server.""" """State information from synchan server."""
playing: bool playing: bool
currentTime: float currentTime: float
duration: float duration: float
@@ -25,7 +26,7 @@ class SynchanState:
class SynchanController: class SynchanController:
"""Controller for synchan server communication.""" """Controller for synchan server communication."""
def __init__(self, synchan_url: str = "http://localhost:3000"): def __init__(self, synchan_url: str = "http://localhost:3000"):
self.synchan_url = synchan_url self.synchan_url = synchan_url
self.headers = {"Content-Type": "application/json"} self.headers = {"Content-Type": "application/json"}
@@ -34,25 +35,17 @@ class SynchanController:
"""Seek to a specific time position.""" """Seek to a specific time position."""
print(f"Seeking to {to}") print(f"Seeking to {to}")
r = requests.post( r = requests.post(
f"{self.synchan_url}/trpc/admin.seek", f"{self.synchan_url}/trpc/admin.seek", headers=self.headers, data=str(to)
headers=self.headers,
data=str(to)
) )
print(r.text) print(r.text)
def play(self): def play(self):
"""Start playback.""" """Start playback."""
requests.post( requests.post(f"{self.synchan_url}/trpc/admin.play", headers=self.headers)
f"{self.synchan_url}/trpc/admin.play",
headers=self.headers
)
def pause(self): def pause(self):
"""Pause playback.""" """Pause playback."""
requests.post( requests.post(f"{self.synchan_url}/trpc/admin.pause", headers=self.headers)
f"{self.synchan_url}/trpc/admin.pause",
headers=self.headers
)
def create_socket(observer: ObserverBase[SynchanState], scheduler, synchan_url: str): def create_socket(observer: ObserverBase[SynchanState], scheduler, synchan_url: str):
@@ -66,8 +59,6 @@ def create_socket(observer: ObserverBase[SynchanState], scheduler, synchan_url:
@sio.event @sio.event
def connect_error(data): def connect_error(data):
print(f"Synchan connection failed: {data}") print(f"Synchan connection failed: {data}")
observer.on_error(data)
observer.on_completed()
@sio.event @sio.event
def control(data): def control(data):
@@ -87,11 +78,10 @@ def create_socket(observer: ObserverBase[SynchanState], scheduler, synchan_url:
@sio.event @sio.event
def disconnect(): def disconnect():
print("Disconnected from synchan server") print("Disconnected from synchan server")
observer.on_completed()
print(f"Connecting to synchan at {synchan_url}") print(f"Connecting to synchan at {synchan_url}")
try: try:
sio.connect(synchan_url, wait_timeout=5) sio.connect(synchan_url, wait_timeout=5, retry=True)
sio.wait() sio.wait()
except Exception as e: except Exception as e:
print(f"Failed to connect to synchan: {e}") print(f"Failed to connect to synchan: {e}")
@@ -99,15 +89,18 @@ def create_socket(observer: ObserverBase[SynchanState], scheduler, synchan_url:
observer.on_completed() observer.on_completed()
def create_synchan(synchan_url: str = "http://localhost:3000") -> Observable[SynchanState]: def create_synchan(
synchan_url: str = "http://localhost:3000",
) -> Observable[SynchanState]:
"""Create an observable stream of synchan state updates.""" """Create an observable stream of synchan state updates."""
# Create a thread pool for socket operations # Create a thread pool for socket operations
thread_count = multiprocessing.cpu_count() thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count) thread_pool_scheduler = ThreadPoolScheduler(thread_count)
return reactivex.create( return reactivex.create(
lambda observer, scheduler: create_socket(observer, scheduler, synchan_url) lambda observer, scheduler: create_socket(observer, scheduler, synchan_url)
).pipe( ).pipe(
ops.catch(lambda err, src: reactivex.of()), # ignore the error
ops.share(), ops.share(),
ops.subscribe_on(thread_pool_scheduler) ops.subscribe_on(thread_pool_scheduler)
) )