This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 27de37524 chore(triple): add doc comments and boundary tests for
stream APIs (#3663)
27de37524 is described below
commit 27de37524f62147e756467aef03aa162a422e664
Author: Li Zining <[email protected]>
AuthorDate: Sun Aug 16 08:12:52 2026 +0800
chore(triple): add doc comments and boundary tests for stream APIs (#3663)
* docs(triple): add doc comments for stream exported APIs
Document the close paths, error propagation, stop semantics, nil msg
handling and metadata behavior of the stream exported APIs, and add a doc
comment for NewCompatStreamHandler which had none. Fix the misleading nil
msg comment on ServerStream.Send and BidiStream.Send: it commits the
response headers without writing a message frame.
Signed-off-by: lizining <[email protected]>
* test(triple): add boundary tests for stream close and error propagation
Add 20 tests covering stream close, repeated close, error propagation and
empty metadata, using two minimal stubs that drive errors and count close
calls without a real transport.
Signed-off-by: lizining <[email protected]>
* docs(triple): clarify CloseAndReceive close is best-effort
Signed-off-by: lizining <[email protected]>
* test(triple): implement Spec and Peer on the streaming client stub
Signed-off-by: lizining <[email protected]>
---------
Signed-off-by: lizining <[email protected]>
---
protocol/triple/triple_protocol/client_stream.go | 15 +-
.../triple/triple_protocol/client_stream_test.go | 198 +++++++++++++++++++++
protocol/triple/triple_protocol/handler_stream.go | 18 +-
.../triple_protocol/handler_stream_compat.go | 10 ++
.../triple/triple_protocol/handler_stream_test.go | 158 ++++++++++++++++
5 files changed, 390 insertions(+), 9 deletions(-)
diff --git a/protocol/triple/triple_protocol/client_stream.go
b/protocol/triple/triple_protocol/client_stream.go
index 9b218c648..4da972aae 100644
--- a/protocol/triple/triple_protocol/client_stream.go
+++ b/protocol/triple/triple_protocol/client_stream.go
@@ -72,7 +72,8 @@ func (c *ClientStreamForClient) Send(request any) error {
}
// CloseAndReceive closes the send side of the stream and waits for the
-// response.
+// response. If closing the send side or receiving the response fails,
+// CloseResponse is called on a best-effort basis before the error is returned.
func (c *ClientStreamForClient) CloseAndReceive(response *Response) error {
if c.err != nil {
return c.err
@@ -111,7 +112,8 @@ type ServerStreamForClient struct {
// available through the Msg method. It returns false when the stream stops,
// either by reaching the end or by encountering an unexpected error. After
// Receive returns false, the Err method will return any unexpected error
-// encountered.
+// encountered. Once Receive returns false, subsequent calls return false
+// without advancing the stream.
// todo(DMwangnima): add classic usage
func (s *ServerStreamForClient) Receive(msg any) bool {
if s.constructErr != nil || s.receiveErr != nil {
@@ -122,7 +124,9 @@ func (s *ServerStreamForClient) Receive(msg any) bool {
return s.receiveErr == nil
}
-// Msg returns the most recent message unmarshaled by a call to Receive.
+// Msg returns the most recent message passed to Receive. It returns nil
+// if Receive has not been called. If Receive returned false, the returned
+// message may not have been unmarshaled.
func (s *ServerStreamForClient) Msg() any {
return s.msg
}
@@ -160,7 +164,8 @@ func (s *ServerStreamForClient) ResponseTrailer()
http.Header {
return s.conn.ResponseTrailer()
}
-// Close the receive side of the stream.
+// Close closes the receive side of the stream. If the stream could not be
+// constructed, the construction error is returned instead.
func (s *ServerStreamForClient) Close() error {
if s.constructErr != nil {
return s.constructErr
@@ -232,6 +237,8 @@ func (b *BidiStreamForClient) CloseRequest() error {
// Receive a message. When the server is done sending messages and no other
// errors have occurred, Receive will return an error that wraps [io.EOF].
+// Any error encountered while receiving, such as a server-side error, is
+// returned directly.
func (b *BidiStreamForClient) Receive(msg any) error {
if b.err != nil {
return b.err
diff --git a/protocol/triple/triple_protocol/client_stream_test.go
b/protocol/triple/triple_protocol/client_stream_test.go
index 28bba66d2..5c0353091 100644
--- a/protocol/triple/triple_protocol/client_stream_test.go
+++ b/protocol/triple/triple_protocol/client_stream_test.go
@@ -20,6 +20,7 @@ package triple_protocol
import (
"errors"
"fmt"
+ "io"
"net/http"
"testing"
)
@@ -107,3 +108,200 @@ type nopStreamingClientConn struct {
func (c *nopStreamingClientConn) Receive(msg any) error {
return nil
}
+
+// stubStreamingClientConn is a configurable StreamingClientConn used to
+// exercise close paths, error propagation and empty metadata without
+// involving a real transport.
+type stubStreamingClientConn struct {
+ StreamingClientConn
+
+ sendErr error
+ receiveErr error
+ closeReqErr error
+ closeRespErr error
+ requestHdr http.Header
+ responseHdr http.Header
+ responseTrlr http.Header
+
+ sendCalls int
+ receiveCalls int
+ closeReqCalls int
+ closeRespCalls int
+}
+
+func (c *stubStreamingClientConn) Spec() Spec { return Spec{} }
+func (c *stubStreamingClientConn) Peer() Peer { return Peer{} }
+
+func (c *stubStreamingClientConn) Send(any) error {
+ c.sendCalls++
+ return c.sendErr
+}
+
+func (c *stubStreamingClientConn) RequestHeader() http.Header {
+ return c.requestHdr
+}
+
+func (c *stubStreamingClientConn) CloseRequest() error {
+ c.closeReqCalls++
+ return c.closeReqErr
+}
+
+func (c *stubStreamingClientConn) Receive(any) error {
+ c.receiveCalls++
+ return c.receiveErr
+}
+
+func (c *stubStreamingClientConn) ResponseHeader() http.Header {
+ return c.responseHdr
+}
+
+func (c *stubStreamingClientConn) ResponseTrailer() http.Header {
+ return c.responseTrlr
+}
+
+func (c *stubStreamingClientConn) CloseResponse() error {
+ c.closeRespCalls++
+ return c.closeRespErr
+}
+
+// TestServerStreamForClientClose verifies that Close delegates to the
+// underlying connection's CloseResponse and returns its result.
+func TestServerStreamForClientClose(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{}
+ stream := &ServerStreamForClient{conn: conn}
+
+ assert.Nil(t, stream.Close())
+ assert.Equal(t, conn.closeRespCalls, 1)
+}
+
+// TestServerStreamForClientCloseRepeated verifies that calling Close more
+// than once keeps delegating to the underlying connection.
+func TestServerStreamForClientCloseRepeated(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{}
+ stream := &ServerStreamForClient{conn: conn}
+
+ assert.Nil(t, stream.Close())
+ assert.Nil(t, stream.Close())
+ assert.Equal(t, conn.closeRespCalls, 2)
+}
+
+// TestServerStreamForClientErrPropagation verifies that Err returns the
+// first non-EOF error encountered by Receive and returns nil for a normal
+// end of stream.
+func TestServerStreamForClientErrPropagation(t *testing.T) {
+ t.Parallel()
+ receiveErr := errors.New("receive failure")
+ conn := &stubStreamingClientConn{receiveErr: receiveErr}
+ stream := &ServerStreamForClient{conn: conn}
+
+ // A non-EOF error is returned by Err.
+ assert.False(t, stream.Receive(NewResponse(&pingv1.PingResponse{})))
+ assert.ErrorIs(t, stream.Err(), receiveErr)
+
+ // EOF is treated as a normal end of stream and Err returns nil.
+ connEOF := &stubStreamingClientConn{receiveErr: io.EOF}
+ streamEOF := &ServerStreamForClient{conn: connEOF}
+ assert.False(t, streamEOF.Receive(NewResponse(&pingv1.PingResponse{})))
+ assert.Nil(t, streamEOF.Err())
+}
+
+// TestServerStreamForClientReceiveStopsAfterError verifies that once Receive
+// has recorded an error, subsequent calls return false without advancing the
+// underlying connection.
+func TestServerStreamForClientReceiveStopsAfterError(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{receiveErr: errors.New("receive
failure")}
+ stream := &ServerStreamForClient{conn: conn}
+
+ assert.False(t, stream.Receive(NewResponse(&pingv1.PingResponse{})))
+ assert.False(t, stream.Receive(NewResponse(&pingv1.PingResponse{})))
+ assert.Equal(t, conn.receiveCalls, 1)
+}
+
+// TestServerStreamForClientEmptyMetadata verifies that ResponseHeader and
+// ResponseTrailer return empty headers.
+func TestServerStreamForClientEmptyMetadata(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{responseHdr: http.Header{},
responseTrlr: http.Header{}}
+ stream := &ServerStreamForClient{conn: conn}
+
+ assert.Equal(t, len(stream.ResponseHeader()), 0)
+ assert.Equal(t, len(stream.ResponseTrailer()), 0)
+}
+
+// TestBidiStreamForClientReceiveErrorPropagation verifies that Receive
+// returns the underlying connection error unchanged.
+func TestBidiStreamForClientReceiveErrorPropagation(t *testing.T) {
+ t.Parallel()
+ receiveErr := errors.New("receive failure")
+ conn := &stubStreamingClientConn{receiveErr: receiveErr}
+ stream := &BidiStreamForClient{conn: conn}
+
+ err := stream.Receive(NewResponse(&pingv1.PingResponse{}))
+ assert.ErrorIs(t, err, receiveErr)
+}
+
+// TestBidiStreamForClientRequestHeaderEmpty verifies that RequestHeader
+// returns an empty header.
+func TestBidiStreamForClientRequestHeaderEmpty(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{requestHdr: http.Header{}}
+ stream := &BidiStreamForClient{conn: conn}
+
+ assert.Equal(t, len(stream.RequestHeader()), 0)
+}
+
+// TestBidiStreamForClientCloseRequestRepeated verifies that calling
+// CloseRequest more than once keeps delegating to the underlying connection.
+func TestBidiStreamForClientCloseRequestRepeated(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{}
+ stream := &BidiStreamForClient{conn: conn}
+
+ assert.Nil(t, stream.CloseRequest())
+ assert.Nil(t, stream.CloseRequest())
+ assert.Equal(t, conn.closeReqCalls, 2)
+}
+
+// TestBidiStreamForClientCloseResponseRepeated verifies that calling
+// CloseResponse more than once keeps delegating to the underlying connection.
+func TestBidiStreamForClientCloseResponseRepeated(t *testing.T) {
+ t.Parallel()
+ conn := &stubStreamingClientConn{}
+ stream := &BidiStreamForClient{conn: conn}
+
+ assert.Nil(t, stream.CloseResponse())
+ assert.Nil(t, stream.CloseResponse())
+ assert.Equal(t, conn.closeRespCalls, 2)
+}
+
+// TestClientStreamForClientCloseAndReceiveClosesResponseOnCloseRequestError
+// verifies that CloseAndReceive closes the receive side before returning
+// the error when closing the send side fails.
+func
TestClientStreamForClientCloseAndReceiveClosesResponseOnCloseRequestError(t
*testing.T) {
+ t.Parallel()
+ closeReqErr := errors.New("close request failure")
+ conn := &stubStreamingClientConn{closeReqErr: closeReqErr}
+ stream := &ClientStreamForClient{conn: conn}
+
+ err := stream.CloseAndReceive(NewResponse(&pingv1.PingResponse{}))
+ assert.ErrorIs(t, err, closeReqErr)
+ assert.Equal(t, conn.closeRespCalls, 1)
+}
+
+// TestClientStreamForClientCloseAndReceiveClosesResponseOnReceiveError
+// verifies that CloseAndReceive closes the receive side before returning
+// the error when receiving the response fails.
+func TestClientStreamForClientCloseAndReceiveClosesResponseOnReceiveError(t
*testing.T) {
+ t.Parallel()
+ receiveErr := errors.New("receive failure")
+ conn := &stubStreamingClientConn{receiveErr: receiveErr}
+ stream := &ClientStreamForClient{conn: conn}
+
+ err := stream.CloseAndReceive(NewResponse(&pingv1.PingResponse{}))
+ assert.ErrorIs(t, err, receiveErr)
+ assert.Equal(t, conn.closeReqCalls, 1)
+ assert.Equal(t, conn.closeRespCalls, 1)
+}
diff --git a/protocol/triple/triple_protocol/handler_stream.go
b/protocol/triple/triple_protocol/handler_stream.go
index 8b0508331..4250f3086 100644
--- a/protocol/triple/triple_protocol/handler_stream.go
+++ b/protocol/triple/triple_protocol/handler_stream.go
@@ -52,7 +52,8 @@ func (c *ClientStream) RequestHeader() http.Header {
// available through the Msg method. It returns false when the stream stops,
// either by reaching the end or by encountering an unexpected error. After
// Receive returns false, the Err method will return any unexpected error
-// encountered.
+// encountered. Once Receive returns false, subsequent calls return false
+// without advancing the stream.
func (c *ClientStream) Receive(msg any) bool {
if c.err != nil {
return false
@@ -62,7 +63,9 @@ func (c *ClientStream) Receive(msg any) bool {
return c.err == nil
}
-// Msg returns the most recent message unmarshaled by a call to Receive.
+// Msg returns the most recent message passed to Receive. It returns nil
+// if Receive has not been called. If Receive returned false, the returned
+// message may not have been unmarshaled.
func (c *ClientStream) Msg() any {
// todo:// process nil pointer
//if c.msg == nil {
@@ -112,7 +115,8 @@ func (s *ServerStream) ResponseTrailer() http.Header {
}
// Send a message to the client. The first call to Send also sends the response
-// headers.
+// headers. A nil msg commits the response headers without writing a message
+// frame.
func (s *ServerStream) Send(msg any) error {
if msg == nil {
return s.conn.Send(nil)
@@ -149,7 +153,10 @@ func (b *BidiStream) RequestHeader() http.Header {
return b.conn.RequestHeader()
}
-// ExportableHeader returns the headers could be exported to users.
+// ExportableHeader returns the request headers that can be exported to
+// users. The underlying protocol implementation decides which headers are
+// exported: under the gRPC protocol, reserved headers are filtered out
+// (except the whitelisted ones) and header keys are lowercased.
func (b *BidiStream) ExportableHeader() http.Header {
return b.conn.ExportableHeader()
}
@@ -182,7 +189,8 @@ func (b *BidiStream) ResponseTrailer() http.Header {
}
// Send a message to the client. The first call to Send also sends the response
-// headers.
+// headers. A nil msg commits the response headers without writing a message
+// frame.
func (b *BidiStream) Send(msg any) error {
if msg == nil {
return b.conn.Send(nil)
diff --git a/protocol/triple/triple_protocol/handler_stream_compat.go
b/protocol/triple/triple_protocol/handler_stream_compat.go
index 70130f18b..133f1c4d1 100644
--- a/protocol/triple/triple_protocol/handler_stream_compat.go
+++ b/protocol/triple/triple_protocol/handler_stream_compat.go
@@ -65,6 +65,16 @@ func (c *compatHandlerStream) RecvMsg(m any) error {
return c.conn.Receive(m)
}
+// NewCompatStreamHandler constructs a [Handler] that adapts a
+// protoc-gen-go-triple generated stream function to the Triple streaming
+// model. The generated function receives a [grpc.ServerStream] backed by the
+// underlying protocol connection. Before invocation, the request headers are
+// injected into the context as gRPC-compatible metadata together with the
+// [constant.TripleGoInterfaceName] value. When an [Interceptor] is provided,
+// it wraps the whole implementation, so it runs before the metadata
+// injection. Note that SetHeader, SendHeader and SetTrailer on the adapted
+// [grpc.ServerStream] are no-ops: metadata set through them is silently
+// dropped.
func NewCompatStreamHandler(
procedure string,
srv any,
diff --git a/protocol/triple/triple_protocol/handler_stream_test.go
b/protocol/triple/triple_protocol/handler_stream_test.go
new file mode 100644
index 000000000..f1a3b649d
--- /dev/null
+++ b/protocol/triple/triple_protocol/handler_stream_test.go
@@ -0,0 +1,158 @@
+/*
+ * 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.
+ */
+
+package triple_protocol
+
+import (
+ "errors"
+ "io"
+ "net/http"
+ "testing"
+)
+
+import (
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/assert"
+ pingv1
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1"
+)
+
+// stubHandlerConn is a configurable StreamingHandlerConn used to exercise
+// receive stops, error propagation and empty metadata without involving a
+// real transport.
+type stubHandlerConn struct {
+ receiveErr error
+ sendErr error
+ exportable http.Header
+
+ receiveCalls int
+ sent []any
+}
+
+func (c *stubHandlerConn) Spec() Spec { return Spec{} }
+func (c *stubHandlerConn) Peer() Peer { return Peer{} }
+func (c *stubHandlerConn) Receive(any) error { c.receiveCalls++;
return c.receiveErr }
+func (c *stubHandlerConn) RequestHeader() http.Header { return nil }
+func (c *stubHandlerConn) ExportableHeader() http.Header { return c.exportable
}
+func (c *stubHandlerConn) Send(msg any) error {
+ c.sent = append(c.sent, msg)
+ return c.sendErr
+}
+func (c *stubHandlerConn) ResponseHeader() http.Header { return http.Header{}
}
+func (c *stubHandlerConn) ResponseTrailer() http.Header { return http.Header{}
}
+
+// TestClientStreamReceiveStopsAfterError verifies that once Receive has
+// recorded an error, subsequent calls return false without advancing the
+// underlying connection.
+func TestClientStreamReceiveStopsAfterError(t *testing.T) {
+ t.Parallel()
+ conn := &stubHandlerConn{receiveErr: errors.New("receive failure")}
+ stream := &ClientStream{conn: conn}
+
+ assert.False(t, stream.Receive(&pingv1.PingRequest{}))
+ assert.False(t, stream.Receive(&pingv1.PingRequest{}))
+ assert.Equal(t, conn.receiveCalls, 1)
+}
+
+// TestClientStreamErrPropagation verifies that Err returns the first non-EOF
+// error encountered by Receive and returns nil for a normal end of stream.
+func TestClientStreamErrPropagation(t *testing.T) {
+ t.Parallel()
+ receiveErr := errors.New("receive failure")
+ conn := &stubHandlerConn{receiveErr: receiveErr}
+ stream := &ClientStream{conn: conn}
+
+ // A non-EOF error is returned by Err.
+ assert.False(t, stream.Receive(&pingv1.PingRequest{}))
+ assert.ErrorIs(t, stream.Err(), receiveErr)
+
+ // EOF is treated as a normal end of stream and Err returns nil.
+ connEOF := &stubHandlerConn{receiveErr: io.EOF}
+ streamEOF := &ClientStream{conn: connEOF}
+ assert.False(t, streamEOF.Receive(&pingv1.PingRequest{}))
+ assert.Nil(t, streamEOF.Err())
+}
+
+// TestClientStreamMsgNil verifies that Msg returns nil when Receive has not
+// been called.
+func TestClientStreamMsgNil(t *testing.T) {
+ t.Parallel()
+ stream := &ClientStream{conn: &stubHandlerConn{}}
+ assert.Nil(t, stream.Msg())
+}
+
+// TestClientStreamRequestHeaderEmpty verifies that RequestHeader returns an
+// empty header.
+func TestClientStreamRequestHeaderEmpty(t *testing.T) {
+ t.Parallel()
+ stream := &ClientStream{conn: &stubHandlerConn{}}
+ assert.Equal(t, len(stream.RequestHeader()), 0)
+}
+
+// TestServerStreamSendNil verifies that Send delegates a nil msg to the
+// underlying connection unchanged.
+func TestServerStreamSendNil(t *testing.T) {
+ t.Parallel()
+ conn := &stubHandlerConn{}
+ stream := &ServerStream{conn: conn}
+
+ assert.Nil(t, stream.Send(nil))
+ assert.Equal(t, len(conn.sent), 1)
+ assert.Nil(t, conn.sent[0])
+}
+
+// TestServerStreamResponseHeaderEmpty verifies that ResponseHeader and
+// ResponseTrailer return empty headers.
+func TestServerStreamResponseHeaderEmpty(t *testing.T) {
+ t.Parallel()
+ stream := &ServerStream{conn: &stubHandlerConn{}}
+ assert.Equal(t, len(stream.ResponseHeader()), 0)
+ assert.Equal(t, len(stream.ResponseTrailer()), 0)
+}
+
+// TestBidiStreamReceiveErrorPropagation verifies that Receive returns the
+// underlying connection error unchanged.
+func TestBidiStreamReceiveErrorPropagation(t *testing.T) {
+ t.Parallel()
+ receiveErr := errors.New("receive failure")
+ conn := &stubHandlerConn{receiveErr: receiveErr}
+ stream := &BidiStream{conn: conn}
+
+ err := stream.Receive(&pingv1.CumSumRequest{})
+ assert.ErrorIs(t, err, receiveErr)
+}
+
+// TestBidiStreamSendNil verifies that Send delegates a nil msg to the
+// underlying connection unchanged.
+func TestBidiStreamSendNil(t *testing.T) {
+ t.Parallel()
+ conn := &stubHandlerConn{}
+ stream := &BidiStream{conn: conn}
+
+ assert.Nil(t, stream.Send(nil))
+ assert.Equal(t, len(conn.sent), 1)
+ assert.Nil(t, conn.sent[0])
+}
+
+// TestBidiStreamExportableHeader verifies that ExportableHeader forwards the
+// underlying connection's metadata unchanged.
+func TestBidiStreamExportableHeader(t *testing.T) {
+ t.Parallel()
+ hdr := http.Header{"X-Custom": []string{"value"}}
+ conn := &stubHandlerConn{exportable: hdr}
+ stream := &BidiStream{conn: conn}
+
+ assert.Equal(t, stream.ExportableHeader().Get("X-Custom"), "value")
+}
