René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Subscribing to Quote data overwhelms single async io call in websocket. Here is some fixes that could work.","articleBody":"cmsg = await asyncio.wait_for(s.recv(), timeout=1)\nwhich means that messages are received one at a time. Here’s what happens in practice if many messages arrive:\n\nMessage Queuing:\nThe underlying websockets library (and the TCP connection it uses) will buffer incoming messages. So, if messages arrive in rapid succession, they will be queued up internally.\n\nSequential Processing:\nOnce a message is received, the code processes it by optionally decoding and then passing it to the processor callback \n\nHere is updated code,  \nIn this revision, we add an internal asyncio.Queue and a pool of consumer tasks that call your message processor. This decouples receiving messages from processing them, so bursts of incoming messages are queued (with optional back‐pressure via a maximum queue size) and processed concurrently (up to a fixed number of tasks).\n\n```\nimport os\nfrom enum import Enum\nfrom typing import Optional, Union, List, Set, Callable, Awaitable, Any\nimport logging\nimport json\nimport asyncio\nimport ssl\nimport certifi\nfrom .models import *\nfrom websockets.client import connect, WebSocketClientProtocol\nfrom websockets.exceptions import ConnectionClosedOK, ConnectionClosedError\nfrom ..logging import get_logger\nimport logging\nfrom ..exceptions import AuthError\n\nenv_key = \"POLYGON_API_KEY\"\nlogger = get_logger(\"WebSocketClient\")\n\n\nclass WebSocketClient:\n    def __init__(\n        self,\n        api_key: Optional[str] = os.getenv(env_key),\n        feed: Union[str, Feed] = Feed.RealTime,\n        market: Union[str, Market] = Market.Stocks,\n        raw: bool = False,\n        verbose: bool = False,\n        subscriptions: Optional[List[str]] = None,\n        max_reconnects: Optional[int] = 5,\n        secure: bool = True,\n        custom_json: Optional[Any] = None,\n        **kwargs,\n    ):\n        \"\"\"\n        Initialize a Polygon WebSocketClient.\n        \"\"\"\n        if api_key is None:\n            raise AuthError(\n                f\"Must specify env var {env_key} or pass api_key in constructor\"\n            )\n        self.api_key = api_key\n        self.feed = feed\n        self.market = market\n        self.raw = raw\n        if verbose:\n            logger.setLevel(logging.DEBUG)\n        self.websocket_cfg = kwargs\n        if isinstance(feed, Enum):\n            feed = feed.value\n        if isinstance(market, Enum):\n            market = market.value\n        self.url = f\"ws{'s' if secure else ''}://{feed}/{market}\"\n        self.subscribed = False\n        self.subs: Set[str] = set()\n        self.max_reconnects = max_reconnects\n        self.websocket: Optional[WebSocketClientProtocol] = None\n        if subscriptions is None:\n            subscriptions = []\n        self.scheduled_subs: Set[str] = set(subscriptions)\n        self.schedule_resub = True\n        if custom_json:\n            self.json = custom_json\n        else:\n            self.json = json\n\n    async def _message_consumer(\n        self,\n        processor: Union[\n            Callable[[List[WebSocketMessage]], Awaitable],\n            Callable[[Union[str, bytes]], Awaitable],\n        ],\n        message_queue: asyncio.Queue,\n    ):\n        \"\"\"\n        A worker that continuously gets messages from the queue and processes them.\n        \"\"\"\n        while True:\n            msg = await message_queue.get()\n            try:\n                await processor(msg)\n            except Exception as e:\n                logger.exception(\"Error processing message: %s\", e)\n            finally:\n                message_queue.task_done()\n\n    # Updated connect method using a message queue and consumer pool.\n    async def connect(\n        self,\n        processor: Union[\n            Callable[[List[WebSocketMessage]], Awaitable],\n            Callable[[Union[str, bytes]], Awaitable],\n        ],\n        close_timeout: int = 1,\n        concurrency: int = 10,\n        queue_size: int = 1000,\n        **kwargs,\n    ):\n        \"\"\"\n        Connect to the websocket server and use a pool of tasks to process messages concurrently.\n        :param processor: Callback to process each message.\n        :param close_timeout: How long to wait for handshake when calling .close.\n        :param concurrency: Number of concurrent consumer tasks to process messages.\n        :param queue_size: Maximum number of messages that can be waiting for processing.\n        :raises AuthError: If invalid API key is supplied.\n        \"\"\"\n        reconnects = 0\n        logger.debug(\"connect: %s\", self.url)\n        ssl_context = None\n        if self.url.startswith(\"wss://\"):\n            ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)\n            ssl_context.load_verify_locations(certifi.where())\n\n        async for s in connect(self.url, close_timeout=close_timeout, ssl=ssl_context, **kwargs):\n            self.websocket = s\n            try:\n                # Initial handshake and authentication.\n                msg = await s.recv()\n                logger.debug(\"connected: %s\", msg)\n                logger.debug(\"authing...\")\n                await s.send(self.json.dumps({\"action\": \"auth\", \"params\": self.api_key}))\n                auth_msg = await s.recv()\n                auth_msg_parsed = self.json.loads(auth_msg)\n                logger.debug(\"authed: %s\", auth_msg)\n                if auth_msg_parsed[0][\"status\"] == \"auth_failed\":\n                    raise AuthError(auth_msg_parsed[0][\"message\"])\n\n                # Create an asyncio queue for incoming messages.\n                message_queue = asyncio.Queue(maxsize=queue_size)\n                # Create a pool of consumer tasks.\n                consumer_tasks = [\n                    asyncio.create_task(self._message_consumer(processor, message_queue))\n                    for _ in range(concurrency)\n                ]\n\n                # Main loop: receive messages and put them on the queue.\n                while True:\n                    if self.schedule_resub:\n                        logger.debug(\"Reconciling subscriptions: current: %s, scheduled: %s\", self.subs, self.scheduled_subs)\n                        new_subs = self.scheduled_subs.difference(self.subs)\n                        await self._subscribe(new_subs)\n                        old_subs = self.subs.difference(self.scheduled_subs)\n                        await self._unsubscribe(old_subs)\n                        self.subs = set(self.scheduled_subs)\n                        self.schedule_resub = False\n\n                    try:\n                        raw_msg = await asyncio.wait_for(s.recv(), timeout=1)\n                    except asyncio.TimeoutError:\n                        continue\n\n                    # Process the raw message.\n                    if not self.raw:\n                        msgJson = self.json.loads(raw_msg)\n                        filtered_msgs = []\n                        for m in msgJson:\n                            if m.get(\"ev\") == \"status\":\n                                logger.debug(\"status: %s\", m.get(\"message\"))\n                            else:\n                                filtered_msgs.append(m)\n                        if not filtered_msgs:\n                            continue\n                        processed_msg = parse(filtered_msgs, logger)\n                    else:\n                        processed_msg = raw_msg\n\n                    # Enqueue the processed message.\n                    await message_queue.put(processed_msg)\n\n                # (If the loop ever ends, cancel consumer tasks.)\n                for task in consumer_tasks:\n                    task.cancel()\n                await asyncio.gather(*consumer_tasks, return_exceptions=True)\n\n            except ConnectionClosedOK as e:\n                logger.debug(\"connection closed (OK): %s\", e)\n                return\n            except ConnectionClosedError as e:\n                logger.debug(\"connection closed (ERR): %s\", e)\n                reconnects += 1\n                # Save subscriptions for reconnection.\n                self.scheduled_subs = set(self.subs)\n                self.subs = set()\n                self.schedule_resub = True\n                if self.max_reconnects is not None and reconnects \u003e self.max_reconnects:\n                    return\n                continue\n\n    def run(\n        self,\n        handle_msg: Union[\n            Callable[[List[WebSocketMessage]], None],\n            Callable[[Union[str, bytes]], None],\n        ],\n        close_timeout: int = 1,\n        **kwargs,\n    ):\n        \"\"\"\n        Synchronous version of .connect.\n        \"\"\"\n        async def handle_msg_wrapper(msgs):\n            handle_msg(msgs)\n\n        asyncio.run(self.connect(handle_msg_wrapper, close_timeout, **kwargs))\n\n    async def _subscribe(self, topics: Union[List[str], Set[str]]):\n        if self.websocket is None or len(topics) == 0:\n            return\n        subs = \",\".join(topics)\n        logger.debug(\"subscribing: %s\", subs)\n        await self.websocket.send(self.json.dumps({\"action\": \"subscribe\", \"params\": subs}))\n\n    async def _unsubscribe(self, topics: Union[List[str], Set[str]]):\n        if self.websocket is None or len(topics) == 0:\n            return\n        subs = \",\".join(topics)\n        logger.debug(\"unsubscribing: %s\", subs)\n        await self.websocket.send(self.json.dumps({\"action\": \"unsubscribe\", \"params\": subs}))\n\n    @staticmethod\n    def _parse_subscription(s: str):\n        s = s.strip()\n        split = s.split(\".\", 1)  # Split at the first period.\n        if len(split) != 2:\n            logger.warning(\"invalid subscription: %s\", s)\n            return [None, None]\n        return split\n\n    def subscribe(self, *subscriptions: str):\n        \"\"\"\n        Subscribe to given subscriptions.\n        \"\"\"\n        for s in subscriptions:\n            topic, sym = self._parse_subscription(s)\n            if topic is None:\n                continue\n            logger.debug(\"Desired subscription: %s\", s)\n            self.scheduled_subs.add(s)\n            # If subscribing to X.*, remove other X.\u003csomething\u003e subscriptions.\n            if sym == \"*\":\n                for t in list(self.subs):\n                    if t.startswith(topic):\n                        self.scheduled_subs.discard(t)\n        self.schedule_resub = True\n\n    def unsubscribe(self, *subscriptions: str):\n        \"\"\"\n        Unsubscribe from given subscriptions.\n        \"\"\"\n        for s in subscriptions:\n            topic, sym = self._parse_subscription(s)\n            if topic is None:\n                continue\n            logger.debug(\"Unsubscribe request: %s\", s)\n            self.scheduled_subs.discard(s)\n            # If unsubscribing from X.*, remove other X.\u003csomething\u003e subscriptions.\n            if sym == \"*\":\n                for t in list(self.subs):\n                    if t.startswith(topic):\n                        self.scheduled_subs.discard(t)\n        self.schedule_resub = True\n\n    def unsubscribe_all(self):\n        \"\"\"\n        Unsubscribe from all subscriptions.\n        \"\"\"\n        self.scheduled_subs = set()\n        self.schedule_resub = True\n\n    async def close(self):\n        \"\"\"\n        Close the websocket connection.\n        \"\"\"\n        logger.debug(\"closing connection\")\n        if self.websocket:\n            await self.websocket.close()\n            self.websocket = None\n        else:\n            logger.warning(\"no websocket open to close\")\n```","author":{"url":"https://github.com/Vetti420","@type":"Person","name":"Vetti420"},"datePublished":"2025-02-11T21:59:48.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":2},"url":"https://github.com/849/client-python/issues/849"}