Port `test_pubsub` off `run_in_actor`
`test_multi_actor_subs_arbiter_pub` used `run_in_actor()` to spawn two forever-ish subscriber actors and hold their portals for a later `cancel_actor()` (its `.result()` was commented out exactly because `subs()` never cleanly returns). That deferred-spawn + cancel shape isn't a blocking `to_actor.run()`, so convert to the successor primitives (#477 removal), - `start_actor()` per subscriber — keeps the portal for the existing `cancel_actor()` teardown, - run `subs()` on each via a background `Portal.run()` task in a local `trio` nursery so both subscribe concurrently with the test's `wait_for_actor` / topic checks, - each bg runner swallows the `RemoteActorError`/`ContextCancelled` that `cancel_actor()` relays; a trailing `tn.cancel_scope.cancel()` drops any lingering runner. Suite: 8 passed. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-codedrop_ria_nursery
parent
dfa2be7078
commit
44d730c723
|
|
@ -176,10 +176,13 @@ def test_multi_actor_subs_arbiter_pub(
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
|
|
||||||
async with tractor.open_nursery(
|
async with (
|
||||||
registry_addrs=[reg_addr],
|
tractor.open_nursery(
|
||||||
enable_modules=[__name__],
|
registry_addrs=[reg_addr],
|
||||||
) as n:
|
enable_modules=[__name__],
|
||||||
|
) as n,
|
||||||
|
trio.open_nursery() as tn,
|
||||||
|
):
|
||||||
|
|
||||||
name = 'root'
|
name = 'root'
|
||||||
|
|
||||||
|
|
@ -191,18 +194,37 @@ def test_multi_actor_subs_arbiter_pub(
|
||||||
)
|
)
|
||||||
name = 'streamer'
|
name = 'streamer'
|
||||||
|
|
||||||
even_portal = await n.run_in_actor(
|
# spawn the two subscriber actors as daemons and run
|
||||||
subs,
|
# `subs()` on each as a background task (was the legacy
|
||||||
which=['even'],
|
# `run_in_actor()`); keep the portals for the explicit
|
||||||
name='evens',
|
# `cancel_actor()` teardown below. Each runner swallows
|
||||||
pub_actor_name=name
|
# the teardown error that `cancel_actor()` relays.
|
||||||
|
async def _run_subs(
|
||||||
|
portal: tractor.Portal,
|
||||||
|
which: list[str],
|
||||||
|
) -> None:
|
||||||
|
try:
|
||||||
|
await portal.run(
|
||||||
|
subs,
|
||||||
|
which=which,
|
||||||
|
pub_actor_name=name,
|
||||||
|
)
|
||||||
|
except (
|
||||||
|
tractor.RemoteActorError,
|
||||||
|
tractor.ContextCancelled,
|
||||||
|
):
|
||||||
|
pass # expected once we `cancel_actor()` below
|
||||||
|
|
||||||
|
even_portal = await n.start_actor(
|
||||||
|
'evens',
|
||||||
|
enable_modules=[__name__],
|
||||||
)
|
)
|
||||||
odd_portal = await n.run_in_actor(
|
odd_portal = await n.start_actor(
|
||||||
subs,
|
'odds',
|
||||||
which=['odd'],
|
enable_modules=[__name__],
|
||||||
name='odds',
|
|
||||||
pub_actor_name=name
|
|
||||||
)
|
)
|
||||||
|
tn.start_soon(_run_subs, even_portal, ['even'])
|
||||||
|
tn.start_soon(_run_subs, odd_portal, ['odd'])
|
||||||
|
|
||||||
async with tractor.wait_for_actor('evens'):
|
async with tractor.wait_for_actor('evens'):
|
||||||
# block until 2nd actor is initialized
|
# block until 2nd actor is initialized
|
||||||
|
|
@ -257,6 +279,9 @@ def test_multi_actor_subs_arbiter_pub(
|
||||||
else:
|
else:
|
||||||
await master_portal.cancel_actor()
|
await master_portal.cancel_actor()
|
||||||
|
|
||||||
|
# drop the bg `subs()` runners now the subs are cancelled
|
||||||
|
tn.cancel_scope.cancel()
|
||||||
|
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue