61 lines
1.6 KiB
Python
61 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
MAX_SEQUENCE = 0xFFFFFFFF
|
|
_HALF_RANGE = 0x80000000
|
|
|
|
|
|
def next_sequence(current: int) -> int:
|
|
if not isinstance(current, int) or isinstance(current, bool) or not 0 <= current <= MAX_SEQUENCE:
|
|
raise ValueError("snapshot sequence must be an unsigned 32-bit integer")
|
|
value = (current + 1) & MAX_SEQUENCE
|
|
return 1 if value == 0 else value
|
|
|
|
|
|
def is_newer(candidate: int, baseline: int) -> bool:
|
|
if not isinstance(candidate, int) or isinstance(candidate, bool):
|
|
return False
|
|
if not isinstance(baseline, int) or isinstance(baseline, bool):
|
|
return False
|
|
if not 0 <= candidate <= MAX_SEQUENCE or not 0 <= baseline <= MAX_SEQUENCE:
|
|
return False
|
|
if candidate == 0:
|
|
return False
|
|
if baseline == 0:
|
|
return True
|
|
delta = (candidate - baseline) & MAX_SEQUENCE
|
|
return 0 < delta < _HALF_RANGE
|
|
|
|
|
|
class SequenceCounter:
|
|
def __init__(self) -> None:
|
|
self._value = 0
|
|
|
|
@property
|
|
def current(self) -> int:
|
|
return self._value
|
|
|
|
def advance(self) -> int:
|
|
self._value = next_sequence(self._value)
|
|
return self._value
|
|
|
|
def reset(self) -> None:
|
|
self._value = 0
|
|
|
|
|
|
class SequenceWindow:
|
|
def __init__(self) -> None:
|
|
self._last_accepted = 0
|
|
|
|
@property
|
|
def last_accepted(self) -> int:
|
|
return self._last_accepted
|
|
|
|
def accept(self, sequence: int) -> bool:
|
|
if not is_newer(sequence, self._last_accepted):
|
|
return False
|
|
self._last_accepted = sequence
|
|
return True
|
|
|
|
def reset(self) -> None:
|
|
self._last_accepted = 0
|