mirror of
https://github.com/ryanamay/sparkle.git
synced 2024-09-20 00:40:35 +00:00
feat: cross socket forwarding
This commit is contained in:
parent
08bb926220
commit
039fe9dcba
2 changed files with 34 additions and 15 deletions
|
@ -7,11 +7,14 @@ from app.core.config import settings
|
|||
from typing import List
|
||||
|
||||
connected_clients: List[WebSocket] = []
|
||||
finnhub_ws: websocket.WebSocketApp = None
|
||||
|
||||
# WebSocket event handlers
|
||||
async def on_message(ws, message):
|
||||
print("Client list:", connected_clients)
|
||||
print("Received message from Finnhub WebSocket:", message)
|
||||
print("Forwarding message to connected clients:", connected_clients)
|
||||
for client in connected_clients:
|
||||
try:
|
||||
try:
|
||||
await client.send_text(message)
|
||||
except Exception as e:
|
||||
print("Error sending message to client:", e)
|
||||
|
@ -20,22 +23,34 @@ def on_error(ws, error):
|
|||
print("Error:", error)
|
||||
|
||||
def on_close(ws):
|
||||
print("### WebSocket closed ###")
|
||||
print("### Finnhub WebSocket closed ###")
|
||||
|
||||
def on_open(ws):
|
||||
global finnhub_ws
|
||||
finnhub_ws = ws
|
||||
ws.send(json.dumps({'type': 'subscribe', 'symbol': 'AAPL'}))
|
||||
ws.send(json.dumps({'type': 'subscribe', 'symbol': 'AMZN'}))
|
||||
ws.send(json.dumps({'type': 'subscribe', 'symbol': 'BINANCE:BTCUSDT'}))
|
||||
ws.send(json.dumps({'type': 'subscribe', 'symbol': 'IC MARKETS:1'}))
|
||||
|
||||
def forward_message_to_finnhub(message: str):
|
||||
global finnhub_ws
|
||||
if finnhub_ws:
|
||||
finnhub_ws.send(message)
|
||||
print("Forwarded message to Finnhub:", message)
|
||||
else:
|
||||
print("Finnhub WebSocket is not connected.")
|
||||
|
||||
def start_finnhub_websocket():
|
||||
websocket.enableTrace(True)
|
||||
ws = websocket.WebSocketApp(f"wss://ws.finnhub.io?token={settings.FINNHUB_API_KEY}",
|
||||
on_message=lambda ws, msg: asyncio.run(on_message(ws, msg)),
|
||||
on_error=on_error,
|
||||
on_close=on_close)
|
||||
ws = websocket.WebSocketApp(
|
||||
f"wss://ws.finnhub.io?token={settings.FINNHUB_API_KEY}",
|
||||
on_message=lambda ws, msg: asyncio.run(on_message(ws, msg)),
|
||||
on_error=on_error,
|
||||
on_close=on_close
|
||||
)
|
||||
ws.on_open = on_open
|
||||
|
||||
# Running WebSocket in a separate thread
|
||||
# Run WebSocket in a separate thread
|
||||
thread = threading.Thread(target=ws.run_forever)
|
||||
thread.start()
|
|
@ -1,25 +1,29 @@
|
|||
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
|
||||
from app.core.websocket import connected_clients, start_finnhub_websocket
|
||||
from app.core.websocket import connected_clients, start_finnhub_websocket, forward_message_to_finnhub
|
||||
from typing import List
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.websocket("/trades")
|
||||
async def websocket_endpoint(websocket: WebSocket):
|
||||
await websocket.accept()
|
||||
connected_clients.append(websocket)
|
||||
connected_clients.append(websocket)
|
||||
print("Client connected:", websocket)
|
||||
|
||||
try:
|
||||
while True:
|
||||
data = await websocket.receive_text()
|
||||
print(f"Received from client: {data}")
|
||||
print("Received message from client:", data)
|
||||
# Forwarding message from client to Finnhub
|
||||
forward_message_to_finnhub(data)
|
||||
except WebSocketDisconnect:
|
||||
connected_clients.remove(websocket)
|
||||
print("Client disconnected")
|
||||
connected_clients.remove(websocket)
|
||||
print("Client disconnected:", websocket)
|
||||
|
||||
@router.get("/start-websocket")
|
||||
async def start_websocket():
|
||||
"""
|
||||
API to trigger the Finnhub WebSocket connection.
|
||||
"""
|
||||
start_finnhub_websocket() #
|
||||
return {"message": "WebSocket connection started"}
|
||||
start_finnhub_websocket()
|
||||
return {"message": "WebSocket connection to Finnhub started"}
|
Loading…
Reference in a new issue