137 lines
3.6 KiB
Python
137 lines
3.6 KiB
Python
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
|
|
import random
|
|
import time
|
|
import socketio
|
|
import asyncio
|
|
import datetime
|
|
import requests
|
|
|
|
tg_alert = False
|
|
|
|
|
|
# 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.system import files
|
|
from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context
|
|
import os
|
|
|
|
# Define the test params:
|
|
TOPIC = "tickers"
|
|
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)
|
|
parent_dir = cwd
|
|
print("CWD:", cwd)
|
|
# print("PD:", parent_dir)
|
|
SSL_CONTEXT = get_ssl_context(
|
|
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,
|
|
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")
|
|
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")
|
|
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
|
|
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())
|