[Bug] WebSocket Streams: stream registry is global, two clients with the same stream name get each other's data #570

import asyncio, json, logging
from aiohttp import web
from binance_common.configuration import ConfigurationWebSocketStreams
from binance_common.websocket import WebSocketStreamBase, RequestStream
logging.disable(logging.CRITICAL)
subs = {"spot": [], "futures": []}
def mk(name):
    async def handler(request):
        ws = web.WebSocketResponse(); await ws.prepare(request)
        async for msg in ws:
            d = json.loads(msg.data)
            if d["method"] == "SUBSCRIBE":
                subs[name] += d["params"]
                await ws.send_json({"id": d["id"], "result": None})
                for s in d["params"]:
                    await ws.send_json({"stream": s, "data": {"venue": name}})
        return ws
    return handler
async def main():
    app = web.Application()
    app.router.add_get("/spot/stream", mk("spot")); app.router.add_get("/fut/stream", mk("futures"))
    r = web.AppRunner(app); await r.setup(); await web.TCPSite(r, "127.0.0.1", 18766).start()
    spot = WebSocketStreamBase(ConfigurationWebSocketStreams(stream_url="ws://127.0.0.1:18766/spot/stream"))
    fut = WebSocketStreamBase(ConfigurationWebSocketStreams(stream_url="ws://127.0.0.1:18766/fut/stream"))
    await spot.create_connection(); await fut.create_connection()
    got = {"spot_cb": [], "fut_cb": []}
    h1 = await RequestStream(spot, "btcusdt@aggTrade"); h1.on("message", lambda d: got["spot_cb"].append(d))
    h2 = await RequestStream(fut, "btcusdt@aggTrade");  h2.on("message", lambda d: got["fut_cb"].append(d))
    await asyncio.sleep(0.5)
    print("SUBSCRIBE frames received by servers:", subs)
    print("callback payloads:", got)
    await spot.close_connection(); await fut.close_connection(); await r.cleanup()
asyncio.run(main())