arno/trezor-firmware
1
0
Fork 0
mirror of https://github.com/trezor/trezor-firmware.git synced 2025-01-03 03:50:58 +00:00
Code Issues Releases Wiki Activity

refactor(core): simplify context.wait() to only accept one task

Browse Source 
Nobody was using the multi-task version and this is easier to properly
type-annotate.
This commit is contained in:
matejcik 2023-07-26 11:10:30 +02:00 committed by matejcik
parent e736d389f2
commit 29863f84ca
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
core/src/trezor/wire/context.py
View File

@ -38,6 +38,8 @@ if TYPE_CHECKING:
LoadedMessageType = TypeVar("LoadedMessageType", bound=protobuf.MessageType) LoadedMessageType = TypeVar("LoadedMessageType", bound=protobuf.MessageType)
T = TypeVar("T")
class UnexpectedMessage(Exception): class UnexpectedMessage(Exception):
"""A message was received that is not part of the current workflow. """A message was received that is not part of the current workflow.
@ -159,19 +161,19 @@ class Context:
CURRENT_CONTEXT: Context | None = None CURRENT_CONTEXT: Context | None = None
def wait(*tasks: Awaitable) -> Any: def wait(task: Awaitable[T]) -> Awaitable[T]:
""" """
Wait until one of the passed tasks finishes, and return the result, while servicing Wait until the passed in task finishes, and return the result, while servicing the
the wire context. wire context.
Used to make sure the device is responsive on USB while waiting for user Used to make sure the device is responsive on USB while waiting for user
interaction. If a message is received before any of the passed in tasks finish, it interaction. If a message is received before the task finishes, it raises an
raises an `UnexpectedMessage` exception, returning control to the session handler. `UnexpectedMessage` exception, returning control to the session handler.
""" """
if CURRENT_CONTEXT is None: if CURRENT_CONTEXT is None:
return loop.race(*tasks) return task
else: else:
return loop.race(CURRENT_CONTEXT.read(()), *tasks) return loop.race(CURRENT_CONTEXT.read(()), task)
async def call( async def call(