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

(20241224) Live Feed From Kafka (to test).

Browse Source
This commit is contained in:
khushal
2024-12-24 14:06:05 +05:30
parent 7c7b10f27c
commit ae6a959a21
18 changed files with 1018 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
playground/socketio/from_kafka.py
+112
View File
@@ -0,0 +1,112 @@
import random
import time
import socketio
import asyncio
import datetime
import requests
# Create a Socket.IO server instance
sio = socketio.AsyncServer(cors_allowed_origins = "*")
# Create an aiohttp web application
from aiohttp import web
app = web.Application()
# Attach the Socket.IO server to the aiohttp application
sio.attach(app)
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"
)
my_consumer = ConsumerKafka(
topic = TOPIC,
bootstrap_servers = BOOTSTRAP_SERVERS,
security_protocol = "SSL",
ssl_context = SSL_CONTEXT
)
# Event: Client connects
@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}"
}
)
# 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}"
}
)
@sio.event
async def message(sid, data):
print("MESSAGE:", data)
# Function to generate random data
async def broadcast_one_tick(tick):
await sio.emit("ticks", tick)
# Function to broadcast data every second asynchronously
async def broadcast_ticks():
while True:
messages = await my_consumer.consume(count = 100, timeout = 1.0)
print(f"Received {len(messages)} tick(s)")
tasks = [broadcast_one_tick(m["value"]) for m in messages]
if tasks: results = await asyncio.gather(*tasks)
# Start broadcasting random data using asyncio
async def start_broadcast():
await broadcast_ticks()
# Main function to run the aiohttp server and the broadcasting
async def main():
# Start broadcasting random data in the background
asyncio.create_task(start_broadcast())
# Run the web server
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 5214)
print("Server running on http://0.0.0.0:5214")
await site.start()
# Keep the server running
while True:
await asyncio.sleep(3600) # Keep the server alive for 1 hour or adjust as needed
# Run the main asyncio event loop
if __name__ == '__main__':
asyncio.run(main())
playground/socketio/tick_simulator.py
+44 -26
View File
@@ -134,7 +134,7 @@ SYMBOL_TO_PRICE_MAP = {
# *****************************************************************************************************************
# --- Nothing Yet
tg_update = False
# *****************************************************************************************************************
@@ -144,21 +144,39 @@ SYMBOL_TO_PRICE_MAP = {
# *****************************************************************************************************************
@sio.event
async def before_connect(sid, environ):
print(f"Checking connection attempt from {sid}.")
# # Simulate a failed connection based on some conditions (for example, invalid IP or header)
# user_agent = environ.get('HTTP_USER_AGENT', '')
# if 'BadUserAgent' in user_agent:
# print(f"Rejected connection from {sid} due to invalid User-Agent.")
# return False # This will reject the connection attempt
return True # Allow connection
# ---------------------------------------------------------------------------------------------------------------------
@sio.event
async def connect(sid, environ):
print(f"Client {sid} connected")
async with httpx.AsyncClient() as client:
try: await client.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}"
}
)
except: pass
if tg_update:
async with httpx.AsyncClient() as client:
try: await client.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}"
}
)
except: pass
# ---------------------------------------------------------------------------------------------------------------------
@@ -167,18 +185,19 @@ async def connect(sid, environ):
@sio.event
async def disconnect(sid):
print(f"Client {sid} disconnected")
async with httpx.AsyncClient() as client:
try: await client.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}"
}
)
except: pass
if tg_update:
async with httpx.AsyncClient() as client:
try: await client.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}"
}
)
except: pass
def round_tick(price):
@@ -339,7 +358,6 @@ if __name__ == "__main__":
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", 5214)
print("Server running on http://0.0.0.0:5000")
await site.start()
# Keep the server running