pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 5 Packages Projects Releases Wiki Activity

Port to new `Portal.run()` api

basic_alerts
Tyler Goodlet 2021-01-04 14:45:34 -05:00
parent 267c8c6bd3
commit 3c424a153f
1 changed files with 12 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
piker/data/__init__.py
View File

@ -75,10 +75,12 @@ def get_ingestormod(name: str) -> ModuleType:
return module return module
# capable rpc modules
_data_mods = [ _data_mods = [
'piker.brokers.core', 'piker.brokers.core',
'piker.brokers.data', 'piker.brokers.data',
'piker.data', 'piker.data',
'piker.data._buffer',
] ]
@ -104,10 +106,13 @@ async def maybe_spawn_brokerd(
brokermod = get_brokermod(brokername) brokermod = get_brokermod(brokername)
dname = f'brokerd.{brokername}' dname = f'brokerd.{brokername}'
async with tractor.find_actor(dname) as portal: async with tractor.find_actor(dname) as portal:
# WTF: why doesn't this work? # WTF: why doesn't this work?
if portal is not None: if portal is not None:
yield portal yield portal
else:
else: # no daemon has been spawned yet
log.info(f"Spawning {brokername} broker daemon") log.info(f"Spawning {brokername} broker daemon")
tractor_kwargs = getattr(brokermod, '_spawn_kwargs', {}) tractor_kwargs = getattr(brokermod, '_spawn_kwargs', {})
async with tractor.open_nursery() as nursery: async with tractor.open_nursery() as nursery:
@ -115,7 +120,7 @@ async def maybe_spawn_brokerd(
# spawn new daemon # spawn new daemon
portal = await nursery.start_actor( portal = await nursery.start_actor(
dname, dname,
rpc_module_paths=_data_mods + [brokermod.__name__], enable_modules=_data_mods + [brokermod.__name__],
loglevel=loglevel, loglevel=loglevel,
**tractor_kwargs **tractor_kwargs
) )
@ -140,7 +145,7 @@ class Feed:
stream: AsyncIterator[Dict[str, Any]] stream: AsyncIterator[Dict[str, Any]]
shm: ShmArray shm: ShmArray
# ticks: ShmArray # ticks: ShmArray
_broker_portal: tractor._portal.Portal _brokerd_portal: tractor._portal.Portal
_index_stream: Optional[AsyncIterator[Dict[str, Any]]] = None _index_stream: Optional[AsyncIterator[Dict[str, Any]]] = None
async def receive(self) -> dict: async def receive(self) -> dict:
@ -151,9 +156,8 @@ class Feed:
# XXX: this should be singleton on a host, # XXX: this should be singleton on a host,
# a lone broker-daemon per provider should be # a lone broker-daemon per provider should be
# created for all practical purposes # created for all practical purposes
self._index_stream = await self._broker_portal.run( self._index_stream = await self._brokerd_portal.run(
'piker.data', increment_ohlc_buffer,
'increment_ohlc_buffer',
shm_token=self.shm.token, shm_token=self.shm.token,
topics=['index'], topics=['index'],
) )
@ -200,8 +204,7 @@ async def open_feed(
loglevel=loglevel, loglevel=loglevel,
) as portal: ) as portal:
stream = await portal.run( stream = await portal.run(
mod.__name__, mod.stream_quotes,
'stream_quotes',
symbols=symbols, symbols=symbols,
shm_token=shm.token, shm_token=shm.token,
@ -225,5 +228,5 @@ async def open_feed(
name=name, name=name,
stream=stream, stream=stream,
shm=shm, shm=shm,
_broker_portal=portal, _brokerd_portal=portal,
) )