Run a dispatch loop for a single view instance
(
layout: LayoutType[LayoutUpdateMessage, LayoutEventMessage],
send: SendCoroutine,
recv: RecvCoroutine,
)
| 35 | |
| 36 | |
| 37 | async def serve_layout( |
| 38 | layout: LayoutType[LayoutUpdateMessage, LayoutEventMessage], |
| 39 | send: SendCoroutine, |
| 40 | recv: RecvCoroutine, |
| 41 | ) -> None: |
| 42 | """Run a dispatch loop for a single view instance""" |
| 43 | async with layout: |
| 44 | try: |
| 45 | async with create_task_group() as task_group: |
| 46 | task_group.start_soon(_single_outgoing_loop, layout, send) |
| 47 | task_group.start_soon(_single_incoming_loop, task_group, layout, recv) |
| 48 | except Stop: # nocov |
| 49 | warn( |
| 50 | "The Stop exception is deprecated and will be removed in a future version", |
| 51 | UserWarning, |
| 52 | stacklevel=1, |
| 53 | ) |
| 54 | logger.info(f"Stopped serving {layout}") |
| 55 | |
| 56 | |
| 57 | async def _single_outgoing_loop( |