docs: update some samples

Author: cnzakii

dubbo/client.py

@@ -133,7 +133,7 @@ def server_stream(
             )
         )
 
-    def bidi_stream(
+    def bi_stream(
         self,
         method_name: str,
         request_serializer: Optional[SerializingFunction] = None,

dubbo/protocol/triple/call/server_call.py

@@ -42,6 +42,7 @@
     DirectDeserializer,
     DirectSerializer,
 )
+from dubbo.types import RpcType, RpcTypes
 from dubbo.utils import FunctionHelper
 
 
@@ -169,18 +170,16 @@ def __init__(
         self,
         func: Callable,
         read_write_stream: ReadWriteStream,
-        client_stream: bool,
-        server_stream: bool,
+        rpc_type: RpcType,
     ):
         self._read_write_stream = read_write_stream
         self._func = func
 
-        self._is_client_stream = client_stream
-        self._is_server_stream = server_stream
+        self._rpc_type = rpc_type
 
     def run(self) -> None:
         try:
-            if not self._is_client_stream:
+            if self._rpc_type == RpcTypes.UNARY.value:
                 result = self._func(self._read_write_stream.read())
             else:
                 result = self._func(self._read_write_stream)
@@ -196,15 +195,15 @@ def handle_result(self, result: Any) -> None:
             if not self._read_write_stream.can_write_more():
                 return
 
-            if not self._is_server_stream:
+            if not self._rpc_type.server_stream:
                 # get single result
                 self._read_write_stream.write(result)
-                self._read_write_stream.done_writing()
             else:
                 # get multi results
                 for message in result:
                     self._read_write_stream.write(message)
-                self._read_write_stream.done_writing()
+
+            self._read_write_stream.done_writing()
         except Exception as e:
             self.handle_exception(e)
 
@@ -240,11 +239,9 @@ def create(
         """
 
         method_descriptor = method_handler.method_descriptor
-        rpc_type = method_descriptor.get_rpc_type()
 
         return DefaultMethodRunner(
             method_descriptor.get_method(),
             read_write_stream,
-            rpc_type.client_stream,
-            rpc_type.server_stream,
+            method_descriptor.get_rpc_type(),
         )

samples/serialization/README.md

@@ -13,7 +13,7 @@ SerializingFunction = Callable[..., bytes]
 
 For deserialization functions, we specify:
 
-```
+```python
 # A function that takes an argument of type bytes and returns data of any type
 DeserializingFunction = Callable[[bytes], Any]
 ```

samples/stream/README.md

@@ -1,70 +1,94 @@
 ## Streaming Calls
 
-Dubbo-python supports streaming calls, including `ClientStream`, `ServerStream`, and `BidirectionalStream`. The key difference in these calls is the use of iterators: passing an iterator as a parameter for `ClientStream`, receiving an iterator for `ServerStream`, or both passing and receiving iterators for `BidirectionalStream`.
+Dubbo-Python supports streaming calls, including `ClientStream`, `ServerStream`, and `BidirectionalStream` modes.
 
-When using `BidirectionalStream`, the client needs to pass an iterator as a parameter to send multiple data points, while also receiving an iterator to handle multiple responses from the server.
+Streaming calls can be divided into write-streams and read-streams. For `ClientStream`, it’s multiple writes with a single read; for `ServerStream`, a single write with multiple reads; and `BidirectionalStream` allows multiple writes and reads.
 
-Here’s an example of the client-side code:
+### Write-Stream
+
+Write operations in streaming calls can be divided into single write (`ServerStream`) and multiple writes (`ClientStream` and `BidirectionalStream`).
+
+#### Single Write
+
+Single write calls are similar to unary mode. For example:
 
 ```python
-class GreeterServiceStub:
-    def __init__(self, client: dubbo.Client):
-        self.bidi_stream = client.bidi_stream(
-            method_name="biStream",
-            request_serializer=greeter_pb2.GreeterRequest.SerializeToString,
-            response_deserializer=greeter_pb2.GreeterReply.FromString,
-        )
+stub.server_stream(greeter_pb2.GreeterRequest(name="hello world from dubbo-python"))
+```
 
-    def bi_stream(self, values):
-        return self.bidi_stream(values)
+#### Multiple Writes
 
+For multiple writes, users can write data using either an iterator or `writeStream` (only one of these options should be used).
 
-if __name__ == "__main__":
-    reference_config = ReferenceConfig.from_url(
-        "tri://127.0.0.1:50051/org.apache.dubbo.samples.proto.Greeter"
-    )
-    dubbo_client = dubbo.Client(reference_config)
+1. **Iterator-based Write**: Writing via iterator is similar to unary mode, with the main difference being the use of an iterator for multiple writes. For example:
 
-    stub = GreeterServiceStub(dubbo_client)
+   ```python
+   # Use an iterator to send multiple requests
+   def request_generator():
+       for i in ["hello", "world", "from", "dubbo-python"]:
+           yield greeter_pb2.GreeterRequest(name=str(i))
 
-    # Iterator of request
-    def request_generator():
-        for item in ["hello", "world", "from", "dubbo-python"]:
-            yield greeter_pb2.GreeterRequest(name=str(item))
+   # Call the remote method and return a read_stream
+   stream = stub.client_stream(request_generator())
+   ```
 
-    result = stub.bi_stream(request_generator())
+2. **Using `writeStream`**: This method requires an empty argument, after which data is written incrementally using `write`, and `done_writing` is called to end the write-stream. For example:
 
-    for i in result:
-        print(f"Received response: {i.message}")
-```
+   ```python
+   stream = stub.bi_stream()
+   # Use the write method to send messages
+   stream.write(greeter_pb2.GreeterRequest(name="jock"))
+   stream.write(greeter_pb2.GreeterRequest(name="jane"))
+   stream.write(greeter_pb2.GreeterRequest(name="alice"))
+   stream.write(greeter_pb2.GreeterRequest(name="dave"))
+   # Call done_writing to notify the server that the client has finished writing
+   stream.done_writing()
+   ```
 
-And here’s the server-side code:
+### Read-Stream
+
+Read operations for streaming calls can be single read (`ClientStream`) or multiple reads (`ServerStream` and `BidirectionalStream`). A `ReadStream` is returned in all cases, and data can be read using the `read` method or an iterator. When using `read`, please note:
+
+1. The `read` method supports a `timeout` parameter (in seconds).
+2. The `read` method can return one of three values: the expected data, `None` (timeout exceeded), or `EOF` (end of the read-stream).
+
+#### Single Read
+
+A single call to the `read` method will retrieve the data, for example:
 
 ```python
-def bi_stream(request_stream):
-    for request in request_stream:
-        print(f"Received message from {request.name}")
-        yield greeter_pb2.GreeterReply(message=request.name)
-
-
-if __name__ == "__main__":
-    # build a method handler
-    method_handler = RpcMethodHandler.bi_stream(
-        bi_stream,
-        request_deserializer=greeter_pb2.GreeterRequest.FromString,
-        response_serializer=greeter_pb2.GreeterReply.SerializeToString,
-    )
-    # build a service handler
-    service_handler = RpcServiceHandler(
-        service_name="org.apache.dubbo.samples.data.Greeter",
-        method_handlers={"biStream": method_handler},
-    )
-
-    service_config = ServiceConfig(service_handler)
-
-    # start the server
-    server = dubbo.Server(service_config).start()
-
-    input("Press Enter to stop the server...\n")
+result = stream.read()
+print(f"Received response: {result.message}")
 ```
 
+#### Multiple Reads
+
+Multiple reads can be done by repeatedly calling `read`, with handling for `None` and `EOF` values. Since `ReadStream` implements `__iter__` and `__next__`, an iterator-based approach can also be used, which automatically handles these values but doesn’t support a timeout.
+
+1. **Using Iterator (Recommended)**:
+
+   ```python
+   def client_stream(self, request_iterator):
+       response = ""
+       for request in request_iterator:
+           print(f"Received request: {request.name}")
+           response += f"{request.name} "
+       return greeter_pb2.GreeterReply(message=response)
+   ```
+
+2. **Multiple Calls to `read` Method**:
+
+   ```python
+   # Use read method to receive messages
+   # If no message arrives within the specified time, returns None
+   # If the server has finished sending messages, returns EOF
+   while True:
+       i = stream.read(timeout=0.5)
+       if i is dubbo.classes.EOF:
+           break
+       elif i is None:
+           print("No message received")
+           continue
+       print(f"Received response: {i.message}")
+   ```
+

samples/stream/bidi_stream/client.py

@@ -17,12 +17,12 @@
 import dubbo
 from dubbo.classes import EOF
 from dubbo.configs import ReferenceConfig
-from samples.data import greeter_pb2
+from samples.proto import greeter_pb2
 
 
 class GreeterServiceStub:
     def __init__(self, client: dubbo.Client):
-        self.bidi_stream = client.bidi_stream(
+        self.bidi_stream = client.bi_stream(
             method_name="biStream",
             request_serializer=greeter_pb2.GreeterRequest.SerializeToString,
             response_deserializer=greeter_pb2.GreeterReply.FromString,
@@ -33,42 +33,30 @@ def bi_stream(self, *args):
 
 
 if __name__ == "__main__":
+    # Create a reference config
     reference_config = ReferenceConfig.from_url(
         "tri://127.0.0.1:50051/org.apache.dubbo.samples.data.Greeter"
     )
     dubbo_client = dubbo.Client(reference_config)
-
     stub = GreeterServiceStub(dubbo_client)
 
-    # # Iterator of request
-    # def request_generator():
-    #     for item in ["hello", "world", "from", "dubbo-python"]:
-    #         yield greeter_pb2.GreeterRequest(name=str(item))
-    #
-    # result = stub.bi_stream(request_generator())
-    #
-    # for i in result:
-    #     print(f"Received response: {i.message}")
-
     stream = stub.bi_stream()
+    # use write method to send message
+    stream.write(greeter_pb2.GreeterRequest(name="jock"))
 
-    stream.write(greeter_pb2.GreeterRequest(name="hello"))
-
-    # print(f"Received response: {stream.read().message}")
+    # use read method to receive message
+    print(f"Received response: {stream.read().message}")
 
-    stream.write(greeter_pb2.GreeterRequest(name="world"))
-    stream.write(greeter_pb2.GreeterRequest(name="from"))
-    stream.write(greeter_pb2.GreeterRequest(name="dubbo-python"))
+    # continue to send message
+    stream.write(greeter_pb2.GreeterRequest(name="jane"))
+    stream.write(greeter_pb2.GreeterRequest(name="alice"))
+    stream.write(greeter_pb2.GreeterRequest(name="dave"))
+    # done_writing method must be called to notify the server that the client has finished writing
     stream.done_writing()
 
-    # 直接调用read方法
-    print(stream.read())
-
-    # 迭代器调用read方法（推荐）
-    for i in stream:
-        print(f"Received response: {i.message}")
-
-    # 循环调用read方法
+    # use read method to receive message
+    # If no message arrives within the specified time, returns None
+    # If the server has finished sending messages and the client has received all messages, returns EOF
     while True:
         i = stream.read(timeout=0.5)
         if i is EOF:

samples/stream/bidi_stream/server.py

@@ -16,32 +16,43 @@
 import dubbo
 from dubbo.configs import ServiceConfig
 from dubbo.proxy.handlers import RpcMethodHandler, RpcServiceHandler
-from samples.data import greeter_pb2
+from samples.proto import greeter_pb2
 
 import time
 
 
-def bi_stream(request_stream):
-    for request in request_stream:
-        print(f"Received message from {request.name}")
-        yield greeter_pb2.GreeterReply(message=request.name)
-        time.sleep(1)
+class GreeterServiceServicer:
+    def bi_stream(self, stream):
+        counter = 0
+        for request in stream:
+            print(f"Received request: {request.name}")
+            # simulate a long time to process
+            if counter == 1:
+                time.sleep(1)
+            counter += 1
 
+            stream.write(greeter_pb2.GreeterReply(message=f"Hello, {request.name}!"))
 
-if __name__ == "__main__":
-    # build a method handler
+        stream.done_writing()
+
+
+def build_server_handler():
     method_handler = RpcMethodHandler.bi_stream(
-        bi_stream,
+        GreeterServiceServicer().bi_stream,
         method_name="biStream",
         request_deserializer=greeter_pb2.GreeterRequest.FromString,
         response_serializer=greeter_pb2.GreeterReply.SerializeToString,
     )
-    # build a service handler
     service_handler = RpcServiceHandler(
         service_name="org.apache.dubbo.samples.data.Greeter",
         method_handlers=[method_handler],
     )
+    return service_handler
 
+
+if __name__ == "__main__":
+    # build a service config
+    service_handler = build_server_handler()
     service_config = ServiceConfig(service_handler, host="127.0.0.1", port=50051)
 
     # start the server

samples/stream/client_stream/client.py

@@ -15,7 +15,7 @@
 # limitations under the License.
 import dubbo
 from dubbo.configs import ReferenceConfig
-from samples.data import greeter_pb2
+from samples.proto import greeter_pb2
 
 
 class GreeterServiceStub:
@@ -26,31 +26,23 @@ def __init__(self, client: dubbo.Client):
             response_deserializer=greeter_pb2.GreeterReply.FromString,
         )
 
-    def client_stream(self, *args):
-        return self.unary_stream(args)
+    def client_stream(self, request_iterator):
+        return self.unary_stream(request_iterator)
 
 
 if __name__ == "__main__":
     reference_config = ReferenceConfig.from_url(
         "tri://127.0.0.1:50051/org.apache.dubbo.samples.data.Greeter"
     )
     dubbo_client = dubbo.Client(reference_config)
-
     stub = GreeterServiceStub(dubbo_client)
 
-    # 迭代器方法
+    # use iterator to send multiple requests
     def request_generator():
         for i in ["hello", "world", "from", "dubbo-python"]:
             yield greeter_pb2.GreeterRequest(name=str(i))
 
+    # call the remote method and return a read_stream
     stream = stub.client_stream(request_generator())
-    print(stream.read())
-
-    # stream 调用方法
-    stream = stub.client_stream()
-    for i in ["hello", "world", "from", "dubbo-python"]:
-        stream.write(greeter_pb2.GreeterRequest(name=str(i)))
-
-    stream.done_writing()
-
+    # use read method to get the response
     print(stream.read())