gsoc containing json serialization handler and dubbo codec

samples/llm/main.py

@@ -15,12 +15,12 @@
 # limitations under the License.
 from time import sleep
 
+import chat_pb2
 from lmdeploy import GenerationConfig, TurbomindEngineConfig, pipeline
 
 from dubbo import Dubbo
 from dubbo.configs import RegistryConfig, ServiceConfig
 from dubbo.proxy.handlers import RpcMethodHandler, RpcServiceHandler
-import chat_pb2
 
 # the path of a model. It could be one of the following options:
 # 1. A local directory path of a turbomind model

src/dubbo/classes.py

@@ -16,8 +16,9 @@
 
 import abc
 import threading
-from typing import Any, Callable, Optional, Union,Type
 from abc import ABC, abstractmethod
+from typing import Any, Callable, Optional, Union
+
 from dubbo.types import DeserializingFunction, RpcType, RpcTypes, SerializingFunction
 
 __all__ = [
@@ -248,7 +249,7 @@ class ReadWriteStream(ReadStream, WriteStream, abc.ABC):
 
 
 class Codec(ABC):
-    def __init__(self, model_type: Optional[Type[Any]] = None, **kwargs):
+    def __init__(self, model_type: Optional[type[Any]] = None, **kwargs):
         self.model_type = model_type
 
     @abstractmethod
@@ -259,6 +260,7 @@ def encode(self, data: Any) -> bytes:
     def decode(self, data: bytes) -> Any:
         pass
 
+
 class CodecHelper:
     @staticmethod
     def get_class():

src/dubbo/client.py

@@ -8,15 +8,14 @@
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an "AS IS" BASIS,
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
 import threading
-import inspect
-from typing import Optional, Callable, List, Type, Any, get_type_hints
+from typing import Optional, List, Type
 
 from dubbo.bootstrap import Dubbo
 from dubbo.classes import MethodDescriptor
@@ -33,7 +32,7 @@
     SerializingFunction,
 )
 from dubbo.url import URL
-from dubbo.codec import DubboTransportService
+from dubbo.codec import DubboSerializationService
 
 __all__ = ["Client"]
 
@@ -64,17 +63,11 @@ def _initialize(self):
                 return
 
             # get the protocol
-            protocol = extensionLoader.get_extension(
-                Protocol, self._reference.protocol
-            )()
+            protocol = extensionLoader.get_extension(Protocol, self._reference.protocol)()
 
             registry_config = self._dubbo.registry_config
 
-            self._protocol = (
-                RegistryProtocol(registry_config, protocol)
-                if registry_config
-                else protocol
-            )
+            self._protocol = RegistryProtocol(registry_config, protocol) if registry_config else protocol
 
             # build url
             reference_url = self._reference.to_url()
@@ -91,116 +84,82 @@ def _initialize(self):
 
             self._initialized = True
 
-    @classmethod
-    def _infer_types_from_interface(cls, interface: Callable) -> tuple:
-        """
-        Infer method name, parameter types, and return type from a callable.
-        """
-        try:
-            type_hints = get_type_hints(interface)
-            sig = inspect.signature(interface)
-            method_name = interface.__name__
-            params = list(sig.parameters.values())
-
-            # skip 'self' for bound methods
-            if params and params[0].name == "self":
-                params = params[1:]
-
-            param_types = [type_hints.get(p.name, Any) for p in params]
-            return_type = type_hints.get("return", Any)
-
-            return method_name, param_types, return_type
-        except Exception:
-            return interface.__name__, [Any], Any
-
     def _create_rpc_callable(
         self,
         rpc_type: str,
-        interface: Optional[Callable] = None,
-        method_name: Optional[str] = None,
-        params_types: Optional[List[Type]] = None,
-        return_type: Optional[Type] = None,
+        method_name: str,
+        params_types: List[Type],
+        return_type: Type,
         codec: Optional[str] = None,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
-        default_method_name: str = "rpc_call",
     ) -> RpcCallable:
         """
         Create RPC callable with the specified type.
         """
-        if interface is None and method_name is None:
-            raise ValueError("Either 'interface' or 'method_name' must be provided")
-
-        # Start with explicit values
-        m_name = method_name
-        p_types = params_types
-        r_type = return_type
-
-        # Infer from interface if needed
-        if interface:
-            if p_types is None or r_type is None or m_name is None:
-                inf_name, inf_params, inf_return = self._infer_types_from_interface(
-                    interface
-                )
-                m_name = m_name or inf_name
-                p_types = p_types or inf_params
-                r_type = r_type or inf_return
-
-        # Fallback to default
-        m_name = m_name or default_method_name
-
+        print("2", params_types)
         # Determine serializers
         if request_serializer and response_deserializer:
             req_ser = request_serializer
             res_deser = response_deserializer
         else:
-            req_ser, res_deser = DubboTransportService.create_serialization_functions(
-                codec or "json",  # fallback to json
-                parameter_types=p_types,
-                return_type=r_type,
+            req_ser, res_deser = DubboSerializationService.create_serialization_functions(
+                codec or "json",
+                parameter_types=params_types,
+                return_type=return_type,
             )
 
         # Create MethodDescriptor
         descriptor = MethodDescriptor(
-            method_name=m_name,
+            method_name=method_name,
             arg_serialization=(req_ser, None),
             return_serialization=(None, res_deser),
             rpc_type=rpc_type,
         )
 
         return self._callable(descriptor)
 
-    def unary(self, **kwargs) -> RpcCallable:
+    def unary(self, method_name: str, params_types: List[Type], return_type: Type, **kwargs) -> RpcCallable:
         return self._create_rpc_callable(
-            rpc_type=RpcTypes.UNARY.value, default_method_name="unary", **kwargs
+            rpc_type=RpcTypes.UNARY.value,
+            method_name=method_name,
+            params_types=params_types,
+            return_type=return_type,
+            **kwargs,
         )
 
-    def client_stream(self, **kwargs) -> RpcCallable:
+    def client_stream(self, method_name: str, params_types: List[Type], return_type: Type, **kwargs) -> RpcCallable:
         return self._create_rpc_callable(
             rpc_type=RpcTypes.CLIENT_STREAM.value,
-            default_method_name="client_stream",
+            method_name=method_name,
+            params_types=params_types,
+            return_type=return_type,
             **kwargs,
         )
 
-    def server_stream(self, **kwargs) -> RpcCallable:
+    def server_stream(self, method_name: str, params_types: List[Type], return_type: Type, **kwargs) -> RpcCallable:
         return self._create_rpc_callable(
             rpc_type=RpcTypes.SERVER_STREAM.value,
-            default_method_name="server_stream",
+            method_name=method_name,
+            params_types=params_types,
+            return_type=return_type,
             **kwargs,
         )
 
-    def bi_stream(self, **kwargs) -> RpcCallable:
+    def bi_stream(self, method_name: str, params_types: List[Type], return_type: Type, **kwargs) -> RpcCallable:
         return self._create_rpc_callable(
-            rpc_type=RpcTypes.BI_STREAM.value, default_method_name="bi_stream", **kwargs
+            rpc_type=RpcTypes.BI_STREAM.value,
+            method_name=method_name,
+            params_types=params_types,
+            return_type=return_type,
+            **kwargs,
         )
 
     def _callable(self, method_descriptor: MethodDescriptor) -> RpcCallable:
         """
         Generate a proxy for the given method.
         """
         url = self._invoker.get_url().copy()
-        url.parameters[common_constants.METHOD_KEY] = (
-            method_descriptor.get_method_name()
-        )
+        url.parameters[common_constants.METHOD_KEY] = method_descriptor.get_method_name()
         url.attributes[common_constants.METHOD_DESCRIPTOR_KEY] = method_descriptor
         return self._callable_factory.get_callable(self._invoker, url)

src/dubbo/codec/__init__.py

@@ -14,6 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .dubbo_codec import DubboTransportService
+from .dubbo_codec import DubboSerializationService
 
-__all__ = ['DubboTransportService']
+__all__ = ["DubboSerializationService"]