Skip to content
This repository was archived by the owner on Jun 17, 2026. It is now read-only.

adds self.run_forever_flag to streams #501

Open
0xfMissingNo wants to merge 1 commit into
base: master
Choose a base branch
from
Open

adds self.run_forever_flag to streams #501

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion alpaca_trade_api/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,4 +3,4 @@
from .stream import Stream # noqa
from .stream2 import StreamConn # noqa

__version__ = '1.4.0'
__version__ = '1.4.1'

@haxdds haxdds Nov 15, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi 0xfMissingNo! Version changes are necessary in your PR. They are made when we release and deploy a new version.

26 changes: 24 additions & 2 deletions alpaca_trade_api/stream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ def __init__(self,
'dailyBars': {},
}
self._name = 'data'
self._run_forever_flag = False

async def _connect(self):
self._ws = await websockets.connect(
Expand Down Expand Up @@ -79,6 +80,7 @@ async def close(self):
await self._ws.close()
self._ws = None
self._running = False
self._run_forever_flag = False

async def stop_ws(self):
self._stop_stream_queue.put_nowait({"should_stop": True})
Expand Down Expand Up @@ -191,7 +193,8 @@ async def _run_forever(self):
await asyncio.sleep(0.1)
log.info(f'started {self._name} stream')
self._running = False
while True:
self._run_forever_flag = True
while self._run_forever_flag:
try:
if not self._running:
log.info("starting websocket connection")
Expand Down Expand Up @@ -376,6 +379,7 @@ def __init__(self,
self._ws = None
self._running = False
self._stop_stream_queue = queue.Queue()
self._run_forever_flag = False

@haxdds haxdds Nov 15, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the purpose of this flag? What functionality does it add over while True?


async def _connect(self):
self._ws = await websockets.connect(self._endpoint)
Expand Down Expand Up @@ -443,7 +447,8 @@ async def _run_forever(self):
await asyncio.sleep(0.1)
log.info('started trading stream')
self._running = False
while True:
self._run_forever_flag = True
while self._run_forever_flag:
try:
if not self._running:
log.info("starting websocket connection")
Expand All @@ -466,6 +471,7 @@ async def close(self):
await self._ws.close()
self._ws = None
self._running = False
self._run_forever_flag = False

async def stop_ws(self):
self._stop_stream_queue.put_nowait({"should_stop": True})
Expand Down Expand Up @@ -648,6 +654,22 @@ def run(self):
print('keyboard interrupt, bye')
pass

async def _close(self):

@haxdds haxdds Nov 15, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for taking the time to contribute! To help us out, could you please elaborate a bit further on the goal of your PR?

await asyncio.gather(
self.stop_ws(),
self._trading_ws.close(),
self._data_ws.close(),
self._crypto_ws.close()
)

def close(self):
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(self._close())
except KeyboardInterrupt:
print('keyboard interrupt, bye')
pass

async def stop_ws(self):
"""
Signal the ws connections to stop listenning to api stream.
Expand Down