Move `brokerd` service task to root `.data` mod
parent
470079665f
commit
844626f6dc
|
@ -22,6 +22,12 @@ and storing data from your brokers as well as
|
||||||
sharing live streams over a network.
|
sharing live streams over a network.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
import tractor
|
||||||
|
import trio
|
||||||
|
|
||||||
|
from ..log import (
|
||||||
|
get_console_log,
|
||||||
|
)
|
||||||
from ._normalize import iterticks
|
from ._normalize import iterticks
|
||||||
from ._sharedmem import (
|
from ._sharedmem import (
|
||||||
maybe_open_shm_array,
|
maybe_open_shm_array,
|
||||||
|
@ -32,7 +38,6 @@ from ._sharedmem import (
|
||||||
)
|
)
|
||||||
from .feed import (
|
from .feed import (
|
||||||
open_feed,
|
open_feed,
|
||||||
_setup_persistent_brokerd,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ -44,5 +49,40 @@ __all__ = [
|
||||||
'attach_shm_array',
|
'attach_shm_array',
|
||||||
'open_shm_array',
|
'open_shm_array',
|
||||||
'get_shm_token',
|
'get_shm_token',
|
||||||
'_setup_persistent_brokerd',
|
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context
|
||||||
|
async def _setup_persistent_brokerd(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
brokername: str,
|
||||||
|
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Allocate a actor-wide service nursery in ``brokerd``
|
||||||
|
such that feeds can be run in the background persistently by
|
||||||
|
the broker backend as needed.
|
||||||
|
|
||||||
|
'''
|
||||||
|
get_console_log(tractor.current_actor().loglevel)
|
||||||
|
|
||||||
|
from .feed import (
|
||||||
|
_bus,
|
||||||
|
get_feed_bus,
|
||||||
|
)
|
||||||
|
global _bus
|
||||||
|
assert not _bus
|
||||||
|
|
||||||
|
async with trio.open_nursery() as service_nursery:
|
||||||
|
# assign a nursery to the feeds bus for spawning
|
||||||
|
# background tasks from clients
|
||||||
|
get_feed_bus(brokername, service_nursery)
|
||||||
|
|
||||||
|
# unblock caller
|
||||||
|
await ctx.started()
|
||||||
|
|
||||||
|
# we pin this task to keep the feeds manager active until the
|
||||||
|
# parent actor decides to tear it down
|
||||||
|
await trio.sleep_forever()
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -54,7 +54,10 @@ import numpy as np
|
||||||
|
|
||||||
from ..brokers import get_brokermod
|
from ..brokers import get_brokermod
|
||||||
from ..calc import humanize
|
from ..calc import humanize
|
||||||
from ..log import get_logger, get_console_log
|
from ..log import (
|
||||||
|
get_logger,
|
||||||
|
get_console_log,
|
||||||
|
)
|
||||||
from .._daemon import (
|
from .._daemon import (
|
||||||
maybe_spawn_brokerd,
|
maybe_spawn_brokerd,
|
||||||
check_for_service,
|
check_for_service,
|
||||||
|
@ -224,36 +227,6 @@ def get_feed_bus(
|
||||||
return _bus
|
return _bus
|
||||||
|
|
||||||
|
|
||||||
@tractor.context
|
|
||||||
async def _setup_persistent_brokerd(
|
|
||||||
ctx: tractor.Context,
|
|
||||||
brokername: str,
|
|
||||||
|
|
||||||
) -> None:
|
|
||||||
'''
|
|
||||||
Allocate a actor-wide service nursery in ``brokerd``
|
|
||||||
such that feeds can be run in the background persistently by
|
|
||||||
the broker backend as needed.
|
|
||||||
|
|
||||||
'''
|
|
||||||
get_console_log(tractor.current_actor().loglevel)
|
|
||||||
|
|
||||||
global _bus
|
|
||||||
assert not _bus
|
|
||||||
|
|
||||||
async with trio.open_nursery() as service_nursery:
|
|
||||||
# assign a nursery to the feeds bus for spawning
|
|
||||||
# background tasks from clients
|
|
||||||
get_feed_bus(brokername, service_nursery)
|
|
||||||
|
|
||||||
# unblock caller
|
|
||||||
await ctx.started()
|
|
||||||
|
|
||||||
# we pin this task to keep the feeds manager active until the
|
|
||||||
# parent actor decides to tear it down
|
|
||||||
await trio.sleep_forever()
|
|
||||||
|
|
||||||
|
|
||||||
def diff_history(
|
def diff_history(
|
||||||
array: np.ndarray,
|
array: np.ndarray,
|
||||||
timeframe: int,
|
timeframe: int,
|
||||||
|
|
Loading…
Reference in New Issue