123COMMENT

wsio_v2/test/echo.py

@@ -0,0 +1,137 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Tuesday, 30th Sept., 2025
+
+    OBJECTIVE:
+
+        A simple namespace to test the Socket.IO app with a simple echo utility.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append("../wsio")
+sys.path.append("..")
+
+# System-level activities:
+import io
+import os
+import random
+
+# My utils:
+from utils_v2.string import json
+from utils_v2.string import regex
+from utils_v2.system import files
+from utils_v2.date_time import date_time
+from utils_v2.database.async_mongo_v2 import AsyncMongo
+from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
+from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
+from utils_v2.serialization.json_serializer import JSONSerializer
+
+# To make HTTP calls:
+import httpx
+
+# To work with date and time:
+import datetime
+import time
+
+# Models:
+from models.core.user import CoreUserInfoModel
+from models.finstitutions.trading.ticks import TradingTick
+
+# To work with SocketIO:
+import socket
+import socketio
+
+# To maintain the app's state:
+from wsio_v2.app_state import AppState
+
+# For asynchronous activities:
+import asyncio
+
+# To work with various datatypes:
+from typing import List
+
+# Debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class TestNamespace(socketio.AsyncNamespace):
+
+    # Internal variables:
+    app_state: AppState = None
+
+    def __init__(self, namespace):
+        super().__init__(namespace)
+        print("Test Namespace Init")
+
+    async def on_connect(self, sid, environ, *args):
+        print(f"ON CONNECT TEST — SID: {sid}")
+
+    async def on_disconnect(self, sid, reason, *args):
+        print(f"ON DISCONNECT TEST — SID: {sid}")
+
+    async def on_echo(self, sid, data):
+        print(f"Received echo from {sid}: {data}")
+        await self.emit("echo", data, to=sid)
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass