no more changes until next year

openbb_platform/providers/intrinio/openbb_intrinio/utils/websocket_client.py

@@ -99,12 +99,6 @@ async def read_stdin_and_queue_commands():
             break
 
         try:
-            if line == "qsize":
-                logger.info(
-                    "Queue size: %i - Writer Queue: %i",
-                    command_queue.queue.qsize(),
-                    DATABASE.batch_processor.task_queue.qsize(),
-                )
             command = json.loads(line.strip())
             await command_queue.enqueue(command)
         except json.JSONDecodeError:
@@ -115,6 +109,13 @@ async def process_stdin_queue():
     """Process the command queue."""
     while True:
         command = await command_queue.dequeue()
+        if command == "qsize":
+            logger.info(
+                "Queue size: %i - Writer Queue: %i",
+                command_queue.queue.qsize(),
+                DATABASE.batch_processor.task_queue.qsize(),
+            )
+
         symbol = ["lobby" if d == "*" else d.upper() for d in command.get("symbol", [])]
         event = command.get("event")
         if symbol and event:

openbb_platform/providers/polygon/openbb_polygon/utils/websocket_client.py

@@ -324,9 +324,8 @@ async def connect_and_stream():
                 asyncio.run(process_message(message_data))
 
         # Run processing in thread
-        await asyncio.get_event_loop().run_in_executor(
-            message_thread_pool, _process_in_thread
-        )
+        process = asyncio.to_thread(_process_in_thread)
+        asyncio.create_task(process)
 
     try:
         handler_task = asyncio.create_task(
@@ -338,14 +337,13 @@ async def connect_and_stream():
 
         await DATABASE.start_writer()
 
-        processor_task = asyncio.create_task(
-            input_queue.process_queue(lambda message: process_input_messages(message))
-        )
-        tasks.add(processor_task)
-        handler_task = asyncio.create_task(
-            process_queue.process_queue(lambda message: process_message(message))
-        )
-        tasks.add(handler_task)
+        for i in range(16):
+            processor_task = asyncio.create_task(
+                input_queue.process_queue(
+                    lambda message: process_input_messages(message)
+                )
+            )
+            tasks.add(processor_task)
 
         async for websocket in connect(URL, **conn_kwargs):
             try:
@@ -360,20 +358,13 @@ async def connect_and_stream():
                 await login(websocket)
 
                 response = await websocket.recv()
-                messages = json.loads(response)
+                messages = orjson.loads(response)
 
                 await process_message(messages)
 
                 await subscribe(websocket, kwargs["symbol"], "subscribe")
 
-                if not any(
-                    task.name == "receiver" for task in tasks if hasattr(task, "name")
-                ):
-                    receiver_task = asyncio.create_task(
-                        message_receiver(websocket), name="receiver"
-                    )
-                    tasks.add(receiver_task)
-                    await asyncio.gather(receiver_task)
+                await message_receiver(websocket)
 
             # Attempt to reopen the connection
             except (