From 6417a72e0137204dbc366aa2f877db0e3233a939 Mon Sep 17 00:00:00 2001
From: KKlochko <KKlochko@protonmail.com>
Date: Tue, 12 Dec 2023 15:27:23 +0200
Subject: [PATCH] Add the server and the entry point.

---
 notification_server/main.py   |  7 +++
 notification_server/server.py | 84 +++++++++++++++++++++++++++++++++++
 2 files changed, 91 insertions(+)
 create mode 100644 notification_server/main.py
 create mode 100644 notification_server/server.py

diff --git a/notification_server/main.py b/notification_server/main.py
new file mode 100644
index 0000000..d8d2645
--- /dev/null
+++ b/notification_server/main.py
@@ -0,0 +1,7 @@
+import asyncio
+from .server import Server
+
+if __name__ == '__main__':
+    server = Server('127.0.0.1', 5555, 5554)
+    asyncio.run(server.up())
+
diff --git a/notification_server/server.py b/notification_server/server.py
new file mode 100644
index 0000000..40516db
--- /dev/null
+++ b/notification_server/server.py
@@ -0,0 +1,84 @@
+import struct
+from rich.console import Console
+import asyncio
+
+
+class Server:
+    BUFFER_SIZE = 1024
+    NOTIFICATION_DIR = 'notifications'
+    console = Console()
+
+    def __init__(self, ip: str, port: int, producer_port: int, notification_dir: str = 'notifications'):
+        self.NOTIFICATION_DIR = notification_dir
+        self.__client_socket = None
+        self.__producer_socket = None
+        self.__ip = ip
+        self.__port = port
+        self.__producer_port = producer_port
+        self.__client_writers = []
+
+    async def up(self):
+        self.__client_socket = await asyncio.start_server(
+            self.handle_client, self.__ip, self.__port
+        )
+
+        self.__producer_socket = await asyncio.start_server(
+            lambda reader, writer: self.handle_producer(reader, writer, self.NOTIFICATION_DIR),
+            self.__ip, self.__producer_port
+        )
+
+        self.console.print(f'Serves on {self.__ip}:{self.__port} for clients.')
+        self.console.print(f'Receives notifications from producers at {self.__ip}:{self.__producer_port}.')
+
+        async with self.__producer_socket, self.__client_socket:
+            await asyncio.gather(
+                self.__producer_socket.serve_forever(),
+                self.__client_socket.serve_forever()
+            )
+
+    @staticmethod
+    async def send_message(writer, message: str):
+        writer.write(struct.pack('<L', len(message)))
+        writer.write(message.encode('utf-8'))
+
+    @staticmethod
+    async def receive_message(reader) -> str:
+        size, = struct.unpack('<L', await reader.readexactly(4))
+        message = await reader.readexactly(size)
+        return message.decode('utf-8')
+
+    async def broadcast_message(self, message):
+        for writer in self.__client_writers:
+            await self.send_message(writer, message)
+
+    async def handle_producer(self, reader, writer, notification_dir):
+        addr = writer.get_extra_info('peername')
+        self.console.print(f"A producer ({addr}) connected.")
+
+        try:
+            while True:
+                message = await self.receive_message(reader)
+                print(f"{message=}")
+                await self.broadcast_message(message)
+
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self.console.print(f"The producer ({addr}) disconnected.")
+            writer.close()
+
+    async def handle_client(self, reader, writer):
+        addr = writer.get_extra_info('peername')
+        self.console.print(f"A client ({addr}) connected.")
+        self.__client_writers.append(writer)
+
+        try:
+            while True:
+                pass
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self.console.print(f"The client ({addr}) disconnected.")
+            self.__client_writers.remove(writer)
+            writer.close()
+