Merge branch 'main' of https://github.com/ModelSurge/sd-webui-comfyui …

…into multi-clients

lib_comfyui/ipc/strategies.py

@@ -8,6 +8,9 @@
 from typing import IO, Union
 
 
+BytesLike = Union[bytes, bytearray, memoryview]
+
+
 __all__ = [
     'FileSystemIpcStrategy',
     'SharedMemoryIpcStrategy',
@@ -21,12 +24,12 @@ def is_empty(self, lock_file: IO) -> bool:
         pass
 
     @abstractmethod
-    def set_data(self, lock_file: IO, data: Union[bytes, bytearray, memoryview]) -> None:
+    def set_data(self, lock_file: IO, data: BytesLike) -> None:
         pass
 
     @abstractmethod
     @contextlib.contextmanager
-    def get_data(self, lock_file: IO) -> Union[bytes, bytearray, memoryview]:
+    def get_data(self, lock_file: IO) -> BytesLike:
         pass
 
     @abstractmethod
@@ -42,11 +45,11 @@ def is_empty(self, lock_file: IO) -> bool:
         lock_file.seek(0, os.SEEK_END)
         return lock_file.tell() == 0
 
-    def set_data(self, lock_file: IO, data: Union[bytes, bytearray, memoryview]):
+    def set_data(self, lock_file: IO, data: BytesLike):
         lock_file.write(data)
 
     @contextlib.contextmanager
-    def get_data(self, lock_file: IO) -> Union[bytes, bytearray, memoryview]:
+    def get_data(self, lock_file: IO) -> BytesLike:
         lock_file.seek(0)
         yield lock_file.read()
         self.clear(lock_file)
@@ -56,9 +59,6 @@ def clear(self, lock_file: IO):
         lock_file.truncate()
 
 
-BytesLike = Union[bytes, bytearray, memoryview]
-
-
 class SharedMemoryIpcStrategy(IpcStrategy):
     def __init__(self, shm_name: str):
         self._shm_name = shm_name