Message id len: 8 bytes

Author: attdona

pyproject.toml

@@ -12,7 +12,7 @@ authors = [
 description = "Rembus for python"
 readme = "README.md"
 keywords = ["rembus", "rpc", "publish", "subscribe", "websocket", "cbor"]
-requires-python = ">=3.7"
+requires-python = ">=3.12"
 classifiers = [
     "Programming Language :: Python :: 3",
     "Operating System :: OS Independent",

src/rembus/__init__.py

@@ -11,6 +11,7 @@
     QOSLevel,
     CBOR,
     JSON,
+    MSGID_LEN,
     SIG_ECDSA,
     SIG_RSA
 )

src/rembus/protocol.py

@@ -38,6 +38,7 @@
 CBOR = 0
 JSON = 1
 
+MSGID_LEN = 8
 
 class QOSLevel(IntEnum):
     """The Pub/Sub message QOS level."""
@@ -97,26 +98,10 @@ class QOSLevel(IntEnum):
 
 
 def msgid():
-    """Return an array of 16 random bytes."""
-    return int.from_bytes(os.urandom(16))
+    """Return an array of MSGID_LEN random bytes."""
+    return int.from_bytes(os.urandom(MSGID_LEN))
 
 
-def bytes2id(byte_data: bytearray) -> int:
-    """
-    Converts a 16-byte bytearray to a UInt128 integer.
-
-    Args:
-        byte_data: A bytearray of 16 bytes.
-
-    Returns:
-        A Python integer representing the UInt128 value.
-    """
-    if len(byte_data) != 16:
-        raise ValueError("bytearray must be exactly 16 bytes long")
-
-    # Convert bytes to integer (assuming little-endian, adjust if big-endian)
-    return int.from_bytes(byte_data, byteorder='little', signed=False)
-
 
 class RembusException(Exception):
     """Base class for all Rembus exceptions."""
@@ -151,8 +136,8 @@ def __str__(self):
 
 
 def to_bytes(val: int) -> bytes:
-    """Convert an int to 16 bytes"""
-    return val.to_bytes(16)
+    """Convert an int to MSGID_LEN bytes"""
+    return val.to_bytes(MSGID_LEN)
 
 
 class RembusMsg(BaseModel):

tests/api/test_errors.py

@@ -65,5 +65,5 @@ def test_response_no_data():
     mid = 1234
     # This is not an error, data is set by default to None
     rb._runner.run(rb._rb._send(cbor2.dumps(
-        [rp.TYPE_RESPONSE, mid.to_bytes(16), rp.STS_OK])))
+        [rp.TYPE_RESPONSE, mid.to_bytes(rp.MSGID_LEN), rp.STS_OK])))
     rb.close()

tests/unit/test_protocol.py

@@ -17,18 +17,6 @@ def test_no_impl():
         msg.to_payload(enc=rp.CBOR)
 
 
-def test_bytes2id():
-    """Test the bytes2id function for converting byte arrays to IDs."""
-    byte_data = bytearray(range(16))  # 0x00 to 0x0F
-    result = rp.bytes2id(byte_data)
-    assert isinstance(result, int)
-
-    with pytest.raises(ValueError):
-        rp.bytes2id(bytearray(range(15)))
-
-    with pytest.raises(ValueError):
-        rp.bytes2id(bytearray(range(17)))
-
 
 def test_types_str():
     """Test the string representation of protocol types."""

tests/unit/test_publish_qos2_duplicate.py

@@ -55,7 +55,7 @@ async def test_publish(mocker, ws_mock):
     await rb.subscribe(mytopic)
 
     # send two pubsub messages with the same msgid
-    msgid = bytes([i for i in range(16)])
+    msgid = bytes([i for i in range(rp.MSGID_LEN)])
     topic = mytopic.__name__
     for i in range(3):
         rb.outreq[int.from_bytes(msgid)] = rembus.core.FutureResponse(True)

tests/unit/test_unknown_msgid.py

@@ -36,7 +36,7 @@ async def test_unknown_message_id(mocker, ws_mock):
     assert mocked_connect.call_args[0][0] == "ws://127.0.0.1:8000/foo"
     assert rb.uid.id == 'foo'
     # send a response message with an unknown msgid
-    msgid = bytes([i for i in range(16)])
+    msgid = bytes([i for i in range(rp.MSGID_LEN)])
     req = rp.encode(
         [rp.TYPE_RESPONSE, msgid, rp.STS_OK, 'payload']
     )