fix: add singleton

broadcast_service/_core.py

@@ -15,6 +15,7 @@
 import logging
 from typing import Optional, List, Callable
 from concurrent.futures import ThreadPoolExecutor
+from broadcast_service.singleton import Singleton
 
 __all__ = ['broadcast_service', 'BroadcastService', 'enable_log']
 
@@ -23,7 +24,7 @@ def enable_log():
     logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
 
 
-class BroadcastService:
+class BroadcastService(metaclass=Singleton):
     """
     This class implements broadcast mode, you can import the instance by single class.
     By BroadcastService, you can send topic message,it will automatically execute the

broadcast_service/singleton.py

@@ -0,0 +1,24 @@
+"""The singleton metaclass for ensuring only one instance of a class."""
+import abc
+
+
+class Singleton(abc.ABCMeta, type):
+    """
+    Singleton metaclass for ensuring only one instance of a class.
+    """
+
+    _instances = {}
+
+    def __call__(cls, *args, **kwargs):
+        """Call method for the singleton metaclass."""
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+
+class AbstractSingleton(abc.ABC, metaclass=Singleton):
+    """
+    Abstract singleton class for ensuring only one instance of a class.
+    """
+
+    pass

setup.py

@@ -20,7 +20,7 @@
 
 setuptools.setup(
     name="broadcast_service",
-    version="1.3.0",
+    version="1.3.1",
     author="Zeeland",
     author_email="[email protected]",
     description="A lightweight third-party broadcast/pubsub library",