Jun 9, 2025 · Jun 11, 2025 · Jun 19, 2025 · Jun 19, 2025 · Jun 19, 2025 · Jun 19, 2025
diff --git a/.gitignore b/.gitignore
@@ -64,3 +64,4 @@ venv.bak/
 .mypy_cache/
 .dmypy.json
 dmypy.json
+.DS_Store
diff --git a/src/.DS_Store b/src/.DS_Store
diff --git a/src/dubbo/.DS_Store b/src/dubbo/.DS_Store
diff --git a/src/dubbo/classes.py b/src/dubbo/classes.py
@@ -13,9 +13,12 @@
 # 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 abc
 import threading
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, Union, Type
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
 
 from dubbo.types import DeserializingFunction, RpcType, RpcTypes, SerializingFunction
 
@@ -244,3 +247,20 @@ class ReadWriteStream(ReadStream, WriteStream, abc.ABC):
     """
 
     pass
+
+class Codec(ABC):
+    def __init__(self, model_type: Type[BaseModel] = None, **kwargs):
+        self.model_type = model_type
+
+    @abstractmethod
+    def encode(self, data: Any) -> bytes:
+        pass
+
+    @abstractmethod
+    def decode(self, data: bytes) -> Any:
+        pass
+
+class CodecHelper:
+    @staticmethod
+    def get_class():
+        return Codec
diff --git a/src/dubbo/client.py b/src/dubbo/client.py
@@ -13,6 +13,7 @@
 # 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
 from typing import Optional
 
@@ -30,6 +31,7 @@
     RpcTypes,
     SerializingFunction,
 )
+from dubbo.codec import DubboCodec
 from dubbo.url import URL
 
 __all__ = ["Client"]
@@ -64,9 +66,10 @@ def _initialize(self):
             protocol = extensionLoader.get_extension(Protocol, self._reference.protocol)()
 
             registry_config = self._dubbo.registry_config
+            print("config",registry_config)
 
             self._protocol = RegistryProtocol(registry_config, protocol) if self._dubbo.registry_config else protocol
-
+            print(self._protocol)
             # build url
             reference_url = self._reference.to_url()
             if registry_config:
@@ -81,13 +84,30 @@ def _initialize(self):
             self._invoker = self._protocol.refer(self._url)
 
             self._initialized = True
+
+    @staticmethod
+    def get_codec(**kwargs) -> tuple:
+        """
+        Get the method encode and decode
+        :return: tuple of the encode and decode method 
+        """
+        return DubboCodec.get_serializer_deserializer(**kwargs)
+
 
     def unary(
         self,
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -102,7 +122,15 @@ def client_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -117,7 +145,15 @@ def server_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         return self._callable(
             MethodDescriptor(
                 method_name=method_name,
@@ -132,7 +168,15 @@ def bi_stream(
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,
         response_deserializer: Optional[DeserializingFunction] = None,
+        **kwargs
     ) -> RpcCallable:
+
+        # Use custom serializers if provided, otherwise get from codec
+        if request_serializer is None or response_deserializer is None:
+            default_deserializer, default_serializer = self.get_codec(**kwargs)
+            request_serializer = request_serializer or default_serializer
+            response_deserializer = response_deserializer or default_deserializer
+
         # create method descriptor
         return self._callable(
             MethodDescriptor(
@@ -160,4 +204,4 @@ def _callable(self, method_descriptor: MethodDescriptor) -> RpcCallable:
         url.attributes[common_constants.METHOD_DESCRIPTOR_KEY] = method_descriptor
 
         # create proxy
-        return self._callable_factory.get_callable(self._invoker, url)
+        return self._callable_factory.get_callable(self._invoker, url)
diff --git a/src/dubbo/codec/.DS_Store b/src/dubbo/codec/.DS_Store
diff --git a/src/dubbo/codec/__init__.py b/src/dubbo/codec/__init__.py
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     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,
+# 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.
+
+from .dubbo_codec import DubboCodec
+
+__all__ = ["DubboCodec"]
diff --git a/src/dubbo/codec/dubbo_codec.py b/src/dubbo/codec/dubbo_codec.py
@@ -0,0 +1,78 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     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,
+# 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.
+
+from abc import ABC, abstractmethod
+from typing import Any, Type, Optional, Callable
+from pydantic import BaseModel
+from dubbo.classes import CodecHelper
+
+class DubboCodec:
+    _codec_instance: Optional[Any] = None
+F438
+    @staticmethod
+    def init(codec_type: str = 'json', model_type: Optional[Type[BaseModel]] = None, **codec_kwargs):
+        """Initialize codec with specified type and options"""
+        if model_type is None:
+            raise ValueError("model_type is required for all codecs")
+
+        from dubbo.extension.extension_loader import ExtensionLoader
+
+        # Get the Codec class from CodecHelper
+        Codec = CodecHelper.get_class()
+
+        codec_class = ExtensionLoader().get_extension(Codec, codec_type)
+        DubboCodec._codec_instance = codec_class(model_type=model_type, **codec_kwargs)
+
+    @staticmethod
+    def get_instance():
+        if DubboCodec._codec_instance is None:
+            raise RuntimeError("DubboCodec is not initialized. Call DubboCodec.init(...) first.")
+        return DubboCodec._codec_instance
+
+    @staticmethod
+    def encode(data: Any) -> bytes:
+        return DubboCodec.get_instance().encode(data)
+
+    @staticmethod
+    def decode(data: bytes) -> Any:
+        return DubboCodec.get_instance().decode(data)
+
+    @staticmethod
+    def get_serializer_deserializer(
+        codec_type: str,
+        request_model: Type[BaseModel] = None,
+        response_model: Type[BaseModel] = None,
+    ) -> tuple[Callable, Callable]:
+        """Get serializer and deserializer functions for RPC"""
+
+        from dubbo.extension.extension_loader import ExtensionLoader
+
+        # Get the Codec class from CodecHelper
+        Codec = CodecHelper.get_class()
+
+        request_codec_class = ExtensionLoader().get_extension(Codec, codec_type)
+        request_codec = request_codec_class(model_type=request_model)
+
+        response_codec_class = ExtensionLoader().get_extension(Codec, codec_type)
+        response_codec = response_codec_class(model_type=response_model)
+
+        def request_deserializer(data: bytes):
+            return request_codec.decode(data)
+
+        def response_serializer(response):
+            return response_codec.encode(response)
+
+        return request_deserializer, response_serializer
diff --git a/src/dubbo/codec/json_codec/__init__.py b/src/dubbo/codec/json_codec/__init__.py
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     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,
+# 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.
+from .json_codec_handler import JsonCodec
+
+__all__ = ["JsonCodec"]
diff --git a/src/dubbo/codec/json_codec/json_codec_handler.py b/src/dubbo/codec/json_codec/json_codec_handler.py
@@ -0,0 +1,50 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     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,
+# 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.
+
+from abc import ABC, abstractmethod
+from typing import Any, Type, Protocol, Callable, TypeVar, Generic
+from functools import wraps
+from pydantic import BaseModel
+from dubbo.classes import Codec 
+import orjson
+
+class EncodingFunction(Protocol):
+    def __call__(self, obj: Any) -> bytes: ...
+
+class DecodingFunction(Protocol):
+    def __call__(self, data: bytes) -> Any: ...
+
+ModelT = TypeVar('ModelT', bound=BaseModel)
+
+class JsonCodec(Codec, Generic[ModelT]):
+    """JSON codec for Pydantic models using orjson for performance"""
+
+    def __init__(self, model_type: Type[ModelT], **kwargs):
+        super().__init__(model_type=model_type, **kwargs)  # Call parent constructor
+        self.model_type = model_type
+
+    def encode(self, data: Any) -> bytes:
+        """Encode data to JSON bytes"""
+        if isinstance(data, dict):
+            data = self.model_type(**data)
+        elif not isinstance(data, self.model_type):
+            raise TypeError(f"Expected {self.model_type.__name__} or dict, got {type(data).__name__}")
+        return orjson.dumps(data.model_dump())
+
+    def decode(self, data: bytes) -> ModelT:
+        """Decode JSON bytes to Pydantic model"""
+        json_data = orjson.loads(data)
+        return self.model_type(**json_data)
diff --git a/src/dubbo/extension/__init__.py b/src/dubbo/extension/__init__.py
@@ -16,6 +16,6 @@
 
 from dubbo.extension.extension_loader import ExtensionError, ExtensionLoader as _ExtensionLoader
 
-__all__ = ["ExtensionError", "extensionLoader"]
+__all__ = ["ExtensionError", "ExtensionLoader"]
 
 extensionLoader = _ExtensionLoader()
diff --git a/src/dubbo/extension/extension_loader.py b/src/dubbo/extension/extension_loader.py
@@ -48,12 +48,14 @@ def __init__(self):
         """
         if not hasattr(self, "_initialized"):  # Ensure __init__ runs only once
             self._registries = {}
+
             for name in registries_module.registries:
                 registry = getattr(registries_module, name)
                 self._registries[registry.interface] = registry.impls
             self._initialized = True
 
     def get_extension(self, interface: Any, impl_name: str) -> Any:
+
         """
         Get the extension implementation for the interface.
 

diff --git a/src/dubbo/extension/registries.py b/src/dubbo/extension/registries.py
@@ -22,7 +22,7 @@
 from dubbo.protocol import Protocol
 from dubbo.registry import RegistryFactory
 from dubbo.remoting import Transporter
-
+from dubbo.classes import Codec
 
 @dataclass
 class ExtendedRegistry:
@@ -47,6 +47,7 @@ class ExtendedRegistry:
     "compressorRegistry",
     "decompressorRegistry",
     "transporterRegistry",
+    "codecRegistry"
 ]
 
 # RegistryFactory registry
@@ -103,3 +104,11 @@ class ExtendedRegistry:
         "aio": "dubbo.remoting.aio.aio_transporter.AioTransporter",
     },
 )
+
+# Codec registry 
+codecRegistry = ExtendedRegistry(
+    interface=Codec, 
+    impls={
+        "json": "dubbo.codec.json_codec.JsonCodec",  
+    }
+)