1class Message:
2    def __init__(self, payload: str):
3        self.payload = payload
4        self.timestamp = datetime.now()
5
6    def get_payload(self) -> str:
7        return self.payload
8
9    def __str__(self) -> str:
10        return f"Message{{payload='{self.payload}'}}"

1class Subscriber(ABC):
2    @abstractmethod
3    def get_id(self) -> str:
4        pass
5
6    @abstractmethod
7    def on_message(self, message: Message):
8        pass
9
10class AlertSubscriber(Subscriber):
11    def __init__(self, subscriber_id: str):
12        self.id = subscriber_id
13
14    def get_id(self) -> str:
15        return self.id
16
17    def on_message(self, message: Message):
18        print(f"!!! [ALERT - {self.id}] : '{message.get_payload()}' !!!")
19
20class NewsSubscriber(Subscriber):
21    def __init__(self, subscriber_id: str):
22        self.id = subscriber_id
23
24    def get_id(self) -> str:
25        return self.id
26
27    def on_message(self, message: Message):
28        print(f"[Subscriber {self.id}] received message '{message.get_payload()}'")

1class Topic:
2    def __init__(self, name: str, delivery_executor: ThreadPoolExecutor):
3        self.name = name
4        self.delivery_executor = delivery_executor
5        self.subscribers: Set[Subscriber] = set()
6
7    def get_name(self) -> str:
8        return self.name
9
10    def add_subscriber(self, subscriber: Subscriber):
11        self.subscribers.add(subscriber)
12
13    def remove_subscriber(self, subscriber: Subscriber):
14        self.subscribers.discard(subscriber)
15
16    def broadcast(self, message: Message):
17        for subscriber in self.subscribers:
18            self.delivery_executor.submit(self._deliver_message, subscriber, message)
19
20    def _deliver_message(self, subscriber: Subscriber, message: Message):
21        try:
22            subscriber.on_message(message)
23        except Exception as e:
24            print(f"Error delivering message to subscriber {subscriber.get_id()}: {str(e)}")

1class PubSubService:
2    _instance = None
3    _lock = threading.Lock()
4
5    def __new__(cls):
6        if cls._instance is None:
7            with cls._lock:
8                if cls._instance is None:
9                    cls._instance = super().__new__(cls)
10                    cls._instance._initialized = False
11        return cls._instance
12
13    def __init__(self):
14        if self._initialized:
15            return
16        
17        self.topic_registry: Dict[str, Topic] = {}
18        # A cached thread pool is suitable for handling many short-lived, bursty tasks (message deliveries).
19        self.delivery_executor = ThreadPoolExecutor()
20        self._initialized = True
21
22    @classmethod
23    def get_instance(cls):
24        return cls()
25
26    def create_topic(self, topic_name: str):
27        if topic_name not in self.topic_registry:
28            self.topic_registry[topic_name] = Topic(topic_name, self.delivery_executor)
29        print(f"Topic {topic_name} created")
30
31    def subscribe(self, topic_name: str, subscriber: Subscriber):
32        topic = self.topic_registry.get(topic_name)
33        if topic is None:
34            raise ValueError(f"Topic not found: {topic_name}")
35        topic.add_subscriber(subscriber)
36        print(f"Subscriber '{subscriber.get_id()}' subscribed to topic: {topic_name}")
37
38    def unsubscribe(self, topic_name: str, subscriber: Subscriber):
39        topic = self.topic_registry.get(topic_name)
40        if topic is not None:
41            topic.remove_subscriber(subscriber)
42        print(f"Subscriber '{subscriber.get_id()}' unsubscribed from topic: {topic_name}")
43
44    def publish(self, topic_name: str, message: Message):
45        print(f"Publishing message to topic: {topic_name}")
46        topic = self.topic_registry.get(topic_name)
47        if topic is None:
48            raise ValueError(f"Topic not found: {topic_name}")
49        topic.broadcast(message)
50
51    def shutdown(self):
52        print("PubSubService shutting down...")
53        self.delivery_executor.shutdown(wait=True)
54        print("PubSubService shutdown complete.")

1class PubSubDemo:
2    @staticmethod  
3    def main():
4        pub_sub_service = PubSubService.get_instance()
5
6        # --- Create Subscribers ---
7        sports_fan1 = NewsSubscriber("SportsFan1")
8        sports_fan2 = NewsSubscriber("SportsFan2")
9        techie1 = NewsSubscriber("Techie1")
10        all_news_reader = NewsSubscriber("AllNewsReader")
11        system_admin = AlertSubscriber("SystemAdmin")
12
13        # --- Create Topics and Subscriptions ---
14        SPORTS_TOPIC = "SPORTS"
15        TECH_TOPIC = "TECH"
16        WEATHER_TOPIC = "WEATHER"
17
18        pub_sub_service.create_topic(SPORTS_TOPIC)
19        pub_sub_service.create_topic(TECH_TOPIC)
20        pub_sub_service.create_topic(WEATHER_TOPIC)
21
22        pub_sub_service.subscribe(SPORTS_TOPIC, sports_fan1)
23        pub_sub_service.subscribe(SPORTS_TOPIC, sports_fan2)
24        pub_sub_service.subscribe(SPORTS_TOPIC, all_news_reader)
25        pub_sub_service.subscribe(SPORTS_TOPIC, system_admin)
26
27        pub_sub_service.subscribe(TECH_TOPIC, techie1)
28        pub_sub_service.subscribe(TECH_TOPIC, all_news_reader)
29
30        print("\n--- Publishing Messages ---")
31
32        # --- Publish to SPORTS topic ---
33        pub_sub_service.publish(SPORTS_TOPIC, Message("Team A wins the championship!"))
34        # Expected: SportsFan1, SportsFan2, AllNewsReader, SystemAdmin receive this.
35
36        # --- Publish to TECH topic ---
37        pub_sub_service.publish(TECH_TOPIC, Message("New AI model released."))
38        # Expected: Techie1, AllNewsReader receive this.
39
40        # --- Publish to WEATHER topic (no subscribers) ---
41        pub_sub_service.publish(WEATHER_TOPIC, Message("Sunny with a high of 75°F."))
42        # Expected: Message is dropped.
43
44        # Allow some time for async messages to be processed
45        time.sleep(0.5)
46
47        print("\n--- Unsubscribing a user and re-publishing ---")
48
49        # SportsFan2 gets tired of sports news
50        pub_sub_service.unsubscribe(SPORTS_TOPIC, sports_fan2)
51
52        # Publish another message to SPORTS
53        pub_sub_service.publish(SPORTS_TOPIC, Message("Major player traded to Team B."))
54        # Expected: SportsFan1, AllNewsReader, SystemAdmin receive this. SportsFan2 does NOT.
55
56        # Give messages time to be delivered
57        time.sleep(0.5)
58
59        # --- Shutdown the service ---
60        pub_sub_service.shutdown()
61
62if __name__ == "__main__":
63    PubSubDemo.main()