Fix "Event loop closed" error on Windows

  For more information look at
  aio-libs/aiohttp#4324

yddg/constants.py

@@ -33,7 +33,31 @@ async def aenumerate(
         i += 1
 
 
-async def path_list_agen(
-        path_list: List[Any]) -> AsyncGenerator[Any, None]:
+async def path_list_agen(path_list: List[Any]) -> AsyncGenerator[Any, None]:
     for path in path_list:
         yield path
+
+
+def event_loop_closed_workaround() -> None:
+    import sys
+
+    # thanks for workaround:
+    # https://github.com/aio-libs/aiohttp/issues/4324#issuecomment-733884349
+    if (sys.platform.startswith("win")
+            and sys.version_info[2] < 10):
+        from asyncio.proactor_events import _ProactorBasePipeTransport
+        from functools import wraps
+
+        def silence_del(func):
+            @wraps(func)
+            def wrapper(self, *args, **kwargs):
+                try:
+                    return func(self, *args, **kwargs)
+                except RuntimeError as e:
+                    if str(e) != 'Event loop is closed':
+                        raise
+
+            return wrapper
+
+        _ProactorBasePipeTransport.__del__ = silence_del(
+            _ProactorBasePipeTransport.__del__)

yddg/data_generator.py

@@ -20,7 +20,6 @@ async def flush(queue: Union[T.YDiskPathQueue, T.ItemQueue],
 
 
 class YDDataGenerator(Iterable):
-
     def __init__(
         self,
         urls: List[str],
@@ -70,6 +69,12 @@ async def __path_extracting(self) -> None:
         await self.paths_queue.put(None)
 
     async def start(self) -> None:
+        # To silence 'Event loop is closed' RuntimeError in Windows
+        # and Python < 3.10
+        # For more information read:
+        # https://github.com/aio-libs/aiohttp/issues/4324
+        const.event_loop_closed_workaround()
+
         self.path_extract_task = asyncio.ensure_future(
             self.__path_extracting())
         self.item_extract_task = asyncio.ensure_future(
@@ -91,6 +96,14 @@ async def stop(self) -> None:
         await self.paths_queue.join()
         await self.item_queue.join()
 
+        if (self.path_extract_task is not None
+                and not self.path_extract_task.cancelled()):
+            self.path_extract_task.cancel()
+
+        if (self.item_extract_task is not None
+                and not self.item_extract_task.cancelled()):
+            self.item_extract_task.cancel()
+
     async def __anext__(self):
         item = await self.item_queue.get()
         self.item_queue.task_done()