feat: Complete WebSocket integration with core services
- Enhanced DownloadService broadcasts for all queue operations - Download progress, complete, and failed broadcasts with full metadata - Queue operations (add, remove, reorder, retry, clear) broadcast queue status - Queue control (start, stop, pause, resume) broadcasts state changes - AnimeService scan progress fully integrated with ProgressService - Scan lifecycle events (start, update, complete, fail) broadcasted - Progress tracking via ProgressService to scan_progress room - ProgressService WebSocket integration - Broadcast callback registered during application startup - All progress types route to appropriate rooms - Throttled broadcasts for performance (>1% changes) - Comprehensive integration tests - Test download progress and completion broadcasts - Test queue operation broadcasts - Test scan progress lifecycle - Test progress service integration - End-to-end flow testing - Updated infrastructure documentation - Detailed broadcast message formats - Room structure and subscription patterns - Production deployment considerations - Architecture benefits and scalability notes
This commit is contained in:
@@ -113,12 +113,21 @@ class DownloadService:
|
||||
logger.debug("Broadcast callback registered")
|
||||
|
||||
async def _broadcast_update(self, update_type: str, data: dict) -> None:
|
||||
"""Broadcast update to connected WebSocket clients."""
|
||||
"""Broadcast update to connected WebSocket clients.
|
||||
|
||||
Args:
|
||||
update_type: Type of update (download_progress, queue_status, etc.)
|
||||
data: Update data to broadcast
|
||||
"""
|
||||
if self._broadcast_callback:
|
||||
try:
|
||||
await self._broadcast_callback(update_type, data)
|
||||
except Exception as e:
|
||||
logger.error("Failed to broadcast update", error=str(e))
|
||||
logger.error(
|
||||
"Failed to broadcast update",
|
||||
update_type=update_type,
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
def _generate_item_id(self) -> str:
|
||||
"""Generate unique identifier for download items."""
|
||||
@@ -238,9 +247,15 @@ class DownloadService:
|
||||
|
||||
self._save_queue()
|
||||
|
||||
# Broadcast update
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_updated", {"added_ids": created_ids}
|
||||
"queue_status",
|
||||
{
|
||||
"action": "items_added",
|
||||
"added_ids": created_ids,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
return created_ids
|
||||
@@ -288,8 +303,15 @@ class DownloadService:
|
||||
|
||||
if removed_ids:
|
||||
self._save_queue()
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_updated", {"removed_ids": removed_ids}
|
||||
"queue_status",
|
||||
{
|
||||
"action": "items_removed",
|
||||
"removed_ids": removed_ids,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
return removed_ids
|
||||
@@ -334,9 +356,17 @@ class DownloadService:
|
||||
self._pending_queue = deque(queue_list)
|
||||
|
||||
self._save_queue()
|
||||
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_reordered",
|
||||
{"item_id": item_id, "position": new_position}
|
||||
"queue_status",
|
||||
{
|
||||
"action": "queue_reordered",
|
||||
"item_id": item_id,
|
||||
"new_position": new_position,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
logger.info(
|
||||
@@ -410,13 +440,31 @@ class DownloadService:
|
||||
"""Pause download processing."""
|
||||
self._is_paused = True
|
||||
logger.info("Download queue paused")
|
||||
await self._broadcast_update("queue_paused", {})
|
||||
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_paused",
|
||||
{
|
||||
"is_paused": True,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
async def resume_queue(self) -> None:
|
||||
"""Resume download processing."""
|
||||
self._is_paused = False
|
||||
logger.info("Download queue resumed")
|
||||
await self._broadcast_update("queue_resumed", {})
|
||||
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_resumed",
|
||||
{
|
||||
"is_paused": False,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
async def clear_completed(self) -> int:
|
||||
"""Clear completed downloads from history.
|
||||
@@ -427,6 +475,19 @@ class DownloadService:
|
||||
count = len(self._completed_items)
|
||||
self._completed_items.clear()
|
||||
logger.info("Cleared completed items", count=count)
|
||||
|
||||
# Broadcast queue status update
|
||||
if count > 0:
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_status",
|
||||
{
|
||||
"action": "completed_cleared",
|
||||
"cleared_count": count,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
return count
|
||||
|
||||
async def retry_failed(
|
||||
@@ -471,8 +532,15 @@ class DownloadService:
|
||||
|
||||
if retried_ids:
|
||||
self._save_queue()
|
||||
# Broadcast queue status update
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"items_retried", {"item_ids": retried_ids}
|
||||
"queue_status",
|
||||
{
|
||||
"action": "items_retried",
|
||||
"retried_ids": retried_ids,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
return retried_ids
|
||||
@@ -530,7 +598,11 @@ class DownloadService:
|
||||
self._broadcast_update(
|
||||
"download_progress",
|
||||
{
|
||||
"download_id": item.id,
|
||||
"item_id": item.id,
|
||||
"serie_name": item.serie_name,
|
||||
"season": item.episode.season,
|
||||
"episode": item.episode.episode,
|
||||
"progress": item.progress.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
@@ -615,7 +687,17 @@ class DownloadService:
|
||||
)
|
||||
|
||||
await self._broadcast_update(
|
||||
"download_completed", {"item_id": item.id}
|
||||
"download_complete",
|
||||
{
|
||||
"download_id": item.id,
|
||||
"item_id": item.id,
|
||||
"serie_name": item.serie_name,
|
||||
"season": item.episode.season,
|
||||
"episode": item.episode.episode,
|
||||
"downloaded_mb": item.progress.downloaded_mb
|
||||
if item.progress
|
||||
else 0,
|
||||
},
|
||||
)
|
||||
else:
|
||||
raise AnimeServiceError("Download returned False")
|
||||
@@ -643,7 +725,15 @@ class DownloadService:
|
||||
|
||||
await self._broadcast_update(
|
||||
"download_failed",
|
||||
{"item_id": item.id, "error": item.error},
|
||||
{
|
||||
"download_id": item.id,
|
||||
"item_id": item.id,
|
||||
"serie_name": item.serie_name,
|
||||
"season": item.episode.season,
|
||||
"episode": item.episode.episode,
|
||||
"error": item.error,
|
||||
"retry_count": item.retry_count,
|
||||
},
|
||||
)
|
||||
|
||||
finally:
|
||||
@@ -698,6 +788,16 @@ class DownloadService:
|
||||
asyncio.create_task(self._queue_processor())
|
||||
|
||||
logger.info("Download queue service started")
|
||||
|
||||
# Broadcast queue started event
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_started",
|
||||
{
|
||||
"is_running": True,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Stop the download queue processor."""
|
||||
@@ -726,6 +826,16 @@ class DownloadService:
|
||||
self._executor.shutdown(wait=True)
|
||||
|
||||
logger.info("Download queue service stopped")
|
||||
|
||||
# Broadcast queue stopped event
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._broadcast_update(
|
||||
"queue_stopped",
|
||||
{
|
||||
"is_running": False,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# Singleton instance
|
||||
|
||||
Reference in New Issue
Block a user