104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
"""
|
|
|
|
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(".")
|
|
sys.path.append("..")
|
|
|
|
# To work with SocketIO:
|
|
import socketio
|
|
|
|
# To maintain the app's state:
|
|
from wsio_v2.app_state import AppState
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class EchoNamespace(socketio.AsyncNamespace):
|
|
|
|
def __init__(
|
|
self,
|
|
namespace: str,
|
|
app_state: AppState,
|
|
):
|
|
super().__init__(namespace)
|
|
self.app_state = app_state
|
|
self.app_state.printer("Registered!")
|
|
|
|
async def on_connect(self, sid, environ, *args):
|
|
self.app_state.printer("On Connect", sid)
|
|
|
|
async def on_disconnect(self, sid, reason, *args):
|
|
self.app_state.printer("On Disconnect", sid)
|
|
|
|
async def on_echo(self, sid, data):
|
|
self.app_state.printer("Event", sid, data, type(data).__name__)
|
|
await self.emit("echo", data, to = sid)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|