in elastic_transport/_transport.py [0:0]
def _should_sniff(self, is_initial_sniff: bool) -> bool:
"""Decide if we should sniff or not. If we return ``True`` from this
method the caller has a responsibility to unlock the ``_sniffing_lock``
"""
if not is_initial_sniff and (
time.time() - self._last_sniffed_at < self._min_delay_between_sniffing
):
return False
return self._sniffing_lock.acquire(False)