diff --git a/playground/socketio/from_kafka.py b/playground/socketio/from_kafka.py index eefda4a..4a3c734 100644 --- a/playground/socketio/from_kafka.py +++ b/playground/socketio/from_kafka.py @@ -10,6 +10,8 @@ import asyncio import datetime import requests +tg_alert = False + # Create a Socket.IO server instance sio = socketio.AsyncServer(cors_allowed_origins = "*") @@ -21,16 +23,27 @@ app = web.Application() # Attach the Socket.IO server to the aiohttp application sio.attach(app) +from utils_v2.system import files from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context import os # Define the test params: TOPIC = "kft_file_upload" BOOTSTRAP_SERVERS = "del.ditscentre.in:9092" +# SSL_CONTEXT = get_ssl_context( +# ca_file = "../../creds/kafka/cert_authority.pem", +# cert_file = "../../creds/kafka/fullchain.pem", +# key_file = "../../creds/kafka/privkey.pem" +# ) + +cwd = files.get_cwd() +parent_dir = files.get_parent_directory(cwd, 2) +print("PD:", parent_dir) + SSL_CONTEXT = get_ssl_context( - ca_file = "../../creds/kafka/cert_authority.pem", - cert_file = "../../creds/kafka/fullchain.pem", - key_file = "../../creds/kafka/privkey.pem" + ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"), + cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"), + key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem") ) my_consumer = ConsumerKafka( topic = TOPIC, @@ -44,32 +57,34 @@ my_consumer = ConsumerKafka( @sio.event async def connect(sid, environ): print(f"Client {sid} connected") - requests.post( - url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", - json = { - "type": "info", - "chatClient": "telegram", - "chatId": "-4206946032", - # "chatId": "1275560043", - "message": f"*SocketIO Connected!*\nšŸ‘ SID: {sid}" - } - ) + if tg_alert: + requests.post( + url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", + json = { + "type": "info", + "chatClient": "telegram", + "chatId": "-4206946032", + # "chatId": "1275560043", + "message": f"*SocketIO Connected!*\nšŸ‘ SID: {sid}" + } + ) # Event: Client disconnects @sio.event async def disconnect(sid): print(f"Client {sid} disconnected") - requests.post( - url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", - json = { - "type": "info", - "chatClient": "telegram", - "chatId": "-4206946032", - # "chatId": "1275560043", - "message": f"*SocketIO Disconnected!*\nāŒ SID: {sid}" - } - ) + if tg_alert: + requests.post( + url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", + json = { + "type": "info", + "chatClient": "telegram", + "chatId": "-4206946032", + # "chatId": "1275560043", + "message": f"*SocketIO Disconnected!*\nāŒ SID: {sid}" + } + ) @sio.event