69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
import struct
|
|
|
|
|
|
class SnowflakeId:
|
|
SNOWFLAKE_ID_WIDTH = 10 # with in bytes (8 bits each), can be more than long int width (8 bytes)
|
|
|
|
def __init__(self, _input: bytes = None):
|
|
self.bits = [0, 0]
|
|
if _input is not None:
|
|
i = 0
|
|
shift = 8 # starting from Hi value, which is only 2 bytes wide
|
|
n = 1
|
|
while i < len(_input) and i <= self.SNOWFLAKE_ID_WIDTH:
|
|
self.bits[n] |= _input[i] << shift
|
|
if shift == 0:
|
|
n -= 1
|
|
shift = 56
|
|
else:
|
|
shift -= 8
|
|
i += 1
|
|
|
|
def __init__(self, hi_bits: int, lo_bits: int):
|
|
self.bits = [lo_bits, hi_bits]
|
|
|
|
@staticmethod
|
|
def from_hex(_input):
|
|
expected_length = 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH
|
|
if _input is None or len(_input) < expected_length:
|
|
raise RuntimeError(
|
|
f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string "
|
|
f"of length: {expected_length}, real input: [{_input}] of length: {len(_input) if _input else 0}"
|
|
)
|
|
|
|
bits = [0, 0]
|
|
n = 1
|
|
for i in range(0, expected_length, 2):
|
|
if i == expected_length - (2 * 8):
|
|
n = 0
|
|
bits[n] = (bits[n] << 8) | int(_input[i:i + 2], 16)
|
|
|
|
return SnowflakeId(bits[1], bits[0])
|
|
|
|
def __str__(self):
|
|
_hex = f"{self.bits[1]:016X}{self.bits[0]:016X}"
|
|
return _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):]
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, SnowflakeId):
|
|
return self.bits[1] == other.bits[1] and self.bits[0] == other.bits[0]
|
|
return False
|
|
|
|
def __lt__(self, other):
|
|
if self.bits[1] < other.bits[1]:
|
|
return True
|
|
elif self.bits[1] > other.bits[1]:
|
|
return False
|
|
return self.bits[0] < other.bits[0]
|
|
|
|
def get_bytes(self):
|
|
high = struct.pack("!Q", self.bits[1])
|
|
low = struct.pack("!Q", self.bits[0])
|
|
return high[2:] + low
|
|
|
|
def get_bits(self):
|
|
return self.bits
|
|
|
|
def as_string(self) -> str:
|
|
return self.__str__()
|