This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-website.git
The following commit(s) were added to refs/heads/master by this push:
new a394ceb44cd docs(golang-sdk): improve streaming documentation (#3227)
a394ceb44cd is described below
commit a394ceb44cdaa858b35721998ebf43745811e388
Author: Yuqi Qiao <[email protected]>
AuthorDate: Mon Aug 17 18:54:46 2026 +0800
docs(golang-sdk): improve streaming documentation (#3227)
* docs(golang-sdk): improve streaming documentation
* docs(golang-sdk): link header trailer sample
---
.../mannual/golang-sdk/tutorial/rpc/streaming.md | 376 ++++++++++++---------
.../mannual/golang-sdk/tutorial/rpc/streaming.md | 375 +++++++++++---------
2 files changed, 434 insertions(+), 317 deletions(-)
diff --git a/content/en/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
b/content/en/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
index 49a71f602df..971b64fbc04 100644
--- a/content/en/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
+++ b/content/en/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
@@ -1,5 +1,5 @@
---
-description: Streaming communication
+description: Streaming Communication
title: Streaming Communication
type: docs
weight: 1
@@ -7,27 +7,31 @@ weight: 1
Sample source: <a
href="https://github.com/apache/dubbo-go-samples/tree/main/streaming";
target="_blank">dubbo-go-samples/streaming</a>.
-Streaming communication is a new RPC data transfer mode offered by Dubbo3,
suitable for the following scenarios:
+A unary RPC has one request and one response. Triple streaming allows a single
RPC to carry a sequence of messages, which is useful for chunked data, server
push, and real-time communication.
-- Interfaces that need to send large amounts of data that cannot be placed in
a single RPC request or response, requiring batch sending. However, traditional
multiple RPC calls cannot resolve issues of order and performance, and if order
is to be guaranteed, they must be sent serially.
-- Streaming scenarios where data needs to be processed in the order sent, and
the data itself has no definite boundaries.
-- Push scenarios where multiple messages are sent and processed within the
same call context.
+## Streaming types
-There are three types of Streaming communication:
-- SERVER_STREAM (Server Stream)
-- CLIENT_STREAM (Client Stream)
-- BIDIRECTIONAL_STREAM (Bidirectional Stream)
+In a `.proto` method, the side marked with `stream` can send more than one
message:
-## 1. Introduction
+| Type | Requests | Responses | Definition |
+| --- | ---: | ---: | --- |
+| Server streaming | 1 | N | `rpc Method(Request) returns (stream Response)` |
+| Client streaming | N | 1 | `rpc Method(stream Request) returns (Response)` |
+| Bidirectional streaming | N | N | `rpc Method(stream Request) returns
(stream Response)` |
-This document demonstrates how to use streaming communication in Dubbo-go.
+Server streaming works well for event push and large downloads. Client
streaming is commonly used for chunked uploads or server-side aggregation. In a
bidirectional stream, sending and receiving are independent, making it suitable
for chat and real-time control. The sample echoes each request immediately, but
that is an application choice rather than a protocol requirement.
-## 2. How to use Dubbo-go streaming communication
+## Define the service
-In the proto file, add `stream` before the request or response type of the
methods that require streaming communication,
-and generate the corresponding files using `protoc-gen-go-triple`.
+The sample defines all three streaming methods in
[`streaming/proto/greet.proto`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/proto/greet.proto):
```protobuf
+syntax = "proto3";
+
+package greet;
+
+option go_package = "github.com/apache/dubbo-go-samples/streaming/proto;greet";
+
service GreetService {
rpc Greet(GreetRequest) returns (GreetResponse) {}
rpc GreetStream(stream GreetStreamRequest) returns (stream
GreetStreamResponse) {}
@@ -36,25 +40,52 @@ service GreetService {
}
```
-Write the server handler file.
+Messages are defined in the same way as for unary RPCs; `stream` on the method
signature provides the streaming semantics. After installing `protoc-gen-go`
and `protoc-gen-go-triple`, generate the Go code from the `streaming` directory:
+
+```bash
+protoc \
+ --go_out=. \
+ --go_opt=paths=source_relative \
+ --go-triple_out=. \
+ --go-triple_opt=paths=source_relative \
+ ./proto/greet.proto
+```
-Source file path: dubbo-go-sample/streaming/go-server/cmd/server.go
+This creates `greet.pb.go` for the messages and `greet.triple.go` for the
client, server, and registration APIs.
+
+## Server
+
+The complete server is in
[`streaming/go-server/cmd/server.go`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/go-server/cmd/server.go).
It listens on port `20000` and registers the generated `GreetService` handler:
```go
-type GreetTripleServer struct {
+srv, err := server.NewServer(
+ server.WithServerProtocol(protocol.WithPort(20000)),
+)
+if err != nil {
+ panic(err)
}
-
-func (srv *GreetTripleServer) Greet(ctx context.Context, req
*greet.GreetRequest) (*greet.GreetResponse, error) {
- resp := &greet.GreetResponse{Greeting: req.Name}
- return resp, nil
+if err := greet.RegisterGreetServiceHandler(srv, &GreetTripleServer{}); err !=
nil {
+ panic(err)
}
+if err := srv.Serve(); err != nil {
+ logger.Error(err)
+}
+```
+
+### Bidirectional streaming
-func (srv *GreetTripleServer) GreetStream(ctx context.Context, stream
greet.GreetService_GreetStreamServer) error {
+`Recv()` reads a request and `Send()` writes a response. Once the client
closes its request side, `Recv()` reports the end of the stream.
`triple.IsEnded` distinguishes that normal condition from a transport error.
+
+```go
+func (srv *GreetTripleServer) GreetStream(
+ ctx context.Context,
+ stream greet.GreetService_GreetStreamServer,
+) error {
for {
req, err := stream.Recv()
if err != nil {
if triple.IsEnded(err) {
- break
+ return nil
}
return fmt.Errorf("triple BidiStream recv error: %s",
err)
}
@@ -62,89 +93,74 @@ func (srv *GreetTripleServer) GreetStream(ctx
context.Context, stream greet.Gree
return fmt.Errorf("triple BidiStream send error: %s",
err)
}
}
- return nil
}
+```
-func (srv *GreetTripleServer) GreetClientStream(ctx context.Context, stream
greet.GreetService_GreetClientStreamServer) (*greet.GreetClientStreamResponse,
error) {
- var reqs []string
+### Client streaming
+
+The server reads all requests, joins their names, and returns one response.
When using the iterator-style `Recv()`, check `Err()` after the loop.
+
+```go
+func (srv *GreetTripleServer) GreetClientStream(
+ ctx context.Context,
+ stream greet.GreetService_GreetClientStreamServer,
+) (*greet.GreetClientStreamResponse, error) {
+ var names []string
for stream.Recv() {
- reqs = append(reqs, stream.Msg().Name)
+ names = append(names, stream.Msg().Name)
}
if stream.Err() != nil && !triple.IsEnded(stream.Err()) {
- return nil, fmt.Errorf("triple ClientStream recv err: %s",
stream.Err())
- }
- resp := &greet.GreetClientStreamResponse{
- Greeting: strings.Join(reqs, ","),
+ return nil, fmt.Errorf("triple ClientStream recv error: %s",
stream.Err())
}
-
- return resp, nil
+ return &greet.GreetClientStreamResponse{
+ Greeting: strings.Join(names, ","),
+ }, nil
}
+```
+
+### Server streaming
+
+A server-streaming handler receives one request but may call `Send()` more
than once. Returning from the handler ends the response stream.
-func (srv *GreetTripleServer) GreetServerStream(ctx context.Context, req
*greet.GreetServerStreamRequest, stream
greet.GreetService_GreetServerStreamServer) error {
- for i := 0; i < 5; i++ {
+```go
+func (srv *GreetTripleServer) GreetServerStream(
+ ctx context.Context,
+ req *greet.GreetServerStreamRequest,
+ stream greet.GreetService_GreetServerStreamServer,
+) error {
+ for i := 0; i < 10; i++ {
if err :=
stream.Send(&greet.GreetServerStreamResponse{Greeting: req.Name}); err != nil {
- return fmt.Errorf("triple ServerStream send err: %s",
err)
+ return fmt.Errorf("triple ServerStream send error: %s",
err)
}
}
return nil
}
```
-Write the client file.
+## Client
-Source file path: dubbo-go-sample/streaming/go-client/cmd/client.go
+The complete client is in
[`streaming/go-client/cmd/client.go`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/go-client/cmd/client.go).
Connect to the server and create the generated proxy before opening a stream:
```go
-func main() {
- cli, err := client.NewClient(
- client.WithClientURL("tri://127.0.0.1:20000"),
- )
- if err != nil {
- panic(err)
- }
-
- svc, err := greet.NewGreetService(cli)
- if err != nil {
- panic(err)
- }
- TestClient(svc)
+cli, err := client.NewClient(client.WithClientURL("tri://127.0.0.1:20000"))
+if err != nil {
+ panic(err)
}
-
-func TestClient(cli greet.GreetService) {
- if err := testUnary(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testBidiStream(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testClientStream(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testServerStream(cli); err != nil {
- logger.Error(err)
- }
+svc, err := greet.NewGreetService(cli)
+if err != nil {
+ panic(err)
}
+```
-func testUnary(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE unary call")
- resp, err := cli.Greet(context.Background(), &greet.GreetRequest{Name:
"triple"})
- if err != nil {
- return err
- }
- logger.Infof("TRIPLE unary call resp: %s", resp.Greeting)
- return nil
-}
+For a bidirectional stream, the client can alternate between `Send()` and
`Recv()`, as below, or run them concurrently in separate goroutines:
-func testBidiStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE bidi stream")
- stream, err := cli.GreetStream(context.Background())
- if err != nil {
- return err
- }
- if sendErr := stream.Send(&greet.GreetStreamRequest{Name: "triple"});
sendErr != nil {
+```go
+stream, err := svc.GreetStream(context.Background())
+if err != nil {
+ return err
+}
+for _, name := range []string{"triple-1", "triple-2", "triple-3"} {
+ if err := stream.Send(&greet.GreetStreamRequest{Name: name}); err !=
nil {
return err
}
resp, err := stream.Recv()
@@ -152,104 +168,146 @@ func testBidiStream(cli greet.GreetService) error {
return err
}
logger.Infof("TRIPLE bidi stream resp: %s", resp.Greeting)
- if err := stream.CloseRequest(); err != nil {
- return err
- }
- if err := stream.CloseResponse(); err != nil {
- return err
- }
- return nil
}
+if err := stream.CloseRequest(); err != nil {
+ return err
+}
+if err := stream.CloseResponse(); err != nil {
+ return err
+}
+```
-func testClientStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE client stream")
- stream, err := cli.GreetClientStream(context.Background())
- if err != nil {
- return err
- }
- for i := 0; i < 5; i++ {
- if sendErr := stream.Send(&greet.GreetClientStreamRequest{Name:
"triple"}); sendErr != nil {
- return err
- }
- }
- resp, err := stream.CloseAndRecv()
- if err != nil {
+For client streaming, `CloseAndRecv()` finishes sending and waits for the
single response:
+
+```go
+stream, err := svc.GreetClientStream(context.Background())
+if err != nil {
+ return err
+}
+for i := 0; i < 5; i++ {
+ if err := stream.Send(&greet.GreetClientStreamRequest{Name: "triple"});
err != nil {
return err
}
- logger.Infof("TRIPLE client stream resp: %s", resp.Greeting)
- return nil
}
+resp, err := stream.CloseAndRecv()
+if err != nil {
+ return err
+}
+logger.Infof("TRIPLE client stream resp: %s", resp.Greeting)
+```
-func testServerStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE server stream")
- stream, err := cli.GreetServerStream(context.Background(),
&greet.GreetServerStreamRequest{Name: "triple"})
- if err != nil {
- return err
- }
- for stream.Recv() {
- logger.Infof("TRIPLE server stream resp: %s",
stream.Msg().Greeting)
- }
- if stream.Err() != nil {
- return err
- }
- if err := stream.Close(); err != nil {
- return err
- }
- return nil
+A server-streaming request is sent when the stream is opened. The client then
reads responses until the stream ends:
+
+```go
+stream, err := svc.GreetServerStream(
+ context.Background(),
+ &greet.GreetServerStreamRequest{Name: "triple"},
+)
+if err != nil {
+ return err
+}
+for stream.Recv() {
+ logger.Infof("TRIPLE server stream resp: %s", stream.Msg().Greeting)
+}
+if err := stream.Err(); err != nil {
+ return err
+}
+if err := stream.Close(); err != nil {
+ return err
}
```
-Streaming calls also expose Triple metadata through the generated stream APIs:
+## Closing a stream
+
+The request and response sides of a bidirectional stream can be closed
separately:
+
+| API | When to use it |
+| --- | --- |
+| `CloseRequest()` | The client has finished sending. It may continue to
receive responses afterward. |
+| `CloseResponse()` | The client has read all responses, or no longer needs
the remaining responses, and wants to release the stream resources. |
+| `CloseAndRecv()` | Client streaming only: close the request side and wait
for the server's single response. |
+
+If the server sends more data after reading the final request, call
`CloseRequest()`, keep calling `Recv()` until the response ends, and then call
`CloseResponse()`.
-| Side | API | Purpose |
-| ---- | --- | ------- |
-| Client stream and bidirectional client | `RequestHeader()` | Add request
metadata before the first message is sent |
-| Server stream and bidirectional client | `ResponseHeader()` | Read response
headers returned by the server |
-| Server stream and bidirectional client | `ResponseTrailer()` | Read response
trailers after the response is completed |
-| Provider stream handlers | `RequestHeader()` | Read request metadata sent by
the client |
-| Provider stream handlers | `ResponseHeader()` / `ResponseTrailer()` | Set
response headers and trailers |
+## Metadata, headers, and trailers
-Use standard `http.Header` methods such as `Set`, `Add`, `Get`, and `Values`
when working with these metadata values.
+Stream metadata is represented by `http.Header`. Request metadata is written
to the request header. A server can use the response header for information
available near the start of a call and the trailer for its final status.
-For example, on the client side you can write request metadata before sending
the first message:
+For a complete header and trailer example, see
[`triple_header_trailer`](https://github.com/apache/dubbo-go-samples/tree/main/triple_header_trailer).
+
+For example, the following excerpt adds metadata to the bidirectional handler
and shows only the metadata-related code:
```go
-stream, err := cli.GreetStream(context.Background())
+requestID := stream.RequestHeader().Get("x-request-id")
+stream.ResponseHeader().Set("x-stream-status", "ready")
+stream.ResponseTrailer().Set("x-stream-result", "completed")
+
+logger.Infof("request id: %s", requestID)
+```
+
+Set request headers before the first `Send()`. Read response headers after the
first response arrives, and read trailers after the response stream ends.
+
+```go
+stream, err := svc.GreetStream(context.Background())
if err != nil {
return err
}
-stream.RequestHeader().Set("x-sample-token", "demo-token")
-```
+stream.RequestHeader().Set("x-request-id", "stream-demo-001")
-And on the server side you can return response metadata from a stream handler:
+if err := stream.Send(&greet.GreetStreamRequest{Name: "triple"}); err != nil {
+ return err
+}
+resp, err := stream.Recv()
+if err != nil {
+ return err
+}
+logger.Infof("response: %s", resp.Greeting)
+logger.Infof("header: %s", stream.ResponseHeader().Get("x-stream-status"))
-```go
-func (srv *GreetTripleServer) GreetServerStream(
- ctx context.Context,
- req *greet.GreetServerStreamRequest,
- stream greet.GreetService_GreetServerStreamServer,
-) error {
- stream.ResponseHeader().Set("x-stream-header", "ready")
- stream.ResponseTrailer().Set("x-stream-trailer", "done")
- return nil
+if err := stream.CloseRequest(); err != nil {
+ return err
+}
+if _, err := stream.Recv(); err != nil && !triple.IsEnded(err) {
+ return err
+}
+logger.Infof("trailer: %s", stream.ResponseTrailer().Get("x-stream-result"))
+if err := stream.CloseResponse(); err != nil {
+ return err
}
```
-## 3. Running Effect
+Metadata names beginning with `Triple-` or `Grpc-` are reserved by the
protocol and should not be written by applications.
+
+## Run the sample
-Run the server and client, and you will see the requests return normally.
+In an existing `dubbo-go-samples` checkout, enter the `streaming` directory
and start the server:
+```bash
+cd dubbo-go-samples/streaming
+go run ./go-server/cmd/server.go
```
-[start to test TRIPLE unary call]
-TRIPLE unary call resp: [triple]
-[start to test TRIPLE bidi stream]
-TRIPLE bidi stream resp: [triple]
-[start to test TRIPLE client stream]
-TRIPLE client stream resp: [triple,triple,triple,triple,triple]
-[start to test TRIPLE server stream]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
+
+Open another terminal, re-enter the `streaming` directory, and start the
client:
+
+```bash
+cd dubbo-go-samples/streaming
+go run ./go-client/cmd/client.go
+```
+
+The client runs the unary, bidirectional, client-streaming, and
server-streaming calls in order. Ignoring log prefixes, the relevant output is:
+
+```text
+start to test TRIPLE unary call
+TRIPLE unary call resp: triple
+start to test TRIPLE bidi stream
+TRIPLE bidi stream resp: triple-1
+TRIPLE bidi stream resp: triple-2
+TRIPLE bidi stream resp: triple-3
+start to test TRIPLE client stream
+TRIPLE client stream resp: triple,triple,triple,triple,triple
+start to test TRIPLE server stream
+TRIPLE server stream resp #1: triple
+TRIPLE server stream resp #2: triple
+...
+TRIPLE server stream resp #10: triple
```
diff --git
a/content/zh-cn/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
b/content/zh-cn/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
index 1afad8cc4ec..6c417b1df50 100644
--- a/content/zh-cn/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
+++ b/content/zh-cn/overview/mannual/golang-sdk/tutorial/rpc/streaming.md
@@ -4,29 +4,34 @@ title: 流式通信
type: docs
weight: 1
---
+
示例源码：<a href="https://github.com/apache/dubbo-go-samples/tree/main/streaming";
target="_blank">dubbo-go-samples/streaming</a>。
-Streaming 流式通信是 Dubbo3 新提供的一种 RPC 数据传输模式，适用于以下场景:
+普通 RPC 只有一个请求和一个响应。Triple 流式调用允许在一次 RPC 中连续收发多条消息，适合文件分片、消息推送和实时交互等场景。
-- 接口需要发送大量数据，这些数据无法被放在一个 RPC 的请求或响应中，需要分批发送，但应用层如果按照传统的多次 RPC
方式无法解决顺序和性能的问题，如果需要保证有序，则只能串行发送
-- 流式场景，数据需要按照发送顺序处理, 数据本身是没有确定边界的
-- 推送类场景，多个消息在同一个调用的上下文中被发送和处理
+## 流式调用的类型
-Streaming 流式通信类型分为以下三种:
-- SERVER_STREAM(服务端流)
-- CLIENT_STREAM(客户端流)
-- BIDIRECTIONAL_STREAM(双向流)
+在 `.proto` 中，`stream` 写在哪一侧，就表示哪一侧可以发送多条消息：
-## 1.介绍
+| 类型 | 请求数 | 响应数 | 定义形式 |
+| --- | ---: | ---: | --- |
+| 服务端流（server streaming） | 1 | N | `rpc Method(Request) returns (stream
Response)` |
+| 客户端流（client streaming） | N | 1 | `rpc Method(stream Request) returns
(Response)` |
+| 双向流（bidirectional streaming） | N | N | `rpc Method(stream Request) returns
(stream Response)` |
-本文档演示如何在 Dubbo-go 中使用流式通信。
+服务端流常用于消息推送和大文件下载；客户端流适合分片上传或批量汇总；双向流的发送和接收彼此独立，适合聊天、实时控制等长连接场景。示例中的双向流采用 echo
方式逐条返回消息，这只是业务实现，并不是协议限制。
-## 2.如何使用Dubbo-go流式通信
+## 定义接口
-在 proto 文件中，为需要流式通信的方法请求类型或响应类型添加 `stream`，然后使用 `protoc-gen-go-triple`
-生成对应代码。
+示例在
[`streaming/proto/greet.proto`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/proto/greet.proto)
中同时定义了三种流式方法：
```protobuf
+syntax = "proto3";
+
+package greet;
+
+option go_package = "github.com/apache/dubbo-go-samples/streaming/proto;greet";
+
service GreetService {
rpc Greet(GreetRequest) returns (GreetResponse) {}
rpc GreetStream(stream GreetStreamRequest) returns (stream
GreetStreamResponse) {}
@@ -35,25 +40,52 @@ service GreetService {
}
```
-编写服务端handler文件
+消息类型的写法与普通 RPC 相同，流式语义由方法签名中的 `stream` 决定。安装 `protoc-gen-go` 和
`protoc-gen-go-triple` 后，可以在 `streaming` 目录生成 Go 代码：
-源文件路径: dubbo-go-sample/streaming/go-server/cmd/server.go
+```bash
+protoc \
+ --go_out=. \
+ --go_opt=paths=source_relative \
+ --go-triple_out=. \
+ --go-triple_opt=paths=source_relative \
+ ./proto/greet.proto
+```
+
+生成的 `greet.pb.go` 包含消息类型，`greet.triple.go` 包含客户端、服务端接口和注册函数。
+
+## 服务端
+
+完整代码见
[`streaming/go-server/cmd/server.go`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/go-server/cmd/server.go)。服务端监听
`20000` 端口，并注册生成的 `GreetService` 处理函数：
```go
-type GreetTripleServer struct {
+srv, err := server.NewServer(
+ server.WithServerProtocol(protocol.WithPort(20000)),
+)
+if err != nil {
+ panic(err)
}
-
-func (srv *GreetTripleServer) Greet(ctx context.Context, req
*greet.GreetRequest) (*greet.GreetResponse, error) {
- resp := &greet.GreetResponse{Greeting: req.Name}
- return resp, nil
+if err := greet.RegisterGreetServiceHandler(srv, &GreetTripleServer{}); err !=
nil {
+ panic(err)
+}
+if err := srv.Serve(); err != nil {
+ logger.Error(err)
}
+```
+
+### 双向流
+
+`Recv()` 接收一条请求，`Send()` 返回一条响应。客户端关闭请求侧后，`Recv()` 会返回流结束错误；这里用
`triple.IsEnded` 将正常结束与传输错误区分开。
-func (srv *GreetTripleServer) GreetStream(ctx context.Context, stream
greet.GreetService_GreetStreamServer) error {
+```go
+func (srv *GreetTripleServer) GreetStream(
+ ctx context.Context,
+ stream greet.GreetService_GreetStreamServer,
+) error {
for {
req, err := stream.Recv()
if err != nil {
if triple.IsEnded(err) {
- break
+ return nil
}
return fmt.Errorf("triple BidiStream recv error: %s",
err)
}
@@ -61,89 +93,74 @@ func (srv *GreetTripleServer) GreetStream(ctx
context.Context, stream greet.Gree
return fmt.Errorf("triple BidiStream send error: %s",
err)
}
}
- return nil
}
+```
+
+### 客户端流
+
+服务端读完所有请求后，将名字拼接成一个响应。使用迭代式 `Recv()` 时，应在循环结束后检查 `Err()`。
-func (srv *GreetTripleServer) GreetClientStream(ctx context.Context, stream
greet.GreetService_GreetClientStreamServer) (*greet.GreetClientStreamResponse,
error) {
- var reqs []string
+```go
+func (srv *GreetTripleServer) GreetClientStream(
+ ctx context.Context,
+ stream greet.GreetService_GreetClientStreamServer,
+) (*greet.GreetClientStreamResponse, error) {
+ var names []string
for stream.Recv() {
- reqs = append(reqs, stream.Msg().Name)
+ names = append(names, stream.Msg().Name)
}
if stream.Err() != nil && !triple.IsEnded(stream.Err()) {
- return nil, fmt.Errorf("triple ClientStream recv err: %s",
stream.Err())
+ return nil, fmt.Errorf("triple ClientStream recv error: %s",
stream.Err())
}
- resp := &greet.GreetClientStreamResponse{
- Greeting: strings.Join(reqs, ","),
- }
-
- return resp, nil
+ return &greet.GreetClientStreamResponse{
+ Greeting: strings.Join(names, ","),
+ }, nil
}
+```
+
+### 服务端流
-func (srv *GreetTripleServer) GreetServerStream(ctx context.Context, req
*greet.GreetServerStreamRequest, stream
greet.GreetService_GreetServerStreamServer) error {
- for i := 0; i < 5; i++ {
+服务端流只接收一个请求，但可以多次调用 `Send()`。处理函数返回后，响应流随之结束。
+
+```go
+func (srv *GreetTripleServer) GreetServerStream(
+ ctx context.Context,
+ req *greet.GreetServerStreamRequest,
+ stream greet.GreetService_GreetServerStreamServer,
+) error {
+ for i := 0; i < 10; i++ {
if err :=
stream.Send(&greet.GreetServerStreamResponse{Greeting: req.Name}); err != nil {
- return fmt.Errorf("triple ServerStream send err: %s",
err)
+ return fmt.Errorf("triple ServerStream send error: %s",
err)
}
}
return nil
}
```
-编写客户端client文件
+## 客户端
-源文件路径: dubbo-go-sample/streaming/go-client/cmd/client.go
+完整代码见
[`streaming/go-client/cmd/client.go`](https://github.com/apache/dubbo-go-samples/blob/main/streaming/go-client/cmd/client.go)。连接服务端并创建代理后，就可以调用生成的流式方法：
```go
-func main() {
- cli, err := client.NewClient(
- client.WithClientURL("tri://127.0.0.1:20000"),
- )
- if err != nil {
- panic(err)
- }
-
- svc, err := greet.NewGreetService(cli)
- if err != nil {
- panic(err)
- }
- TestClient(svc)
+cli, err := client.NewClient(client.WithClientURL("tri://127.0.0.1:20000"))
+if err != nil {
+ panic(err)
}
-
-func TestClient(cli greet.GreetService) {
- if err := testUnary(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testBidiStream(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testClientStream(cli); err != nil {
- logger.Error(err)
- }
-
- if err := testServerStream(cli); err != nil {
- logger.Error(err)
- }
+svc, err := greet.NewGreetService(cli)
+if err != nil {
+ panic(err)
}
+```
-func testUnary(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE unary call")
- resp, err := cli.Greet(context.Background(), &greet.GreetRequest{Name:
"triple"})
- if err != nil {
- return err
- }
- logger.Infof("TRIPLE unary call resp: %s", resp.Greeting)
- return nil
-}
+双向流可以交替调用 `Send()` 和 `Recv()`，也可以在不同 goroutine 中并发收发：
-func testBidiStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE bidi stream")
- stream, err := cli.GreetStream(context.Background())
- if err != nil {
- return err
- }
- if sendErr := stream.Send(&greet.GreetStreamRequest{Name: "triple"});
sendErr != nil {
+```go
+stream, err := svc.GreetStream(context.Background())
+if err != nil {
+ return err
+}
+for _, name := range []string{"triple-1", "triple-2", "triple-3"} {
+ if err := stream.Send(&greet.GreetStreamRequest{Name: name}); err !=
nil {
return err
}
resp, err := stream.Recv()
@@ -151,104 +168,146 @@ func testBidiStream(cli greet.GreetService) error {
return err
}
logger.Infof("TRIPLE bidi stream resp: %s", resp.Greeting)
- if err := stream.CloseRequest(); err != nil {
- return err
- }
- if err := stream.CloseResponse(); err != nil {
- return err
- }
- return nil
}
+if err := stream.CloseRequest(); err != nil {
+ return err
+}
+if err := stream.CloseResponse(); err != nil {
+ return err
+}
+```
-func testClientStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE client stream")
- stream, err := cli.GreetClientStream(context.Background())
- if err != nil {
- return err
- }
- for i := 0; i < 5; i++ {
- if sendErr := stream.Send(&greet.GreetClientStreamRequest{Name:
"triple"}); sendErr != nil {
- return err
- }
- }
- resp, err := stream.CloseAndRecv()
- if err != nil {
+客户端流发送完所有请求后，通过 `CloseAndRecv()` 结束发送并等待唯一的响应：
+
+```go
+stream, err := svc.GreetClientStream(context.Background())
+if err != nil {
+ return err
+}
+for i := 0; i < 5; i++ {
+ if err := stream.Send(&greet.GreetClientStreamRequest{Name: "triple"});
err != nil {
return err
}
- logger.Infof("TRIPLE client stream resp: %s", resp.Greeting)
- return nil
}
+resp, err := stream.CloseAndRecv()
+if err != nil {
+ return err
+}
+logger.Infof("TRIPLE client stream resp: %s", resp.Greeting)
+```
-func testServerStream(cli greet.GreetService) error {
- logger.Info("start to test TRIPLE server stream")
- stream, err := cli.GreetServerStream(context.Background(),
&greet.GreetServerStreamRequest{Name: "triple"})
- if err != nil {
- return err
- }
- for stream.Recv() {
- logger.Infof("TRIPLE server stream resp: %s",
stream.Msg().Greeting)
- }
- if stream.Err() != nil {
- return err
- }
- if err := stream.Close(); err != nil {
- return err
- }
- return nil
+服务端流在创建时发送请求，之后持续读取响应：
+
+```go
+stream, err := svc.GreetServerStream(
+ context.Background(),
+ &greet.GreetServerStreamRequest{Name: "triple"},
+)
+if err != nil {
+ return err
+}
+for stream.Recv() {
+ logger.Infof("TRIPLE server stream resp: %s", stream.Msg().Greeting)
+}
+if err := stream.Err(); err != nil {
+ return err
+}
+if err := stream.Close(); err != nil {
+ return err
}
```
-流式调用也可以通过生成的 stream API 读写 Triple metadata：
+## 关闭流
+
+双向流的请求和响应可以分别关闭：
+
+| API | 使用场景 |
+| --- | --- |
+| `CloseRequest()` | 客户端发送完毕后关闭请求侧。调用后仍可继续接收响应。 |
+| `CloseResponse()` | 客户端读取完响应，或确定不再需要后续响应时，关闭响应侧并释放资源。 |
+| `CloseAndRecv()` | 用于客户端流：关闭请求侧，并等待服务端返回唯一响应。 |
+
+如果服务端在读完请求后还会继续发送数据，双向流客户端应先调用 `CloseRequest()`，继续 `Recv()` 直到响应结束，再调用
`CloseResponse()`。
-| 位置 | API | 用途 |
-| ---- | --- | ---- |
-| 客户端流和双向流 client | `RequestHeader()` | 在发送第一条消息前写入请求 metadata |
-| 服务端流和双向流 client | `ResponseHeader()` | 读取服务端返回的响应 header |
-| 服务端流和双向流 client | `ResponseTrailer()` | 响应结束后读取服务端返回的 trailer |
-| Provider 侧 stream handler | `RequestHeader()` | 读取客户端发送的请求 metadata |
-| Provider 侧 stream handler | `ResponseHeader()` / `ResponseTrailer()` | 设置响应
header 和 trailer |
+## Metadata、header 和 trailer
-这些 metadata 使用 `http.Header` 表示，可以用 `Set`、`Add`、`Get`、`Values` 等标准方法读写。
+流上的 metadata 使用 `http.Header` 表示。请求信息写入 request header；服务端可在 response header
中返回调用开始阶段的信息，在 trailer 中返回最终状态。
-例如，在 client 侧可以在发送第一条消息前写入请求 metadata：
+完整的 header 和 trailer 使用示例请参考
[`triple_header_trailer`](https://github.com/apache/dubbo-go-samples/tree/main/triple_header_trailer)。
+
+下面在双向流处理函数中加入一个简单示例，只展示与 metadata 有关的代码：
```go
-stream, err := cli.GreetStream(context.Background())
+requestID := stream.RequestHeader().Get("x-request-id")
+stream.ResponseHeader().Set("x-stream-status", "ready")
+stream.ResponseTrailer().Set("x-stream-result", "completed")
+
+logger.Infof("request id: %s", requestID)
+```
+
+客户端必须在第一次 `Send()` 前设置 request header。Response header 在收到首条响应后即可读取；trailer
要等响应流结束后再读。
+
+```go
+stream, err := svc.GreetStream(context.Background())
if err != nil {
return err
}
-stream.RequestHeader().Set("x-sample-token", "demo-token")
-```
+stream.RequestHeader().Set("x-request-id", "stream-demo-001")
-在服务端 stream handler 中，也可以返回响应 metadata：
+if err := stream.Send(&greet.GreetStreamRequest{Name: "triple"}); err != nil {
+ return err
+}
+resp, err := stream.Recv()
+if err != nil {
+ return err
+}
+logger.Infof("response: %s", resp.Greeting)
+logger.Infof("header: %s", stream.ResponseHeader().Get("x-stream-status"))
-```go
-func (srv *GreetTripleServer) GreetServerStream(
- ctx context.Context,
- req *greet.GreetServerStreamRequest,
- stream greet.GreetService_GreetServerStreamServer,
-) error {
- stream.ResponseHeader().Set("x-stream-header", "ready")
- stream.ResponseTrailer().Set("x-stream-trailer", "done")
- return nil
+if err := stream.CloseRequest(); err != nil {
+ return err
+}
+if _, err := stream.Recv(); err != nil && !triple.IsEnded(err) {
+ return err
+}
+logger.Infof("trailer: %s", stream.ResponseTrailer().Get("x-stream-result"))
+if err := stream.CloseResponse(); err != nil {
+ return err
}
```
-## 3.运行效果
+`Triple-` 和 `Grpc-` 开头的 metadata 名称由协议保留，应用不应自行写入。
+
+## 运行示例
-运行服务端和客户端，可以看到请求正常返回
+在已经下载的 `dubbo-go-samples` 仓库中，进入 `streaming` 目录并启动服务端：
+```bash
+cd dubbo-go-samples/streaming
+go run ./go-server/cmd/server.go
```
-[start to test TRIPLE unary call]
-TRIPLE unary call resp: [triple]
-[start to test TRIPLE bidi stream]
-TRIPLE bidi stream resp: [triple]
-[start to test TRIPLE client stream]
-TRIPLE client stream resp: [triple,triple,triple,triple,triple]
-[start to test TRIPLE server stream]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
-TRIPLE server stream resp: [triple]
+
+打开另一个终端，重新进入 `streaming` 目录并启动客户端：
+
+```bash
+cd dubbo-go-samples/streaming
+go run ./go-client/cmd/client.go
+```
+
+客户端会依次执行普通调用、双向流、客户端流和服务端流。忽略日志前缀后，关键输出如下：
+
+```text
+start to test TRIPLE unary call
+TRIPLE unary call resp: triple
+start to test TRIPLE bidi stream
+TRIPLE bidi stream resp: triple-1
+TRIPLE bidi stream resp: triple-2
+TRIPLE bidi stream resp: triple-3
+start to test TRIPLE client stream
+TRIPLE client stream resp: triple,triple,triple,triple,triple
+start to test TRIPLE server stream
+TRIPLE server stream resp #1: triple
+TRIPLE server stream resp #2: triple
+...
+TRIPLE server stream resp #10: triple
```
