Get a watcher object based on the provided parameters.
PARAMETER | DESCRIPTION |
logger | Optional logger object for logging messages. TYPE: Optional[LoggerProto] |
try_number | Optional parameter to specify the type of watcher. - If set to True, an EndlessWatcher object will be returned. - If set to False, a OneTryWatcher object will be returned. - If set to an integer, a CounterWatcher object with the specified maximum number of tries will be returned. TYPE: Union[bool, int] |
Source code in faststream/broker/acknowledgement_watcher.py
| def get_watcher(
logger: Optional["LoggerProto"],
try_number: Union[bool, int],
) -> BaseWatcher:
"""Get a watcher object based on the provided parameters.
Args:
logger: Optional logger object for logging messages.
try_number: Optional parameter to specify the type of watcher.
- If set to True, an EndlessWatcher object will be returned.
- If set to False, a OneTryWatcher object will be returned.
- If set to an integer, a CounterWatcher object with the specified maximum number of tries will be returned.
"""
watcher: Optional[BaseWatcher]
if try_number is True:
watcher = EndlessWatcher()
elif try_number is False:
watcher = OneTryWatcher()
else:
watcher = CounterWatcher(logger=logger, max_tries=try_number)
return watcher
|