cloud light
best way to summarize this is I'm trying to find out what is causation of this error with socket.io "Failed to connect to Stream labs WebSocket: One or more namespaces failed to connect" i just want to know what am i doing wrong.
import os
import socketio
import asyncio
from dotenv import load_dotenv
load_dotenv()
class StreamlabsConnection:
def __init__(self):
self.token = os.getenv("STREAMLABS_SOCKET")
self.sio = socketio.AsyncClient()
self.queue = asyncio.Queue()
async def connect_stream(self):
@self.sio.on('connect', namespace='/')
async def connect():
print("✅ Connected to Streamlabs")
await self.sio.emit('authenticate', {"token": self.token}, namespace='/')
@self.sio.on("event", namespace='/')
async def on_event(data):
await self.queue.put(data)
try:
await self.sio.connect("https://sockets.streamlabs.com/socket.io/", transports=["websocket"], namespaces=["/"])
except Exception as e:
print(f"❌ Failed to connect to Streamlabs WebSocket: {e}")
return
# This keeps the connection alive and continuously yields messages
while True:
message = await self.queue.get()
yield self.parse_message(message)
def parse_message(self, raw_data):
event_type = raw_data.get('type', '')
msg_data = raw_data.get('message', {})
if isinstance(msg_data, list):
msg_data = msg_data[0] if msg_data else {}
return {
"type": event_type,
"username": msg_data.get('name') or msg_data.get('from'),
"message": msg_data.get('message') or msg_data.get('body'),
}```.
crisp valve
iron wadi
