add initial support for payload codec

Author: om26er

tests/codec_test.py

@@ -0,0 +1,39 @@
+import base64
+from typing import Type
+
+from xconn.client import connect_anonymous
+from xconn import codec
+
+
+class String(str):
+    pass
+
+
+class Base64Codec(codec.Codec[String]):
+    def name(self) -> str:
+        return "base64"
+
+    def encode(self, obj: String) -> str:
+        return base64.b64encode(obj.encode("utf-8")).decode("utf-8")
+
+    def decode(self, data: str, out_type: Type[String]) -> String:
+        return out_type(base64.b64decode(data.encode("utf-8")).decode())
+
+
+def test_base64_codec():
+    encoder = Base64Codec()
+    encoded = encoder.encode(String("hello"))
+    assert isinstance(encoded, str)
+
+    decoded = encoder.decode(encoded, String)
+    assert isinstance(decoded, String)
+    assert decoded == "hello"
+
+
+def test_something():
+    # session = connect_anonymous("ws://localhost:8080/ws", "realm1")
+    # session.set_payload_codec(Base64Codec())
+    # result = session.call_object("io.xconn.object", String("hello"), String)
+    # print(result)
+    # session.leave()
+    pass

xconn/codec.py

@@ -0,0 +1,16 @@
+from typing import Any, Generic, Type, TypeVar
+
+T = TypeVar("T")
+
+
+class Codec(Generic[T]):
+    def name(self) -> str:
+        raise NotImplementedError
+
+    def encode(self, obj: Any) -> bytes | str:
+        """Serialize a Python object to bytes."""
+        raise NotImplementedError
+
+    def decode(self, data: bytes | str, out_type: Type[T]) -> T:
+        """Deserialize bytes into an instance of out_type."""
+        raise NotImplementedError

xconn/session.py

@@ -2,15 +2,19 @@
 
 from concurrent.futures import Future
 from threading import Thread
-from typing import Callable, Any
+from typing import Callable, Any, TypeVar, Type
 from dataclasses import dataclass
 
 from wampproto import messages, session, uris
 
 from xconn import types, exception, uris as xconn_uris
+from xconn.codec import Codec
 from xconn.exception import ApplicationError
 from xconn.helpers import exception_from_error, SessionScopeIDGenerator
 
+TReq = TypeVar("TReq")
+TRes = TypeVar("TRes")
+
 
 @dataclass
 class RegisterRequest:
@@ -90,6 +94,8 @@ def __init__(self, base_session: types.BaseSession):
 
         self._disconnect_callback: list[Callable[[], None] | None] = []
 
+        self._payload_codec: Codec = None
+
         thread = Thread(target=self._wait, daemon=False)
         thread.start()
 
@@ -192,6 +198,18 @@ def _process_incoming_message(self, msg: messages.Message):
         else:
             raise ValueError("received unknown message")
 
+    def set_payload_codec(self, codec: Codec) -> None:
+        self._payload_codec = codec
+
+    def call_object(self, procedure: str, request: TReq, return_type: Type[TRes] = None) -> TReq | None:
+        if self._payload_codec is None:
+            raise ValueError("no payload codec set")
+
+        encoded = self._payload_codec.encode(request)
+        result = self.call(procedure, [encoded])
+
+        return self._payload_codec.decode(result.args[0], return_type)
+
     def call(
         self,
         procedure: str,