khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241224) File location test

Browse Source
This commit is contained in:
khushal
2024-12-24 14:29:35 +05:30
parent f58fbb0131
commit 9867582844
1 changed files with 38 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
playground/socketio/from_kafka.py
+38 -23
View File
@@ -10,6 +10,8 @@ import asyncio
import datetime import datetime
import requests import requests
tg_alert = False
# Create a Socket.IO server instance # Create a Socket.IO server instance
sio = socketio.AsyncServer(cors_allowed_origins = "*") sio = socketio.AsyncServer(cors_allowed_origins = "*")
@@ -21,16 +23,27 @@ app = web.Application()
# Attach the Socket.IO server to the aiohttp application # Attach the Socket.IO server to the aiohttp application
sio.attach(app) sio.attach(app)
from utils_v2.system import files
from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context
import os import os
# Define the test params: # Define the test params:
TOPIC = "kft_file_upload" TOPIC = "kft_file_upload"
BOOTSTRAP_SERVERS = "del.ditscentre.in:9092" 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( SSL_CONTEXT = get_ssl_context(
ca_file = "../../creds/kafka/cert_authority.pem", ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
cert_file = "../../creds/kafka/fullchain.pem", cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
key_file = "../../creds/kafka/privkey.pem" key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
) )
my_consumer = ConsumerKafka( my_consumer = ConsumerKafka(
topic = TOPIC, topic = TOPIC,
@@ -44,32 +57,34 @@ my_consumer = ConsumerKafka(
@sio.event @sio.event
async def connect(sid, environ): async def connect(sid, environ):
print(f"Client {sid} connected") print(f"Client {sid} connected")
requests.post( if tg_alert:
url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", requests.post(
json = { url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend",
"type": "info", json = {
"chatClient": "telegram", "type": "info",
"chatId": "-4206946032", "chatClient": "telegram",
# "chatId": "1275560043", "chatId": "-4206946032",
"message": f"*SocketIO Connected!*\n👍 SID: {sid}" # "chatId": "1275560043",
} "message": f"*SocketIO Connected!*\n👍 SID: {sid}"
) }
)
# Event: Client disconnects # Event: Client disconnects
@sio.event @sio.event
async def disconnect(sid): async def disconnect(sid):
print(f"Client {sid} disconnected") print(f"Client {sid} disconnected")
requests.post( if tg_alert:
url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", requests.post(
json = { url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend",
"type": "info", json = {
"chatClient": "telegram", "type": "info",
"chatId": "-4206946032", "chatClient": "telegram",
# "chatId": "1275560043", "chatId": "-4206946032",
"message": f"*SocketIO Disconnected!*\n❌ SID: {sid}" # "chatId": "1275560043",
} "message": f"*SocketIO Disconnected!*\n❌ SID: {sid}"
) }
)
@sio.event @sio.event