fixes

Author: om26er

tests/codec_test.py

@@ -1,54 +1,34 @@
 import os
-import base64
 from pathlib import Path
-from typing import Type, TypeVar, Any
+from typing import Type, TypeVar
 
 import capnp
 from google.protobuf.message import Message
 
 from xconn.client import connect_anonymous
-from xconn import codec
-from xconn.types import Event
+from xconn.codec import Codec
+from xconn.types import Event, OutgoingDataMessage, IncomingDataMessage
 from tests.schemas.profile_pb2 import ProfileCreate, ProfileGet
 
 
-class String(str):
-    pass
+T = TypeVar("T", bound=Message)
 
 
-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"
-
-
-class ProtobufCodec(codec.Codec[Message]):
+class ProtobufCodec(Codec[T]):
     def name(self) -> str:
         return "protobuf"
 
-    def encode(self, obj: Message) -> bytes:
-        return obj.SerializeToString()
+    def encode(self, obj: T) -> OutgoingDataMessage:
+        payload = obj.SerializeToString()
+        return OutgoingDataMessage(args=[payload], kwargs={}, details={})
 
-    def decode(self, data: bytes, out_type: Type[Message]) -> Message:
-        msg = out_type()
-        msg.ParseFromString(data)
+    def decode(self, msg: IncomingDataMessage, out_type: Type[T]) -> T:
+        if len(msg.args) == 0 or not isinstance(msg.args[0], bytes):
+            raise ValueError("ProtobufCodec: cannot decode, expected first arg to be bytes")
 
-        return msg
+        obj = out_type()
+        obj.ParseFromString(msg.args[0])
+        return obj
 
 
 def test_rpc_object_protobuf():
@@ -78,20 +58,6 @@ def inv_handler(profile: ProfileCreate) -> ProfileGet:
     session.leave()
 
 
-def test_pubsub_object():
-    session = connect_anonymous("ws://localhost:8080/ws", "realm1")
-    session.set_payload_codec(Base64Codec())
-
-    def event_handler(event: Event):
-        assert event.args[0] == "hello"
-
-    session.subscribe_object("io.xconn.object", event_handler, String)
-
-    session.publish_object("io.xconn.object", String("hello"))
-
-    session.leave()
-
-
 def test_pubsub_protobuf():
     session = connect_anonymous("ws://localhost:8080/ws", "realm1")
     session.set_payload_codec(ProtobufCodec())
@@ -192,15 +158,19 @@ def get_profile_handler() -> ProfileGet:
 UserGet = user_capnp.UserGet
 
 
-class CapnpProtoCodec(codec.Codec[T]):
+class CapnpProtoCodec(Codec[T]):
     def name(self) -> str:
         return "capnproto"
 
-    def encode(self, obj: Any) -> bytes:
-        return obj.to_bytes_packed()
+    def encode(self, obj: T) -> OutgoingDataMessage:
+        payload = obj.to_bytes_packed()
+        return OutgoingDataMessage(args=[payload], kwargs={}, details={})
+
+    def decode(self, msg: IncomingDataMessage, out_type: Type[T]) -> T:
+        if len(msg.args) == 0 or not isinstance(msg.args[0], bytes):
+            raise ValueError("CapnpProtoCodec: cannot decode, expected first arg to be bytes")
 
-    def decode(self, data: bytes, out_type: Type[T]) -> T:
-        return out_type.from_bytes_packed(data)
+        return out_type.from_bytes_packed(msg.args[0])
 
 
 def test_rpc_object_capnproto():

xconn/codec.py

@@ -1,4 +1,6 @@
-from typing import Any, Generic, Type, TypeVar
+from typing import Generic, Type, TypeVar
+
+from xconn.types import IncomingDataMessage, OutgoingDataMessage
 
 T = TypeVar("T")
 
@@ -7,10 +9,10 @@ class Codec(Generic[T]):
     def name(self) -> str:
         raise NotImplementedError
 
-    def encode(self, obj: Any) -> bytes | str:
+    def encode(self, obj: T) -> OutgoingDataMessage:
         """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."""
+    def decode(self, msg: IncomingDataMessage, out_type: Type[T]) -> T:
+        """Deserialize the incoming message into an instance of out_type."""
         raise NotImplementedError

xconn/session.py

@@ -3,7 +3,7 @@
 import inspect
 from concurrent.futures import Future
 from threading import Thread
-from typing import Callable, Any, TypeVar, Type
+from typing import Callable, Any, TypeVar, Type, overload
 from dataclasses import dataclass
 
 from wampproto import messages, session, uris
@@ -180,13 +180,25 @@ def _process_incoming_message(self, msg: messages.Message):
     def set_payload_codec(self, codec: Codec) -> None:
         self._payload_codec = codec
 
-    def call_object(self, procedure: str, request: TReq = None, return_type: Type[TRes] = None) -> TRes | None:
+    @overload
+    def call_object(self, procedure: str, request: TReq, return_type: Type[TRes]) -> TRes:
+        ...
+
+    @overload
+    def call_object(self, procedure: str, request: None = None, return_type: None = None) -> None:
+        ...
+
+    @overload
+    def call_object(self, procedure: str, request: None, return_type: Type[TRes]) -> TRes:
+        ...
+
+    def call_object(self, procedure: str, request: TReq = None, return_type: Type[TRes] | None = None) -> TRes | None:
         if self._payload_codec is None:
             raise ValueError("no payload codec set")
 
         if request is not None:
             encoded = self._payload_codec.encode(request)
-            result = self.call(procedure, [encoded])
+            result = self.call(procedure, args=encoded.args, kwargs=encoded.kwargs, options=encoded.details)
         else:
             result = self.call(procedure)
 

xconn/types.py

@@ -26,26 +26,34 @@ class UnsubscribeRequest:
 
 
 @dataclass
-class Result:
+class OutgoingDataMessage:
     args: list | None = None
     kwargs: dict | None = None
     details: dict | None = None
 
 
 @dataclass
-class Invocation:
-    args: list | None
-    kwargs: dict | None
-    details: dict | None
+class Result(OutgoingDataMessage):
+    pass
 
 
 @dataclass
-class Event:
+class IncomingDataMessage:
     args: list | None
     kwargs: dict | None
     details: dict | None
 
 
+@dataclass
+class Invocation(IncomingDataMessage):
+    pass
+
+
+@dataclass()
+class Event(IncomingDataMessage):
+    pass
+
+
 @dataclass
 class TransportConfig:
     # max wait time for connection to be established