server

  1"""Server for publisher subscriber system. For more information, please run `python server.py --help`"""
  2
  3import asyncio
  4import logging
  5import time
  6import functools
  7from argparse import ArgumentParser
  8from datetime import datetime
  9from threading import Lock, Thread
 10from typing import List, Optional, Union, Dict
 11
 12import socketio
 13from aiohttp import web
 14
 15from transport_message import TransportMessage
 16
 17
 18## Setup logging ##
 19# Set all loggers to ERROR level
 20loggers = [logging.getLogger(name) for name in logging.root.manager.loggerDict]
 21for logger in loggers:
 22    logger.setLevel(logging.ERROR)
 23
 24# Set server logger to INFO level
 25# Create file handler which logs even debug messages
 26file_handler = logging.FileHandler("server.log")
 27file_handler.setLevel(logging.DEBUG)
 28
 29# Create console handler
 30console_handler = logging.StreamHandler()
 31console_handler.setLevel(logging.DEBUG)
 32
 33# Create formatter and add it to the handlers
 34formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
 35file_handler.setFormatter(formatter)
 36console_handler.setFormatter(formatter)
 37
 38# Add the handlers to the logger
 39logging.getLogger().addHandler(file_handler)
 40logging.getLogger().addHandler(console_handler)
 41logging.getLogger().setLevel(logging.DEBUG)
 42
 43# logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
 44
 45
 46class ParallelTimer(Thread):
 47    """Class to manage a parallel timer in different thread on the server triggering a heart beat algorithm every 20 seconds."""
 48
 49    def __init__(self, server) -> None:
 50        """Constructor of ParallelTimer class.
 51
 52        :param server: server object
 53        """
 54        super().__init__()
 55        self.server = server
 56
 57    def run(self):
 58        """
 59        Starting parallel timer in loop.
 60        """
 61        while 1:
 62            heartbeat = self.server.heart_beat(20)
 63            asyncio.run(heartbeat)
 64
 65
 66class Topic:
 67    """Class to manage the Topics with needed data."""
 68
 69    name: Union[None, str] = None
 70    """name of the topic"""
 71    content: Union[None, str] = None
 72    """content of the topic"""
 73    subscribers: List[str] = []
 74    """list of subscribers"""
 75    timestamp: Union[None, int] = None
 76    """timestamp"""
 77    last_update: Union[None, int] = None
 78    """last update of topic"""
 79
 80
 81class Server:
 82    def __init__(self) -> None:
 83        self._list_of_topics: List[Topic] = []
 84        self._sid_ip_mapping: Dict[str, str] = {}
 85        self._lock = Lock()
 86
 87        self.sio = socketio.AsyncServer(
 88            async_mode="aiohttp", cors_allowed_origins="*", logger=False, engineio_logger=False
 89        )
 90        self.sio.event(self.connect)
 91        self.sio.on("SUBSCRIBE_TOPIC", self.handle_subscribe)
 92        self.sio.on("UNSUBSCRIBE_TOPIC", self.handle_unsubscribe)
 93        self.sio.on("PUBLISH_TOPIC", self.handle_publish)
 94        self.sio.on("LIST_TOPICS", self.handle_list_topics)
 95        self.sio.on("GET_TOPIC_STATUS", self.handle_topic_status)
 96
 97    def _check_data_none_decorator(func):
 98        """Decorator for checking if data is None.
 99        If data is None, the client will receive an error message.
100        """
101
102        @functools.wraps(func)
103        async def wrapper(self, *args, **kwargs):
104            sid = args[0]
105            data = args[1] if len(args) > 1 else None
106            if data is None:
107                response = TransportMessage(
108                    timestamp=int(time.time()), payload="Missing payload of type TransportMessage."
109                )
110                await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
111                logging.error("%s - %s", self._sid_ip_mapping[sid], response.payload)
112                return None
113            return await func(self, *args, **kwargs)
114
115        return wrapper
116
117    def _check_topic_decorator(func):
118        """Decorator for checking if topic exists.
119        If topic does not exist, the client will receive an error message.
120        """
121
122        @functools.wraps(func)
123        async def wrapper(self, *args, **kwargs):
124            sid = args[0]
125            data = args[1] if len(args) > 1 else None
126            try:
127                parsed_data = TransportMessage.parse_raw(data)
128            except Exception:
129                response = TransportMessage(timestamp=int(time.time()), payload="Invalid payload.")
130                await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
131                logging.error("%s - %s", self._sid_ip_mapping[sid], response.payload)
132                return None
133
134            # Check if data contains topic
135            if parsed_data.topic is None:
136                response = TransportMessage(timestamp=int(time.time()), payload="Missing parameter topic.")
137                await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
138                logging.error("%s - %s", self._sid_ip_mapping[sid], response.payload)
139                return None
140            return await func(self, *args, **kwargs)
141
142        return wrapper
143
144    async def connect(self, sid, environ, auth=None):
145        """Called when a client connects to the server.
146
147        :param sid: Generated session id
148        :param environ: Environment variables
149        :param auth: Unused
150        """
151        logging.info("%s - SID: %s connected", environ["aiohttp.request"].remote, sid)
152        self._sid_ip_mapping[sid] = environ["aiohttp.request"].remote
153
154    @_check_data_none_decorator
155    @_check_topic_decorator
156    async def handle_subscribe(self, sid, data=None) -> None:
157        """Called when a client subscribes to a topic.
158        If the topic does not exist, it will be created. If the client is already subscribed to the topic, nothing
159        changes. Otherwise the client will be subscribed to the topic and will receive updates.
160
161        :param sid: Generated session id
162        :param data: Data sent by the client
163        """
164        data = TransportMessage.parse_raw(data)
165        topic = self._get_topic_by_name(data.topic)
166        if topic is not None:
167            # Check if sid already subscribed to topic
168            if sid in topic.subscribers:
169                response = TransportMessage(timestamp=int(time.time()), payload=f"Already subscribed to {data.topic}.")
170            else:
171                # Subscribe to topic
172                topic.subscribers.append(sid)
173                response = TransportMessage(
174                    timestamp=int(time.time()), payload=f"Successfully subscribed to {data.topic}."
175                )
176        else:
177            # Create new topic if not already existing and subscribe
178            new_topic = Topic()
179            new_topic.name = data.topic
180            new_topic.subscribers.append(sid)
181            self._add_topic(new_topic)
182            response = TransportMessage(
183                timestamp=int(time.time()), payload=f"Created {data.topic} and successfully subscribed."
184            )
185
186        await self.sio.emit("PRINT_MESSAGE", response.json(), room=sid)
187        logging.info("%s - %s", self._sid_ip_mapping[sid], response.payload)
188
189    @_check_data_none_decorator
190    @_check_topic_decorator
191    async def handle_unsubscribe(self, sid, data=None) -> None:
192        """Called when a client unsubscribes from a topic.
193        If the client is not subscribed to the topic or topic does not exist, the client will receive an error message.
194        Otherwise the client will be unsubscribed from the topic and will not receive any updates.
195        If the topic has no subscribers left it will be deleted.
196
197        :param sid: Generated session id
198        :param data: Data sent by the client
199        """
200
201        data = TransportMessage.parse_raw(data)
202        topic = self._get_topic_by_name(data.topic)
203
204        if topic is not None:
205            # Check if sid subscribed to topic and unsubscribe
206            if sid in topic.subscribers:
207                topic.subscribers.remove(sid)
208                response = TransportMessage(
209                    timestamp=int(time.time()), payload=f"Successfully unsubscribed from {data.topic}."
210                )
211                # Delete topic if no subscribers left
212                if len(topic.subscribers) == 0:
213                    self._remove_topic(topic)
214            else:
215                # Not subscribed
216                response = TransportMessage(timestamp=int(time.time()), payload=f"Not subscribed to {data.topic}.")
217
218        else:
219            # Topic not existing
220            response = TransportMessage(timestamp=int(time.time()), payload=f"{data.topic} does not exist.")
221
222        await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
223        logging.info("%s - %s", self._sid_ip_mapping[sid], response.payload)
224
225    @_check_data_none_decorator
226    @_check_topic_decorator
227    async def handle_publish(self, sid, data=None) -> None:
228        """Called when a client publishes a message to a topic.
229        The message will be published to the topic and all subscribers will receive the message.
230
231        :param sid: Generated session id
232        :param data: Data sent by the client
233        """
234        data = TransportMessage.parse_raw(data)
235        topic = self._get_topic_by_name(data.topic)
236
237        # Check if data contains payload
238        if data.payload is None:
239            response = TransportMessage(timestamp=int(time.time()), payload="Missing parameter message.")
240            await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
241            return None
242
243        if topic is not None:
244            # Publish message to topic
245            topic.content = data.payload
246            topic.timestamp = data.timestamp
247            response = TransportMessage(
248                timestamp=int(time.time()), payload=f"Successfully published message to {data.topic}."
249            )
250            await self.update_topic(topic)
251        else:
252            # Topic not existing
253            response = TransportMessage(timestamp=int(time.time()), payload=f"{data.topic} does not exist.")
254
255        await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
256        logging.info("%s - %s", self._sid_ip_mapping[sid], response.payload)
257
258    async def handle_list_topics(self, sid, data=None) -> None:
259        """Called when a client requests a list of all topics.
260        The client will receive a list of all topics.
261
262        :param sid: Generated session id
263        :param data: Data sent by the client. Unused
264        """
265        response_msg = "All topics on the server:"
266        for topic in self._list_of_topics:
267            response_msg += f"\n{topic.name}"
268
269        response = TransportMessage(timestamp=int(time.time()), payload=response_msg)
270        await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
271        logging.info("%s - %s", self._sid_ip_mapping[sid], response.payload)
272
273    @_check_data_none_decorator
274    @_check_topic_decorator
275    async def handle_topic_status(self, sid, data=None) -> None:
276        """Called when a client requests the status of a topic.
277        The client will receive the status of the topic.
278
279        :param sid: Generated session id
280        :param data: Data sent by the client
281        """
282
283        data = TransportMessage.parse_raw(data)
284        topic = self._get_topic_by_name(data.topic)
285
286        if topic is not None:
287            subscribers = ""
288            for subscriber in topic.subscribers:
289                subscribers += f"\t{self._sid_ip_mapping[subscriber]}\n\t"
290
291            if topic.content is None or topic.timestamp is None:
292                topic_status = (
293                    f"\ntopic name:\t{topic.name}\n\nsubscribers:{subscribers}\nThere was no publish on this topic yet."
294                )
295            else:
296                topic_status = f"\ntopic name:\t{topic.name}\n\ntimestamp:\t{datetime.fromtimestamp(int(topic.timestamp)).strftime('%d-%m-%Y %H:%M:%S')}\n\ncontent:\t{topic.content}\n\nsubscribers:{subscribers}"
297
298            response = TransportMessage(timestamp=int(time.time()), payload=topic_status)
299        else:
300            # Topic not existing
301            response = TransportMessage(timestamp=int(time.time()), payload=f"{data.topic} does not exist.")
302
303        await self.sio.emit("PRINT_MESSAGE_AND_EXIT", response.json(), room=sid)
304        logging.info("%s - %s", self._sid_ip_mapping[sid], response.payload)
305
306    async def update_topic(self, topic: Topic) -> None:
307        """Called when a topic is updated.
308        The subscribers of the topic will receive the updated topic.
309
310        :param topic: The topic
311        """
312        topic.last_update = int(time.time())
313        response = TransportMessage(
314            timestamp=int(time.time()),
315            payload=f"{topic.name} ({datetime.fromtimestamp(int(topic.timestamp)).strftime('%d-%m-%Y %H:%M:%S')}): {topic.content}",
316        )
317        for sub in topic.subscribers:
318            await self.sio.emit("PRINT_MESSAGE", response.json(), room=sub)
319
320    async def heart_beat(self, time_delta):
321        """Go through all topics and check if they were updated in the last time_delta seconds.
322        If not, update the topic.
323
324        :param time_delta: Time in seconds
325        """
326        for topic in self._list_of_topics:
327            if topic.last_update is not None and int(time.time()) - topic.last_update > time_delta:
328                await self.update_topic(topic)
329                logging.info("Topic %s was updated through heart beat.", topic.name)
330
331    def _get_topic_by_name(self, name: str) -> Optional[Topic]:
332        """Get a topic by its name.
333
334        :param name: Name of the topic
335        :return: Topic object
336        """
337        for topic in self._list_of_topics:
338            if topic.name == name:
339                return topic
340        return None
341
342    def _add_topic(self, topic: Topic) -> None:
343        """Add a topic to the list of topics.
344
345        :param topic: Topic object
346        """
347        with self._lock:
348            self._list_of_topics.append(topic)
349
350    def _remove_topic(self, topic: Topic) -> None:
351        """Remove a topic from the list of topics.
352
353        :param topic: Topic object
354        """
355        with self._lock:
356            logging.warning("Topic %s was removed.", topic.name)
357            self._list_of_topics.remove(topic)
358
359
360def get_app():
361    """Create an ASGI application for the server.
362
363    :return: ASGI application
364    """
365    server = Server()
366    application = web.Application(logger=None)
367
368    server.sio.attach(application)
369
370    timer = ParallelTimer(server)
371    timer.start()
372
373    return application
374
375
376if __name__ == "__main__":
377    parser = ArgumentParser(prog="server.py", description="Starts a server for publisher subscriber system")
378    parser.add_argument(
379        "-p", "--port", type=str, help="Port to run the server on. Default is 8080", default=8080, metavar="STRING"
380    )
381    parser.add_argument(
382        "--host", type=str, help="Host to run the server on. Default is localhost", default="127.0.0.1", metavar="STRING"
383    )
384    params = parser.parse_args()
385
386    # wrap with ASGI application
387    app = get_app()
388    web.run_app(app, host=params.host, port=params.port)