| 22 | del self.listeners[i] |
| 23 | |
| 24 | class SSEQueueWithTopic: |
| 25 | def __init__(self): |
| 26 | self.pubsub : dict[str, SSEQueue] = {} |
| 27 | |
| 28 | def listen(self, topic: str): |
| 29 | logger.info(f"LISTENING TO: {topic}") |
| 30 | if topic not in self.pubsub: |
| 31 | raise ValueError(f"Channel {topic} not found") |
| 32 | return self.pubsub[topic].listen() |
| 33 | |
| 34 | def publish(self, topic: str, message: str): |
| 35 | logger.debug(f"PUBLISHING TO: {topic} MESSAGE: {message}") |
| 36 | if topic not in self.pubsub: |
| 37 | raise ValueError(f"Topic {topic} not found") |
| 38 | self.pubsub[topic].announce(message=message) |
| 39 | |
| 40 | def add_topic(self, topic: str): |
| 41 | logger.info(f"SUBSCRIBING TO: {topic}") |
| 42 | if topic not in self.pubsub: |
| 43 | self.pubsub[topic] = SSEQueue() |
| 44 | return self.pubsub[topic] |
| 45 | |
| 46 | def get_topic(self, topic: str): |
| 47 | logger.info(f"GETTING TOPIC: {topic}") |
| 48 | if topic not in self.pubsub: |
| 49 | raise ValueError(f"Topic {topic} not found") |
| 50 | return self.pubsub[topic] |
| 51 | |
| 52 | def remove_topic(self, topic: str): |
| 53 | logger.info(f"REMOVING TOPIC: {topic}") |
| 54 | if topic not in self.pubsub: |
| 55 | raise ValueError(f"Topic {topic} not found") |
| 56 | del self.pubsub[topic] |