jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-31 14:31:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

handle huge polgon crypto message volume

Browse Source
This commit is contained in:
Danglewood
2024-12-21 14:28:50 -08:00
parent 696b1da8c2
commit 23013e6c6b
2 changed files with 14 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
openbb_platform/providers/polygon/openbb_polygon/utils/websocket_client.py
View File

@@ -156,11 +156,10 @@ async def read_stdin():
sys.stdin.flush()
if not line:
break
continue
if "qsize" in line:
logger.info(f"PROVIDER INFO: Queue size: {queue.queue.qsize()}")
else:
try:
command = json.loads(line.strip())
@@ -221,21 +220,20 @@ async def connect_and_stream():
)
tasks.add(handler_task)
for i in range(0, 48):
new_task = asyncio.create_task(
queue.process_queue(lambda message: process_message(message))
new_task = asyncio.shield(
asyncio.create_task(
queue.process_queue(lambda message: process_message(message))
)
)
tasks.add(new_task)
stdin_task = asyncio.create_task(read_stdin())
stdin_task = asyncio.shield(asyncio.create_task(read_stdin()))
try:
connect_kwargs = CONNECT_KWARGS.copy()
if "ping_timeout" not in connect_kwargs:
connect_kwargs["ping_timeout"] = None
if "close_timeout" not in connect_kwargs:
connect_kwargs["close_timeout"] = None
connect_kwargs["max_size"] = None
connect_kwargs["read_limit"] = 2**32
try:
async with websockets.connect(URL, **connect_kwargs) as websocket:
await login(websocket)
response = await websocket.recv()
messages = json.loads(response)
@@ -312,7 +310,7 @@ if __name__ == "__main__":
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_exception_handler(lambda loop, context: None)
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, handle_termination_signal, logger)

7
openbb_platform/providers/tiingo/openbb_tiingo/utils/websocket_client.py
View File

@@ -288,7 +288,7 @@ if __name__ == "__main__":
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_exception_handler(lambda loop, context: None)
loop.add_signal_handler(signal.SIGTERM, handle_termination_signal, logger)
asyncio.run_coroutine_threadsafe(
@@ -298,7 +298,10 @@ if __name__ == "__main__":
loop.run_forever()
except Exception as e: # pylint: disable=broad-except
ERR = f"Unexpected error -> {e.__class__.__name__}: {e}"
ERR = (
f"PROVIDER ERROR: Unexpected error -> "
f"{e.__class__.__name__ if hasattr(e, '__class__') else e}: {e}"
)
logger.error(ERR)
finally: