Skip to content

Commit e206a24

Browse files
committed
Move maybe_open_feed() above for readability
1 parent 6b5bf62 commit e206a24

1 file changed

Lines changed: 51 additions & 51 deletions

File tree

piker/data/feed.py

Lines changed: 51 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -1495,6 +1495,57 @@ async def search(text: str) -> dict[str, Any]:
14951495
yield
14961496

14971497

1498+
@acm
1499+
async def maybe_open_feed(
1500+
1501+
fqsns: list[str],
1502+
loglevel: Optional[str] = None,
1503+
1504+
**kwargs,
1505+
1506+
) -> (
1507+
Feed,
1508+
ReceiveChannel[dict[str, Any]],
1509+
):
1510+
'''
1511+
Maybe open a data to a ``brokerd`` daemon only if there is no
1512+
local one for the broker-symbol pair, if one is cached use it wrapped
1513+
in a tractor broadcast receiver.
1514+
1515+
'''
1516+
fqsn = fqsns[0]
1517+
1518+
async with maybe_open_context(
1519+
acm_func=open_feed,
1520+
kwargs={
1521+
'fqsns': fqsns,
1522+
'loglevel': loglevel,
1523+
'tick_throttle': kwargs.get('tick_throttle'),
1524+
1525+
# XXX: super critical to have bool defaults here XD
1526+
'backpressure': kwargs.get('backpressure', True),
1527+
'start_stream': kwargs.get('start_stream', True),
1528+
},
1529+
key=fqsn,
1530+
1531+
) as (cache_hit, feed):
1532+
1533+
if cache_hit:
1534+
log.info(f'Using cached feed for {fqsn}')
1535+
# add a new broadcast subscription for the quote stream
1536+
# if this feed is likely already in use
1537+
1538+
async with gather_contexts(
1539+
mngrs=[stream.subscribe() for stream in feed.streams.values()]
1540+
) as bstreams:
1541+
for bstream, flume in zip(bstreams, feed.flumes.values()):
1542+
flume.stream = bstream
1543+
1544+
yield feed
1545+
else:
1546+
yield feed
1547+
1548+
14981549
@acm
14991550
async def open_feed(
15001551

@@ -1640,54 +1691,3 @@ async def open_feed(
16401691
assert len(feed.mods) == len(feed.portals) == len(feed.streams)
16411692

16421693
yield feed
1643-
1644-
1645-
@acm
1646-
async def maybe_open_feed(
1647-
1648-
fqsns: list[str],
1649-
loglevel: Optional[str] = None,
1650-
1651-
**kwargs,
1652-
1653-
) -> (
1654-
Feed,
1655-
ReceiveChannel[dict[str, Any]],
1656-
):
1657-
'''
1658-
Maybe open a data to a ``brokerd`` daemon only if there is no
1659-
local one for the broker-symbol pair, if one is cached use it wrapped
1660-
in a tractor broadcast receiver.
1661-
1662-
'''
1663-
fqsn = fqsns[0]
1664-
1665-
async with maybe_open_context(
1666-
acm_func=open_feed,
1667-
kwargs={
1668-
'fqsns': fqsns,
1669-
'loglevel': loglevel,
1670-
'tick_throttle': kwargs.get('tick_throttle'),
1671-
1672-
# XXX: super critical to have bool defaults here XD
1673-
'backpressure': kwargs.get('backpressure', True),
1674-
'start_stream': kwargs.get('start_stream', True),
1675-
},
1676-
key=fqsn,
1677-
1678-
) as (cache_hit, feed):
1679-
1680-
if cache_hit:
1681-
log.info(f'Using cached feed for {fqsn}')
1682-
# add a new broadcast subscription for the quote stream
1683-
# if this feed is likely already in use
1684-
1685-
async with gather_contexts(
1686-
mngrs=[stream.subscribe() for stream in feed.streams.values()]
1687-
) as bstreams:
1688-
for bstream, flume in zip(bstreams, feed.flumes.values()):
1689-
flume.stream = bstream
1690-
1691-
yield feed
1692-
else:
1693-
yield feed

0 commit comments

Comments
 (0)