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 221046200 fix(metadata, protocol, proxy): preserve request context
across RPC paths (#3579)
221046200 is described below
commit 221046200f20b511132243c0d747d4935d9a46c7
Author: Nene7ko_ <[email protected]>
AuthorDate: Thu Aug 20 13:27:54 2026 +0800
fix(metadata, protocol, proxy): preserve request context across RPC paths
(#3579)
* fix(context): preserve request context across RPC paths
* fix: resolve context propagation lint findings
* fix(rest): stop invocation after parameter parse errors
* fix(context): address PR review cancellation feedback
* fix: address context lint review findings
* refactor(registry): simplify metadata lookup
* fix(jsonrpc): preserve pipelined response order
* refactor(jsonrpc): reuse content type header constant
* fix(jsonrpc): isolate pipelined request timeouts
* fix(jsonrpc): bound pipelined request window
* chore(jsonrpc): apply modernize formatting
---
metadata/client.go | 33 +-
metadata/client_test.go | 41 ++
protocol/dubbo/dubbo_invoker.go | 6 +-
protocol/invocation/rpcinvocation.go | 43 +-
protocol/invocation/rpcinvocation_test.go | 32 ++
protocol/jsonrpc/server.go | 228 +++++++----
protocol/jsonrpc/server_test.go | 455 +++++++++++++++++++++
protocol/rest/server/rest_server.go | 19 +-
protocol/rest/server/rest_server_test.go | 166 ++++++++
proxy/proxy.go | 1 +
.../servicediscovery/service_discovery_registry.go | 11 +-
.../service_instances_changed_listener_impl.go | 126 ++++--
...service_instances_changed_listener_impl_test.go | 64 +++
remoting/exchange_client.go | 55 ++-
remoting/exchange_client_test.go | 60 ++-
remoting/getty/getty_client.go | 19 +-
16 files changed, 1213 insertions(+), 146 deletions(-)
diff --git a/metadata/client.go b/metadata/client.go
index bfd23c803..d26b43d40 100644
--- a/metadata/client.go
+++ b/metadata/client.go
@@ -55,6 +55,18 @@ func GetMetadataFromMetadataReport(revision string, instance
registry.ServiceIns
}
func GetMetadataFromRpc(revision string, instance registry.ServiceInstance)
(*info.MetadataInfo, error) {
+ return GetMetadataFromRpcWithContext(context.Background(), revision,
instance)
+}
+
+// GetMetadataFromRpcWithContext fetches metadata through the metadata service
+// while preserving the caller's context for the underlying RPC invocation.
+func GetMetadataFromRpcWithContext(ctx context.Context, revision string,
instance registry.ServiceInstance) (*info.MetadataInfo, error) {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
url, err := buildStandardMetadataServiceURL(instance)
if err != nil {
return nil, err
@@ -74,13 +86,12 @@ func GetMetadataFromRpc(revision string, instance
registry.ServiceInstance) (*in
defer func() {
invoker.Destroy()
}()
- return remoteService.getMetadataInfo(context.Background(), revision)
+ return remoteService.getMetadataInfo(ctx, revision)
}
// remoteMetadataService is the internal interface for fetching MetadataInfo
via RPC.
-// The context parameter is accepted for future cancellation support but is
not yet propagated.
type remoteMetadataService interface {
- getMetadataInfo(_ context.Context, revision string)
(*info.MetadataInfo, error)
+ getMetadataInfo(ctx context.Context, revision string)
(*info.MetadataInfo, error)
}
type triMetadataServiceV2 struct {
@@ -88,13 +99,15 @@ type triMetadataServiceV2 struct {
}
// getMetadataInfo fetches metadata via RPC using the Triple protocol
(Protobuf).
-// TODO(context-propagation): ctx is not yet forwarded to the invoker;
cancellation is not respected.
-func (m *triMetadataServiceV2) getMetadataInfo(_ context.Context, revision
string) (*info.MetadataInfo, error) {
+func (m *triMetadataServiceV2) getMetadataInfo(ctx context.Context, revision
string) (*info.MetadataInfo, error) {
const methodName = "GetMetadataInfo"
req := &tripleapi.MetadataRequest{Revision: revision}
metadataInfo := &tripleapi.MetadataInfoV2{}
inv, _ := generateInvocation(m.invoker.GetURL(), methodName, req,
metadataInfo, constant.CallUnary)
- res := m.invoker.Invoke(context.Background(), inv)
+ if rpcInv, ok := inv.(*invocation.RPCInvocation); ok {
+ rpcInv.SetContext(ctx)
+ }
+ res := m.invoker.Invoke(ctx, inv)
if res.Error() != nil {
logger.Errorf("[Metadata][RPC] could not get the metadata info
from remote provider, err=%v", res.Error())
return nil, perrors.Wrapf(res.Error(), "remote metadata call
failed")
@@ -160,15 +173,17 @@ type remoteMetadataServiceV1 struct {
}
// getMetadataInfo fetches metadata via RPC using the dubbo:// protocol
(Hessian2 serialization).
-// TODO(context-propagation): ctx is not yet forwarded to the invoker;
cancellation is not respected.
-func (m *remoteMetadataServiceV1) getMetadataInfo(_ context.Context, revision
string) (*info.MetadataInfo, error) {
+func (m *remoteMetadataServiceV1) getMetadataInfo(ctx context.Context,
revision string) (*info.MetadataInfo, error) {
const methodName = "getMetadataInfo"
// Use interface{} as reply parameter to accept any type (MetadataInfo
or string)
// This avoids panic when Java returns String instead of MetadataInfo
var rawResult any
inv, _ := generateInvocation(m.invoker.GetURL(), methodName, revision,
&rawResult, constant.CallUnary)
+ if rpcInv, ok := inv.(*invocation.RPCInvocation); ok {
+ rpcInv.SetContext(ctx)
+ }
- res := m.invoker.Invoke(context.Background(), inv)
+ res := m.invoker.Invoke(ctx, inv)
if res.Error() != nil {
logger.Errorf("[Metadata][RPC] RPC call failed to %s, err=%v",
m.invoker.GetURL().Location, res.Error())
return nil, perrors.Wrapf(res.Error(), "RPC call failed to %s",
m.invoker.GetURL().Location)
diff --git a/metadata/client_test.go b/metadata/client_test.go
index a2c192e0f..685f96e47 100644
--- a/metadata/client_test.go
+++ b/metadata/client_test.go
@@ -63,6 +63,8 @@ var (
}
)
+type metadataContextKey struct{}
+
func TestConvertMetadataInfoV2PreservesTag(t *testing.T) {
got := convertMetadataInfoV2(&tripleapi.MetadataInfoV2{
App: "dubbo-app",
@@ -186,6 +188,39 @@ func TestGetMetadataFromRpc(t *testing.T) {
})
}
+func TestGetMetadataFromRpcWithContext(t *testing.T) {
+ mockInvoker := new(mockInvoker)
+ mockProtocol := new(mockProtocol)
+ extension.SetProtocol("dubbo", func() base.Protocol {
+ return mockProtocol
+ })
+
+ mockProtocol.On("Refer").Return(mockInvoker).Once()
+ mockInvoker.On("Invoke").Return(&result.RPCResult{
+ Attrs: map[string]any{},
+ Rest: metadataInfo,
+ }).Once()
+ mockInvoker.On("Destroy").Once()
+
+ ctx := context.WithValue(context.Background(), metadataContextKey{},
"request-value")
+ metadata, err := GetMetadataFromRpcWithContext(ctx, "111", ins)
+ require.NoError(t, err)
+ assert.Equal(t, metadataInfo, metadata)
+ assert.Same(t, ctx, mockInvoker.invokedContext)
+}
+
+func TestTriMetadataServiceWithContext(t *testing.T) {
+ mockInvoker := new(mockInvoker)
+ mockInvoker.url =
common.NewURLWithOptions(common.WithProtocol(constant.TriProtocol))
+ mockInvoker.On("Invoke").Return(&result.RPCResult{Attrs:
map[string]any{}}).Once()
+
+ ctx := context.WithValue(context.Background(), metadataContextKey{},
"request-value")
+ metadata, err := (&triMetadataServiceV2{invoker:
mockInvoker}).getMetadataInfo(ctx, "111")
+ require.NoError(t, err)
+ require.NotNil(t, metadata)
+ assert.Same(t, ctx, mockInvoker.invokedContext)
+}
+
func TestGetMetadataFromRpc_MissingURLParams(t *testing.T) {
t.Run("missing protocol", func(t *testing.T) {
insNoProto := ®istry.DefaultServiceInstance{
@@ -377,9 +412,14 @@ func (m *mockProtocol) Destroy() {
type mockInvoker struct {
mock.Mock
+ invokedContext context.Context
+ url *common.URL
}
func (m *mockInvoker) GetURL() *common.URL {
+ if m.url != nil {
+ return m.url
+ }
return
common.NewURLWithOptions(common.WithProtocol(constant.DefaultProtocol))
}
@@ -392,6 +432,7 @@ func (m *mockInvoker) Destroy() {
}
func (m *mockInvoker) Invoke(ctx context.Context, inv base.Invocation)
result.Result {
+ m.invokedContext = ctx
args := m.Called()
res := args.Get(0).(result.Result)
diff --git a/protocol/dubbo/dubbo_invoker.go b/protocol/dubbo/dubbo_invoker.go
index d3a26fcaa..9f444b320 100644
--- a/protocol/dubbo/dubbo_invoker.go
+++ b/protocol/dubbo/dubbo_invoker.go
@@ -137,17 +137,17 @@ func (di *DubboInvoker) Invoke(ctx context.Context, ivc
base.Invocation) result.
timeout := di.getTimeout(inv)
if async {
if callBack, ok := inv.CallBack().(func(response
common.CallbackResponse)); ok {
- err = client.AsyncRequest(&ivc, url, timeout, callBack,
rest)
+ err = client.AsyncRequestContext(ctx, &ivc, url,
timeout, callBack, rest)
res.SetError(err)
} else {
- err = client.Send(&ivc, url, timeout)
+ err = client.SendContext(ctx, &ivc, url, timeout)
res.SetError(err)
}
} else {
if inv.Reply() == nil {
res.SetError(base.ErrNoReply)
} else {
- err = client.Request(&ivc, url, timeout, rest)
+ err = client.RequestContext(ctx, &ivc, url, timeout,
rest)
res.SetError(err)
}
}
diff --git a/protocol/invocation/rpcinvocation.go
b/protocol/invocation/rpcinvocation.go
index 9f5874a6b..256701fe7 100644
--- a/protocol/invocation/rpcinvocation.go
+++ b/protocol/invocation/rpcinvocation.go
@@ -47,6 +47,7 @@ type RPCInvocation struct {
arguments []any
reply any
callBack any
+ ctx context.Context
attachments map[string]any
// Refer to dubbo 2.7.6. It is different from attachment. It is used
in internal process.
attributes map[string]any
@@ -180,6 +181,27 @@ func (r *RPCInvocation) SetCallBack(c any) {
r.callBack = c
}
+// Context returns the request context associated with this invocation.
+func (r *RPCInvocation) Context() context.Context {
+ r.lock.RLock()
+ ctx := r.ctx
+ r.lock.RUnlock()
+ if ctx == nil {
+ return context.Background()
+ }
+ return ctx
+}
+
+// SetContext associates a request context with this invocation.
+func (r *RPCInvocation) SetContext(ctx context.Context) {
+ if ctx == nil {
+ return
+ }
+ r.lock.Lock()
+ defer r.lock.Unlock()
+ r.ctx = ctx
+}
+
func (r *RPCInvocation) ServiceKey() string {
return
common.ServiceKey(strings.TrimPrefix(r.GetAttachmentWithDefaultValue(constant.PathKey,
r.GetAttachmentWithDefaultValue(constant.InterfaceKey, "")), "/"),
r.GetAttachmentWithDefaultValue(constant.GroupKey, ""),
r.GetAttachmentWithDefaultValue(constant.VersionKey, ""))
@@ -250,14 +272,16 @@ func (r *RPCInvocation) GetAttributeWithDefaultValue(key
string, defaultValue an
}
func (r *RPCInvocation) GetAttachmentAsContext() context.Context {
- ctx := context.Background()
- var header = http.Header{}
+ ctx := r.Context()
+ header :=
cloneOutgoingHeader(triple_protocol.ExtractFromOutgoingContext(ctx))
for k, v := range r.Attachments() {
if str, ok := v.(string); ok {
+ header.Del(k)
header.Set(k, str)
continue
}
if str, ok := v.([]string); ok {
+ header.Del(k)
for _, s := range str {
header.Add(k, s)
}
@@ -267,6 +291,14 @@ func (r *RPCInvocation) GetAttachmentAsContext()
context.Context {
return triple_protocol.NewOutgoingContext(ctx, header)
}
+func cloneOutgoingHeader(header http.Header) http.Header {
+ cloned := make(http.Header, len(header))
+ for key, values := range header {
+ cloned[key] = append([]string(nil), values...)
+ }
+ return cloned
+}
+
func (r *RPCInvocation) MergeAttachmentFromContext(ctx context.Context) {
header := triple_protocol.ExtractFromOutgoingContext(ctx)
if header == nil {
@@ -349,6 +381,13 @@ func WithCallBack(callBack any) option {
}
}
+// WithContext creates an option with the request context.
+func WithContext(ctx context.Context) option {
+ return func(invo *RPCInvocation) {
+ invo.SetContext(ctx)
+ }
+}
+
// WithAttachments creates option with @attachments.
func WithAttachments(attachments map[string]any) option {
return func(invo *RPCInvocation) {
diff --git a/protocol/invocation/rpcinvocation_test.go
b/protocol/invocation/rpcinvocation_test.go
index f03f23c2f..c55b24a57 100644
--- a/protocol/invocation/rpcinvocation_test.go
+++ b/protocol/invocation/rpcinvocation_test.go
@@ -408,6 +408,38 @@ func TestRPCInvocation_GetAttachmentAsContext(t
*testing.T) {
assert.NotContains(t, header, "key3")
}
+func TestRPCInvocation_GetAttachmentAsContextPreservesRequestContext(t
*testing.T) {
+ type requestContextKey struct{}
+ requestCtx := context.WithValue(context.Background(),
requestContextKey{}, "request-value")
+ invocation := NewRPCInvocationWithOptions(
+ WithContext(requestCtx),
+ WithAttachment("key", "value"),
+ )
+
+ ctx := invocation.GetAttachmentAsContext()
+ assert.Equal(t, "request-value", ctx.Value(requestContextKey{}))
+ assert.Equal(t, "value",
triple_protocol.ExtractFromOutgoingContext(ctx).Get("key"))
+}
+
+func TestRPCInvocation_GetAttachmentAsContextPreservesOutgoingAttachments(t
*testing.T) {
+ requestCtx := context.Background()
+ requestCtx = triple_protocol.NewOutgoingContext(requestCtx, http.Header{
+ "Existing": {"existing-value"},
+ "Shared": {"old-value"},
+ })
+ invocation := NewRPCInvocationWithOptions(
+ WithContext(requestCtx),
+ WithAttachment("New", "new-value"),
+ WithAttachment("Shared", "new-shared-value"),
+ )
+
+ ctx := invocation.GetAttachmentAsContext()
+ header := triple_protocol.ExtractFromOutgoingContext(ctx)
+ assert.Equal(t, []string{"existing-value"}, header.Values("Existing"))
+ assert.Equal(t, []string{"new-value"}, header.Values("New"))
+ assert.Equal(t, []string{"new-shared-value"}, header.Values("Shared"))
+}
+
func TestRPCInvocation_MergeAttachmentFromContext(t *testing.T) {
invocation := NewRPCInvocationWithOptions()
diff --git a/protocol/jsonrpc/server.go b/protocol/jsonrpc/server.go
index 90dcae027..6685a3aae 100644
--- a/protocol/jsonrpc/server.go
+++ b/protocol/jsonrpc/server.go
@@ -60,6 +60,10 @@ const (
PathPrefix = byte('/')
// Max HTTP header size in Mib
MaxHeaderSize = 8 * 1024 * 1024
+ // ContentTypeHeader is the HTTP Content-Type header name.
+ ContentTypeHeader = "Content-Type"
+ // maxRequestWindowPerConnection bounds requests whose responses have
not been written yet.
+ maxRequestWindowPerConnection = 64
)
// Server is JSON RPC server wrapper
@@ -68,8 +72,7 @@ type Server struct {
once sync.Once
sync.RWMutex
- wg sync.WaitGroup
- timeout time.Duration
+ wg sync.WaitGroup
}
// NewServer creates new JSON RPC server.
@@ -80,6 +83,15 @@ func NewServer() *Server {
}
func (s *Server) handlePkg(conn net.Conn) {
+ connectionCtx, connectionCancel :=
context.WithCancel(context.Background())
+ responses := make(chan orderedResponse)
+ requestWindow := make(chan struct{}, maxRequestWindowPerConnection)
+ responseWriterDone := make(chan struct{})
+ go func() {
+ defer close(responseWriterDone)
+ writeResponsesInOrder(connectionCtx, connectionCancel, conn,
responses, requestWindow)
+ }()
+ var requestWG sync.WaitGroup
defer func() {
if r := recover(); r != nil {
logger.Warnf("[Jsonrpc][Server] connection panic,
local=%v, remote=%v, err=%v, debug stack=%s",
@@ -87,49 +99,30 @@ func (s *Server) handlePkg(conn net.Conn) {
}
conn.Close()
+ requestWG.Wait()
+ <-responseWriterDone
}()
+ // Register this after the cleanup defer so LIFO ordering cancels
request contexts before Wait.
+ defer connectionCancel()
- setTimeout := func(conn net.Conn, timeout time.Duration) {
- t := time.Time{}
- if timeout > time.Duration(0) {
- t = time.Now().Add(timeout)
- }
-
- if err := conn.SetDeadline(t); err != nil {
- logger.Errorf("[Jsonrpc][Server] connection.SetDeadline
failed, t=%v, err=%v", t, err)
- }
- }
-
- sendErrorResp := func(header http.Header, body []byte) error {
- rsp := &http.Response{
- Header: header,
- StatusCode: 500,
- ProtoMajor: 1,
- ProtoMinor: 1,
- ContentLength: int64(len(body)),
- Body: io.NopCloser(bytes.NewReader(body)),
- }
- rsp.Header.Del("Content-Type")
- rsp.Header.Del("Content-Length")
- rsp.Header.Del("Timeout")
-
- rspBuf := bytes.NewBuffer(make([]byte,
DefaultHTTPRspBufferSize))
- rspBuf.Reset()
- err := rsp.Write(rspBuf)
- if err != nil {
- return perrors.WithStack(err)
+ limitedReader := &io.LimitedReader{R: conn}
+ bufReader := bufio.NewReader(limitedReader)
+ var sequence uint64
+ for {
+ select {
+ case requestWindow <- struct{}{}:
+ case <-connectionCtx.Done():
+ return
}
- _, err = rspBuf.WriteTo(conn)
- return perrors.WithStack(err)
- }
- for {
- bufReader := bufio.NewReader(io.LimitReader(conn,
MaxHeaderSize))
+ limitedReader.N = int64(MaxHeaderSize - bufReader.Buffered())
if _, err := bufReader.Peek(1); errors.Is(err, io.EOF) {
+ <-requestWindow
return
}
r, err := http.ReadRequest(bufReader)
if err != nil {
+ <-requestWindow
logger.Warnf("[Jsonrpc][Server] read request failed,
err=%v", err)
return
}
@@ -137,6 +130,7 @@ func (s *Server) handlePkg(conn net.Conn) {
reqBody, err := io.ReadAll(r.Body)
r.Body.Close()
if err != nil {
+ <-requestWindow
return
}
@@ -150,49 +144,139 @@ func (s *Server) handlePkg(conn net.Conn) {
}
reqHeader["HttpMethod"] = r.Method
- httpTimeout := s.timeout
- contentType := reqHeader["Content-Type"]
+ contentType := reqHeader[ContentTypeHeader]
mediaType, _, parseErr := mime.ParseMediaType(contentType)
- if parseErr != nil || (mediaType != "application/json" &&
mediaType != "application/json-rpc") {
- setTimeout(conn, httpTimeout)
- errMsg := "unsupported content type: " + contentType
- if errRsp := sendErrorResp(r.Header, []byte(errMsg));
errRsp != nil {
- logger.Warnf("[Jsonrpc][Server] sendErrorResp
failed, header=%v, err_msg=%v, send_err=%v",
- r.Header, errMsg, errRsp)
- }
- return
- }
-
- ctx := context.Background()
+ unsupportedContentType := parseErr != nil || (mediaType !=
"application/json" && mediaType != "application/json-rpc")
- spanCtx, err :=
opentracing.GlobalTracer().Extract(opentracing.HTTPHeaders,
- opentracing.HTTPHeadersCarrier(r.Header))
- if err == nil {
- ctx = context.WithValue(ctx,
constant.TracingRemoteSpanCtx, spanCtx)
- }
+ requestCtx, requestCancel := context.WithCancel(connectionCtx)
+ r = r.WithContext(requestCtx)
+ ctx := contextFromRequest(r)
+ var timeoutCancel context.CancelFunc
if len(reqHeader["Timeout"]) > 0 {
timeout, err := time.ParseDuration(reqHeader["Timeout"])
if err == nil {
- httpTimeout = timeout
- var cancel context.CancelFunc
- ctx, cancel = context.WithTimeout(ctx,
httpTimeout)
- defer cancel()
+ ctx, timeoutCancel = context.WithTimeout(ctx,
timeout)
}
delete(reqHeader, "Timeout")
}
- setTimeout(conn, httpTimeout)
- if err := serveRequest(ctx, reqHeader, reqBody, conn); err !=
nil {
- if errRsp := sendErrorResp(r.Header,
[]byte(perrors.WithStack(err).Error())); errRsp != nil {
- logger.Warnf("[Jsonrpc][Server] sendErrorResp
failed, header=%v, err=%v, send_err=%v",
- r.Header, perrors.WithStack(err),
errRsp)
+ requestSequence := sequence
+ sequence++
+ requestWG.Add(1)
+ go func(ctx context.Context, requestCancel, timeoutCancel
context.CancelFunc, header map[string]string, body []byte,
+ responseHeader http.Header, contentType string,
unsupportedContentType bool, requestSequence uint64) {
+ defer requestWG.Done()
+ defer requestCancel()
+ if timeoutCancel != nil {
+ defer timeoutCancel()
+ }
+
+ response := buildOrderedResponse(ctx, header, body,
responseHeader, contentType,
+ unsupportedContentType, requestSequence)
+ select {
+ case responses <- response:
+ case <-connectionCtx.Done():
}
+ }(ctx, requestCancel, timeoutCancel, reqHeader, reqBody,
r.Header, contentType, unsupportedContentType, requestSequence)
+ }
+}
+
+type orderedResponse struct {
+ sequence uint64
+ data []byte
+ closeConnection bool
+}
+
+func buildOrderedResponse(ctx context.Context, header map[string]string, body
[]byte, responseHeader http.Header,
+ contentType string, unsupportedContentType bool, sequence uint64)
orderedResponse {
+ responseBuffer := bytes.NewBuffer(nil)
+ if unsupportedContentType {
+ errMsg := "unsupported content type: " + contentType
+ if err := writeHTTPErrorResponse(responseBuffer,
responseHeader, []byte(errMsg)); err != nil {
+ logger.Warnf("[Jsonrpc][Server] write error response
failed, header=%v, err_msg=%v, write_err=%v",
+ responseHeader, errMsg, err)
+ }
+ return orderedResponse{sequence: sequence, data:
responseBuffer.Bytes(), closeConnection: true}
+ }
+
+ err := serveRequest(ctx, header, body, responseBuffer)
+ if err == nil {
+ return orderedResponse{sequence: sequence, data:
responseBuffer.Bytes()}
+ }
+ if writeErr := writeHTTPErrorResponse(responseBuffer, responseHeader,
[]byte(perrors.WithStack(err).Error())); writeErr != nil {
+ logger.Warnf("[Jsonrpc][Server] write error response failed,
header=%v, err=%v, write_err=%v",
+ responseHeader, perrors.WithStack(err), writeErr)
+ }
+ logger.Infof("[Jsonrpc][Server] unexpected error serving request,
closing socket, err=%v", err)
+ return orderedResponse{sequence: sequence, data:
responseBuffer.Bytes(), closeConnection: true}
+}
- logger.Infof("[Jsonrpc][Server] unexpected error
serving request, closing socket, err=%v", err)
+func writeHTTPErrorResponse(writer io.Writer, header http.Header, body []byte)
error {
+ rsp := &http.Response{
+ Header: header.Clone(),
+ StatusCode: 500,
+ ProtoMajor: 1,
+ ProtoMinor: 1,
+ ContentLength: int64(len(body)),
+ Body: io.NopCloser(bytes.NewReader(body)),
+ }
+ rsp.Header.Del(ContentTypeHeader)
+ rsp.Header.Del("Content-Length")
+ rsp.Header.Del("Timeout")
+
+ rspBuf := bytes.NewBuffer(make([]byte, DefaultHTTPRspBufferSize))
+ rspBuf.Reset()
+ if err := rsp.Write(rspBuf); err != nil {
+ return perrors.WithStack(err)
+ }
+ _, err := rspBuf.WriteTo(writer)
+ return perrors.WithStack(err)
+}
+
+func writeResponsesInOrder(ctx context.Context, cancel context.CancelFunc,
conn net.Conn, responses <-chan orderedResponse,
+ requestWindow <-chan struct{}) {
+ nextSequence := uint64(0)
+ pending := make(map[uint64]orderedResponse)
+ for {
+ select {
+ case <-ctx.Done():
return
+ case response := <-responses:
+ pending[response.sequence] = response
}
+
+ for {
+ response, ok := pending[nextSequence]
+ if !ok {
+ break
+ }
+ delete(pending, nextSequence)
+ if _, err :=
bytes.NewReader(response.data).WriteTo(conn); err != nil {
+ logger.Warnf("[Jsonrpc][Server] write response
failed, sequence=%d, err=%v", nextSequence, err)
+ cancel()
+ conn.Close()
+ return
+ }
+ if response.closeConnection {
+ cancel()
+ conn.Close()
+ return
+ }
+ nextSequence++
+ <-requestWindow
+ }
+ }
+}
+
+func contextFromRequest(r *http.Request) context.Context {
+ ctx := r.Context()
+ spanCtx, err :=
opentracing.GlobalTracer().Extract(opentracing.HTTPHeaders,
+ opentracing.HTTPHeadersCarrier(r.Header))
+ if err == nil {
+ ctx = context.WithValue(ctx, constant.TracingRemoteSpanCtx,
spanCtx)
}
+ return ctx
}
func accept(listener net.Listener, fn func(net.Conn)) error {
@@ -270,7 +354,7 @@ func (s *Server) Stop() {
})
}
-func serveRequest(ctx context.Context, header map[string]string, body []byte,
conn net.Conn) error {
+func serveRequest(ctx context.Context, header map[string]string, body []byte,
writer io.Writer) error {
sendErrorResp := func(header map[string]string, body []byte) error {
rsp := &http.Response{
Header: make(http.Header),
@@ -280,7 +364,7 @@ func serveRequest(ctx context.Context, header
map[string]string, body []byte, co
ContentLength: int64(len(body)),
Body: io.NopCloser(bytes.NewReader(body)),
}
- rsp.Header.Del("Content-Type")
+ rsp.Header.Del(ContentTypeHeader)
rsp.Header.Del("Content-Length")
rsp.Header.Del("Timeout")
for k, v := range header {
@@ -293,7 +377,7 @@ func serveRequest(ctx context.Context, header
map[string]string, body []byte, co
if err != nil {
return perrors.WithStack(err)
}
- _, err = rspBuf.WriteTo(conn)
+ _, err = rspBuf.WriteTo(writer)
return perrors.WithStack(err)
}
@@ -306,7 +390,7 @@ func serveRequest(ctx context.Context, header
map[string]string, body []byte, co
ContentLength: int64(len(body)),
Body: io.NopCloser(bytes.NewReader(body)),
}
- rsp.Header.Del("Content-Type")
+ rsp.Header.Del(ContentTypeHeader)
rsp.Header.Del("Content-Length")
rsp.Header.Del("Timeout")
for k, v := range header {
@@ -319,7 +403,7 @@ func serveRequest(ctx context.Context, header
map[string]string, body []byte, co
if err != nil {
return perrors.WithStack(err)
}
- _, err = rspBuf.WriteTo(conn)
+ _, err = rspBuf.WriteTo(writer)
return perrors.WithStack(err)
}
@@ -353,10 +437,12 @@ func serveRequest(ctx context.Context, header
map[string]string, body []byte, co
}
invoker := exporter.(*JsonrpcExporter).GetInvoker()
if invoker != nil {
- result := invoker.Invoke(ctx,
invocation.NewRPCInvocation(methodName, args, map[string]any{
+ rpcInvocation := invocation.NewRPCInvocation(methodName, args,
map[string]any{
constant.PathKey: path,
constant.VersionKey: codec.req.Version,
- }))
+ })
+ rpcInvocation.SetContext(ctx)
+ result := invoker.Invoke(ctx, rpcInvocation)
if err := result.Error(); err != nil {
rspStream, codecErr := codec.Write(err.Error(),
invalidRequest)
if codecErr != nil {
diff --git a/protocol/jsonrpc/server_test.go b/protocol/jsonrpc/server_test.go
index cb59ae5ef..9a3bda91c 100644
--- a/protocol/jsonrpc/server_test.go
+++ b/protocol/jsonrpc/server_test.go
@@ -20,9 +20,14 @@ package jsonrpc
import (
"bufio"
"context"
+ "encoding/json"
+ "fmt"
"io"
"net"
"net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
"testing"
"time"
)
@@ -31,6 +36,12 @@ import (
"github.com/stretchr/testify/require"
)
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ "dubbo.apache.org/dubbo-go/v3/protocol/result"
+)
+
// sendHTTPRequest writes an HTTP request to conn and returns the parsed
response.
// A read deadline is set to avoid hanging when the server may not respond
// (e.g. valid content type but no registered service).
@@ -71,6 +82,18 @@ func TestServeRequest_ServiceNotFound(t *testing.T) {
require.Contains(t, err.Error(), "service not found")
}
+func TestContextFromRequestPreservesRequestContext(t *testing.T) {
+ type contextKey struct{}
+ requestCtx, cancel :=
context.WithCancel(context.WithValue(context.Background(), contextKey{},
"request-value"))
+ defer cancel()
+ request := httptest.NewRequestWithContext(requestCtx, http.MethodPost,
"/test", nil)
+
+ ctx := contextFromRequest(request)
+ require.Equal(t, "request-value", ctx.Value(contextKey{}))
+ cancel()
+ require.Error(t, ctx.Err())
+}
+
func TestHandlePkg_ContentType(t *testing.T) {
tests := []struct {
name string
@@ -144,3 +167,435 @@ func TestHandlePkg_ContentType(t *testing.T) {
})
}
}
+
+type blockingInvoker struct {
+ base.BaseInvoker
+ started chan struct{}
+ canceled chan struct{}
+}
+
+func (i *blockingInvoker) Invoke(ctx context.Context, _ base.Invocation)
result.Result {
+ close(i.started)
+ <-ctx.Done()
+ close(i.canceled)
+ return &result.RPCResult{Err: ctx.Err()}
+}
+
+func TestHandlePkgCancelsInvocationWhenClientDisconnects(t *testing.T) {
+ const servicePath = "context-cancel-test"
+ protocol := GetProtocol().(*JsonrpcProtocol)
+ invoker := &blockingInvoker{
+ BaseInvoker: *base.NewBaseInvoker(common.NewURLWithOptions(
+ common.WithProtocol(JSONRPC),
+ common.WithPath("/"+servicePath),
+ )),
+ started: make(chan struct{}),
+ canceled: make(chan struct{}),
+ }
+ protocol.SetExporterMap(servicePath, NewJsonrpcExporter(servicePath,
invoker, protocol.ExporterMap()))
+ t.Cleanup(func() { protocol.ExporterMap().Delete(servicePath) })
+
+ server := NewServer()
+ serverConn, clientConn := net.Pipe()
+ handleDone := make(chan struct{})
+ go func() {
+ server.handlePkg(serverConn)
+ close(handleDone)
+ }()
+
+ body := `{"jsonrpc":"2.0","method":"Blocked","params":[],"id":1}`
+ request := "POST /" + servicePath + " HTTP/1.1\r\n" +
+ "Host: localhost\r\n" +
+ "Content-Type: application/json\r\n" +
+ "Content-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n" + body
+ _, err := clientConn.Write([]byte(request))
+ require.NoError(t, err)
+
+ select {
+ case <-invoker.started:
+ case <-time.After(time.Second):
+ t.Fatal("invoker was not started")
+ }
+ require.NoError(t, clientConn.Close())
+
+ select {
+ case <-invoker.canceled:
+ case <-time.After(time.Second):
+ t.Fatal("invocation context was not canceled after client
disconnect")
+ }
+ select {
+ case <-handleDone:
+ case <-time.After(time.Second):
+ t.Fatal("connection handler did not exit")
+ }
+ require.NoError(t, serverConn.Close())
+}
+
+type orderedResponseInvoker struct {
+ base.BaseInvoker
+ firstStarted chan struct{}
+ secondReady chan struct{}
+ releaseFirst chan struct{}
+}
+
+type signalingResult struct {
+ ready chan struct{}
+}
+
+func (r signalingResult) MarshalJSON() ([]byte, error) {
+ close(r.ready)
+ return json.Marshal("fast")
+}
+
+func (i *orderedResponseInvoker) Invoke(_ context.Context, invocation
base.Invocation) result.Result {
+ switch invocation.MethodName() {
+ case "Slow":
+ close(i.firstStarted)
+ <-i.releaseFirst
+ return &result.RPCResult{Rest: "slow"}
+ case "Fast":
+ return &result.RPCResult{Rest: signalingResult{ready:
i.secondReady}}
+ default:
+ return &result.RPCResult{Err: fmt.Errorf("unexpected method
%s", invocation.MethodName())}
+ }
+}
+
+func TestHandlePkgPreservesPipelinedResponseOrder(t *testing.T) {
+ const servicePath = "response-order-test"
+ protocol := GetProtocol().(*JsonrpcProtocol)
+ invoker := &orderedResponseInvoker{
+ BaseInvoker:
*base.NewBaseInvoker(common.NewURLWithOptions(common.WithProtocol(JSONRPC))),
+ firstStarted: make(chan struct{}),
+ secondReady: make(chan struct{}),
+ releaseFirst: make(chan struct{}),
+ }
+ protocol.SetExporterMap(servicePath, NewJsonrpcExporter(servicePath,
invoker, protocol.ExporterMap()))
+ t.Cleanup(func() { protocol.ExporterMap().Delete(servicePath) })
+
+ serverConn, clientConn := net.Pipe()
+ handleDone := make(chan struct{})
+ go func() {
+ NewServer().handlePkg(serverConn)
+ close(handleDone)
+ }()
+
+ var releaseOnce sync.Once
+ releaseFirst := func() { releaseOnce.Do(func() {
close(invoker.releaseFirst) }) }
+ t.Cleanup(func() {
+ releaseFirst()
+ _ = clientConn.Close()
+ select {
+ case <-handleDone:
+ case <-time.After(time.Second):
+ t.Error("connection handler did not exit")
+ }
+ })
+ require.NoError(t,
clientConn.SetReadDeadline(time.Now().Add(3*time.Second)))
+
+ type response struct {
+ id int
+ result string
+ err error
+ }
+ responses := make(chan response, 2)
+ go func() {
+ reader := bufio.NewReader(clientConn)
+ for range 2 {
+ httpResponse, err := http.ReadResponse(reader, nil)
+ if err != nil {
+ responses <- response{err: err}
+ return
+ }
+ var payload struct {
+ ID int `json:"id"`
+ Result string `json:"result"`
+ }
+ err =
json.NewDecoder(httpResponse.Body).Decode(&payload)
+ httpResponse.Body.Close()
+ responses <- response{id: payload.ID, result:
payload.Result, err: err}
+ }
+ }()
+
+ makeRequest := func(method string, id int) string {
+ body :=
fmt.Sprintf(`{"jsonrpc":"2.0","method":%q,"params":[],"id":%d}`, method, id)
+ return "POST /" + servicePath + " HTTP/1.1\r\n" +
+ "Host: localhost\r\n" +
+ "Content-Type: application/json\r\n" +
+ "Content-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n"
+ body
+ }
+
+ writeDone := make(chan error, 1)
+ go func() {
+ _, err := clientConn.Write([]byte(makeRequest("Slow", 1) +
makeRequest("Fast", 2)))
+ writeDone <- err
+ }()
+ select {
+ case <-invoker.firstStarted:
+ case <-time.After(time.Second):
+ t.Fatal("first invocation was not started")
+ }
+ select {
+ case <-invoker.secondReady:
+ case <-time.After(time.Second):
+ t.Fatal("second response was not encoded")
+ }
+ require.NoError(t, <-writeDone)
+
+ select {
+ case got := <-responses:
+ t.Fatalf("received response %d before the first request
completed", got.id)
+ case <-time.After(100 * time.Millisecond):
+ }
+
+ releaseFirst()
+ readResponse := func() response {
+ select {
+ case got := <-responses:
+ return got
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for response")
+ return response{}
+ }
+ }
+ first := readResponse()
+ require.NoError(t, first.err)
+ require.Equal(t, 1, first.id)
+ require.Equal(t, "slow", first.result)
+ second := readResponse()
+ require.NoError(t, second.err)
+ require.Equal(t, 2, second.id)
+ require.Equal(t, "fast", second.result)
+}
+
+type boundedWindowInvoker struct {
+ base.BaseInvoker
+ started chan struct{}
+ encoded chan struct{}
+ releaseFirst chan struct{}
+}
+
+type boundedWindowResult struct {
+ encoded chan struct{}
+}
+
+func (r boundedWindowResult) MarshalJSON() ([]byte, error) {
+ r.encoded <- struct{}{}
+ return json.Marshal("done")
+}
+
+func (i *boundedWindowInvoker) Invoke(_ context.Context, invocation
base.Invocation) result.Result {
+ i.started <- struct{}{}
+ if invocation.MethodName() == "Blocked" {
+ <-i.releaseFirst
+ }
+ return &result.RPCResult{Rest: boundedWindowResult{encoded: i.encoded}}
+}
+
+func TestHandlePkgBoundsPipelinedRequestWindow(t *testing.T) {
+ const (
+ servicePath = "bounded-request-window-test"
+ requestCount = maxRequestWindowPerConnection + 1
+ )
+ protocol := GetProtocol().(*JsonrpcProtocol)
+ invoker := &boundedWindowInvoker{
+ BaseInvoker:
*base.NewBaseInvoker(common.NewURLWithOptions(common.WithProtocol(JSONRPC))),
+ started: make(chan struct{}, requestCount),
+ encoded: make(chan struct{}, requestCount),
+ releaseFirst: make(chan struct{}),
+ }
+ protocol.SetExporterMap(servicePath, NewJsonrpcExporter(servicePath,
invoker, protocol.ExporterMap()))
+ t.Cleanup(func() { protocol.ExporterMap().Delete(servicePath) })
+
+ serverConn, clientConn := net.Pipe()
+ handleDone := make(chan struct{})
+ go func() {
+ NewServer().handlePkg(serverConn)
+ close(handleDone)
+ }()
+
+ var releaseOnce sync.Once
+ releaseFirst := func() { releaseOnce.Do(func() {
close(invoker.releaseFirst) }) }
+ t.Cleanup(func() {
+ releaseFirst()
+ _ = clientConn.Close()
+ select {
+ case <-handleDone:
+ case <-time.After(time.Second):
+ t.Error("connection handler did not exit")
+ }
+ })
+ require.NoError(t,
clientConn.SetDeadline(time.Now().Add(5*time.Second)))
+
+ makeRequest := func(method string, id int) string {
+ body :=
fmt.Sprintf(`{"jsonrpc":"2.0","method":%q,"params":[],"id":%d}`, method, id)
+ return "POST /" + servicePath + " HTTP/1.1\r\n" +
+ "Host: localhost\r\n" +
+ "Content-Type: application/json\r\n" +
+ "Content-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n"
+ body
+ }
+
+ var requests strings.Builder
+ requests.WriteString(makeRequest("Blocked", 1))
+ for id := 2; id <= requestCount; id++ {
+ requests.WriteString(makeRequest("Fast", id))
+ }
+ writeDone := make(chan error, 1)
+ go func() {
+ _, err := clientConn.Write([]byte(requests.String()))
+ writeDone <- err
+ }()
+
+ waitForSignals := func(signals <-chan struct{}, count int, description
string) {
+ for range count {
+ select {
+ case <-signals:
+ case <-time.After(2 * time.Second):
+ t.Fatalf("timed out waiting for %s %d",
description, count)
+ }
+ }
+ }
+ waitForSignals(invoker.started, maxRequestWindowPerConnection, "started
invocations")
+ waitForSignals(invoker.encoded, maxRequestWindowPerConnection-1,
"encoded responses")
+ select {
+ case <-invoker.started:
+ t.Fatalf("more than %d invocations started before the response
window advanced", maxRequestWindowPerConnection)
+ case <-time.After(100 * time.Millisecond):
+ }
+
+ readDone := make(chan error, 1)
+ go func() {
+ reader := bufio.NewReader(clientConn)
+ for range requestCount {
+ response, err := http.ReadResponse(reader, nil)
+ if err != nil {
+ readDone <- err
+ return
+ }
+ _, err = io.Copy(io.Discard, response.Body)
+ closeErr := response.Body.Close()
+ if err != nil {
+ readDone <- err
+ return
+ }
+ if closeErr != nil {
+ readDone <- closeErr
+ return
+ }
+ }
+ readDone <- nil
+ }()
+
+ releaseFirst()
+ waitForSignals(invoker.started, 1, "invocations after advancing the
response window")
+ require.NoError(t, <-writeDone)
+ require.NoError(t, <-readDone)
+}
+
+type requestTimeoutInvoker struct {
+ base.BaseInvoker
+ longStarted chan struct{}
+ shortTimedOut chan struct{}
+ longCanceled chan struct{}
+ releaseLong chan struct{}
+}
+
+func (i *requestTimeoutInvoker) Invoke(ctx context.Context, invocation
base.Invocation) result.Result {
+ switch invocation.MethodName() {
+ case "Long":
+ close(i.longStarted)
+ select {
+ case <-ctx.Done():
+ close(i.longCanceled)
+ return &result.RPCResult{Err: ctx.Err()}
+ case <-i.releaseLong:
+ return &result.RPCResult{Rest: "long"}
+ }
+ case "Short":
+ <-ctx.Done()
+ close(i.shortTimedOut)
+ return &result.RPCResult{Err: ctx.Err()}
+ default:
+ return &result.RPCResult{Err: fmt.Errorf("unexpected method
%s", invocation.MethodName())}
+ }
+}
+
+func TestHandlePkgIsolatesPipelinedRequestTimeouts(t *testing.T) {
+ const servicePath = "request-timeout-test"
+ protocol := GetProtocol().(*JsonrpcProtocol)
+ invoker := &requestTimeoutInvoker{
+ BaseInvoker:
*base.NewBaseInvoker(common.NewURLWithOptions(common.WithProtocol(JSONRPC))),
+ longStarted: make(chan struct{}),
+ shortTimedOut: make(chan struct{}),
+ longCanceled: make(chan struct{}),
+ releaseLong: make(chan struct{}),
+ }
+ protocol.SetExporterMap(servicePath, NewJsonrpcExporter(servicePath,
invoker, protocol.ExporterMap()))
+ t.Cleanup(func() { protocol.ExporterMap().Delete(servicePath) })
+
+ serverConn, clientConn := net.Pipe()
+ handleDone := make(chan struct{})
+ go func() {
+ NewServer().handlePkg(serverConn)
+ close(handleDone)
+ }()
+
+ var releaseOnce sync.Once
+ releaseLong := func() { releaseOnce.Do(func() {
close(invoker.releaseLong) }) }
+ t.Cleanup(func() {
+ releaseLong()
+ _ = clientConn.Close()
+ select {
+ case <-handleDone:
+ case <-time.After(time.Second):
+ t.Error("connection handler did not exit")
+ }
+ })
+ require.NoError(t,
clientConn.SetReadDeadline(time.Now().Add(3*time.Second)))
+
+ makeRequest := func(method string, id int, timeout time.Duration)
string {
+ body :=
fmt.Sprintf(`{"jsonrpc":"2.0","method":%q,"params":[],"id":%d}`, method, id)
+ return "POST /" + servicePath + " HTTP/1.1\r\n" +
+ "Host: localhost\r\n" +
+ "Content-Type: application/json\r\n" +
+ "Timeout: " + timeout.String() + "\r\n" +
+ "Content-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n"
+ body
+ }
+
+ writeDone := make(chan error, 1)
+ go func() {
+ _, err := clientConn.Write([]byte(
+ makeRequest("Long", 1, 5*time.Second) +
makeRequest("Short", 2, 50*time.Millisecond),
+ ))
+ writeDone <- err
+ }()
+
+ select {
+ case <-invoker.longStarted:
+ case <-time.After(time.Second):
+ t.Fatal("long invocation was not started")
+ }
+ select {
+ case <-invoker.shortTimedOut:
+ case <-time.After(time.Second):
+ t.Fatal("short invocation did not reach its context deadline")
+ }
+ require.NoError(t, <-writeDone)
+ select {
+ case <-invoker.longCanceled:
+ t.Fatal("short request timeout canceled the long request")
+ case <-time.After(250 * time.Millisecond):
+ }
+
+ releaseLong()
+ reader := bufio.NewReader(clientConn)
+ for id := 1; id <= 2; id++ {
+ httpResponse, err := http.ReadResponse(reader, nil)
+ require.NoError(t, err)
+ var payload struct {
+ ID int `json:"id"`
+ }
+ require.NoError(t,
json.NewDecoder(httpResponse.Body).Decode(&payload))
+ require.NoError(t, httpResponse.Body.Close())
+ require.Equal(t, id, payload.ID)
+ }
+}
diff --git a/protocol/rest/server/rest_server.go
b/protocol/rest/server/rest_server.go
index 4d6a189cd..956a75df4 100644
--- a/protocol/rest/server/rest_server.go
+++ b/protocol/rest/server/rest_server.go
@@ -18,7 +18,6 @@
package server
import (
- "context"
"errors"
"net/http"
"reflect"
@@ -105,12 +104,22 @@ func GetRouteFunc(invoker base.Invoker, methodConfig
*rest_config.RestMethodConf
}
if err != nil {
logger.Errorf("[Rest][Server] parsing http parameters
error, err=%v", err)
- err = resp.WriteError(http.StatusInternalServerError,
errors.New(parseParameterErrorStr))
- if err != nil {
- logger.Errorf("[Rest][Server] write error
string failed, err=%v", err)
+ if writeErr :=
resp.WriteError(http.StatusInternalServerError,
errors.New(parseParameterErrorStr)); writeErr != nil {
+ logger.Errorf("[Rest][Server] write error
string failed, err=%v", writeErr)
}
+ return
+ }
+ rawRequest := req.RawRequest()
+ if rawRequest == nil {
+ logger.Errorf("[Rest][Server] request adapter returned
a nil raw request")
+ if writeErr :=
resp.WriteError(http.StatusInternalServerError, errors.New("raw HTTP request is
nil")); writeErr != nil {
+ logger.Errorf("[Rest][Server] write error
failed, err=%v", writeErr)
+ }
+ return
}
- result := invoker.Invoke(context.Background(),
invocation.NewRPCInvocation(methodConfig.MethodName, args,
make(map[string]any)))
+ rpcInvocation :=
invocation.NewRPCInvocation(methodConfig.MethodName, args, make(map[string]any))
+ rpcInvocation.SetContext(rawRequest.Context())
+ result := invoker.Invoke(rawRequest.Context(), rpcInvocation)
if result.Error() != nil {
err = resp.WriteError(http.StatusInternalServerError,
result.Error())
if err != nil {
diff --git a/protocol/rest/server/rest_server_test.go
b/protocol/rest/server/rest_server_test.go
new file mode 100644
index 000000000..493639ea2
--- /dev/null
+++ b/protocol/rest/server/rest_server_test.go
@@ -0,0 +1,166 @@
+/*
+ * 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 server
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ "dubbo.apache.org/dubbo-go/v3/protocol/invocation"
+ "dubbo.apache.org/dubbo-go/v3/protocol/rest/config"
+ "dubbo.apache.org/dubbo-go/v3/protocol/result"
+)
+
+type RestContextService struct{}
+
+func (s *RestContextService) Handle() (string, error) {
+ return "ok", nil
+}
+
+type contextCaptureInvoker struct {
+ *base.BaseInvoker
+ ctx context.Context
+ lastInvocation base.Invocation
+ invocations int
+}
+
+func (i *contextCaptureInvoker) Invoke(ctx context.Context, inv
base.Invocation) result.Result {
+ i.ctx = ctx
+ i.lastInvocation = inv
+ i.invocations++
+ return &result.RPCResult{Rest: "ok"}
+}
+
+type testRestRequest struct {
+ request *http.Request
+}
+
+func (r *testRestRequest) RawRequest() *http.Request { return r.request }
+
+func (r *testRestRequest) PathParameter(string) string { return "" }
+
+func (r *testRestRequest) PathParameters() map[string]string { return nil }
+
+func (r *testRestRequest) QueryParameter(string) string { return "" }
+
+func (r *testRestRequest) QueryParameters(string) []string { return nil }
+
+func (r *testRestRequest) BodyParameter(string) (string, error) { return "",
nil }
+
+func (r *testRestRequest) HeaderParameter(string) string { return "" }
+
+func (r *testRestRequest) ReadEntity(any) error { return nil }
+
+type testRestResponse struct {
+ *httptest.ResponseRecorder
+}
+
+func (r *testRestResponse) WriteError(status int, err error) error {
+ r.WriteHeader(status)
+ if err == nil {
+ return nil
+ }
+ _, writeErr := fmt.Fprint(r, err)
+ return writeErr
+}
+
+func (r *testRestResponse) WriteEntity(value any) error {
+ _, err := fmt.Fprint(r, value)
+ return err
+}
+
+func TestGetRouteFuncPropagatesRequestContext(t *testing.T) {
+ const interfaceName = "RestContextService"
+ const protocol = "rest"
+ const version = "context-test"
+
+ _, err := common.ServiceMap.Register(interfaceName, protocol, "",
version, &RestContextService{})
+ require.NoError(t, err)
+ t.Cleanup(func() {
+ require.NoError(t, common.ServiceMap.UnRegister(interfaceName,
protocol, common.ServiceKey(interfaceName, "", version)))
+ })
+
+ url := common.NewURLWithOptions(
+ common.WithProtocol(protocol),
+ common.WithPath(interfaceName),
+ common.WithParamsValue(constant.VersionKey, version),
+ )
+ invoker := &contextCaptureInvoker{BaseInvoker: base.NewBaseInvoker(url)}
+ methodConfig := &config.RestMethodConfig{
+ MethodName: "Handle",
+ PathParamsMap: map[int]string{},
+ QueryParamsMap: map[int]string{},
+ HeadersMap: map[int]string{},
+ Body: -1,
+ }
+
+ type contextKey struct{}
+ requestCtx := context.WithValue(context.Background(), contextKey{},
"request-value")
+ request := &testRestRequest{request:
httptest.NewRequestWithContext(requestCtx, http.MethodGet, "/", nil)}
+ response := &testRestResponse{ResponseRecorder: httptest.NewRecorder()}
+
+ GetRouteFunc(invoker, methodConfig)(request, response)
+
+ require.Equal(t, "request-value", invoker.ctx.Value(contextKey{}))
+ assertedInvocation, ok :=
invoker.lastInvocation.(*invocation.RPCInvocation)
+ require.True(t, ok)
+ require.Equal(t, "request-value",
assertedInvocation.Context().Value(contextKey{}))
+}
+
+func TestGetRouteFuncReturnsAfterParameterParseError(t *testing.T) {
+ const interfaceName = "RestContextParseErrorService"
+ const protocol = "rest"
+ const version = "context-parse-error"
+
+ _, err := common.ServiceMap.Register(interfaceName, protocol, "",
version, &RestContextService{})
+ require.NoError(t, err)
+ t.Cleanup(func() {
+ require.NoError(t, common.ServiceMap.UnRegister(interfaceName,
protocol, common.ServiceKey(interfaceName, "", version)))
+ })
+
+ url := common.NewURLWithOptions(
+ common.WithProtocol(protocol),
+ common.WithPath(interfaceName),
+ common.WithParamsValue(constant.VersionKey, version),
+ )
+ invoker := &contextCaptureInvoker{BaseInvoker: base.NewBaseInvoker(url)}
+ methodConfig := &config.RestMethodConfig{
+ MethodName: "Handle",
+ PathParamsMap: map[int]string{0: "missing"},
+ Body: -1,
+ }
+ request := &testRestRequest{request:
httptest.NewRequest(http.MethodGet, "/", nil)}
+ response := &testRestResponse{ResponseRecorder: httptest.NewRecorder()}
+
+ GetRouteFunc(invoker, methodConfig)(request, response)
+
+ require.Equal(t, 0, invoker.invocations)
+ require.Equal(t, http.StatusInternalServerError, response.Code)
+}
diff --git a/proxy/proxy.go b/proxy/proxy.go
index 296ff7780..3b64fa71f 100644
--- a/proxy/proxy.go
+++ b/proxy/proxy.go
@@ -182,6 +182,7 @@ func DefaultProxyImplementFunc(p *Proxy, v
common.RPCService) {
if !replyEmptyFlag {
inv.SetReply(reply.Interface())
}
+ inv.SetContext(invCtx)
for k, value := range p.attachments {
inv.SetAttachment(k, value)
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index cd65c1ecf..4676a25e7 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -18,6 +18,7 @@
package servicediscovery
import (
+ "context"
"errors"
"math/rand/v2"
"sort"
@@ -59,6 +60,8 @@ func init() {
// In order to keep compatible with interface-level registry，
// serviceDiscoveryRegistry = ServiceDiscovery + metadata
type serviceDiscoveryRegistry struct {
+ ctx context.Context
+ cancel context.CancelFunc
lock sync.RWMutex
url *common.URL
serviceDiscovery registry.ServiceDiscovery
@@ -76,7 +79,10 @@ func newServiceDiscoveryRegistry(url *common.URL)
(registry.Registry, error) {
if err != nil {
return nil, perrors.WithMessage(err, "Create service discovery
failed")
}
+ ctx, cancel := context.WithCancel(context.Background())
return &serviceDiscoveryRegistry{
+ ctx: ctx,
+ cancel: cancel,
url: url,
serviceDiscovery: serviceDiscovery,
instanceURLs:
make(map[registry.ServiceInstance]*common.URL),
@@ -341,6 +347,9 @@ func (s *serviceDiscoveryRegistry) IsAvailable() bool {
}
func (s *serviceDiscoveryRegistry) Destroy() {
+ if s.cancel != nil {
+ s.cancel()
+ }
s.stopMetadataTimers()
err := s.serviceDiscovery.Destroy()
if err != nil {
@@ -570,7 +579,7 @@ func (s *serviceDiscoveryRegistry) SubscribeURL(url
*common.URL, notify registry
// calls; holding the registry write lock across them would block every
// other subscribe/unsubscribe on this registry. The lock below only
guards
// the serviceListeners check/install, never the external work.
- listener :=
NewServiceInstancesChangedListener(url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
+ listener := NewServiceInstancesChangedListenerWithContext(s.ctx,
url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
for _, serviceNameTmp := range services.Values() {
serviceName := serviceNameTmp.(string)
instances := s.serviceDiscovery.GetInstances(serviceName)
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl.go
b/registry/servicediscovery/service_instances_changed_listener_impl.go
index 965b90bce..479a0cd11 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl.go
@@ -18,6 +18,7 @@
package servicediscovery
import (
+ "context"
"encoding/gob"
"reflect"
"sync"
@@ -59,6 +60,7 @@ func initCache(app string) {
// ServiceInstancesChangedListenerImpl The Service Discovery Changed Event
Listener
type ServiceInstancesChangedListenerImpl struct {
+ ctx context.Context
app string
registryId string
serviceNames *gxset.HashSet
@@ -70,10 +72,20 @@ type ServiceInstancesChangedListenerImpl struct {
}
func NewServiceInstancesChangedListener(app string, registryId string,
services *gxset.HashSet) registry.ServiceInstancesChangedListener {
+ return
NewServiceInstancesChangedListenerWithContext(context.Background(), app,
registryId, services)
+}
+
+// NewServiceInstancesChangedListenerWithContext creates a listener whose
+// metadata refreshes are canceled with ctx, such as when its registry closes.
+func NewServiceInstancesChangedListenerWithContext(ctx context.Context, app
string, registryId string, services *gxset.HashSet)
registry.ServiceInstancesChangedListener {
+ if ctx == nil {
+ ctx = context.Background()
+ }
cacheOnce.Do(func() {
initCache(app)
})
return &ServiceInstancesChangedListenerImpl{
+ ctx: ctx,
app: app,
registryId: registryId,
serviceNames: services,
@@ -128,7 +140,7 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
revisionToInstances[key] = append(subInstances,
instance)
metadataInfo := lstn.revisionToMetadata[key]
if metadataInfo == nil {
- meta, err := GetMetadataInfo(providerApp,
instance, revision, lstn.registryId)
+ meta, err :=
GetMetadataInfoWithContext(lstn.ctx, providerApp, instance, revision,
lstn.registryId)
if err != nil {
// Skip this instance if metadata fetch
fails (e.g., old Java Dubbo version)
// Try next instance with same revision
@@ -273,6 +285,15 @@ func metadataCacheKey(app, registryId, revision string)
string {
// and falls back to RPC if the report fails or returns nil. For all other
storage
// types (including absent), it uses RPC directly.
func GetMetadataInfo(app string, instance registry.ServiceInstance, revision
string, registryId string) (*info.MetadataInfo, error) {
+ return GetMetadataInfoWithContext(context.Background(), app, instance,
revision, registryId)
+}
+
+// GetMetadataInfoWithContext retrieves metadata using the supplied lifecycle
+// context for metadata RPC fallbacks.
+func GetMetadataInfoWithContext(ctx context.Context, app string, instance
registry.ServiceInstance, revision string, registryId string)
(*info.MetadataInfo, error) {
+ if ctx == nil {
+ ctx = context.Background()
+ }
cacheOnce.Do(func() {
initCache(app)
})
@@ -281,65 +302,80 @@ func GetMetadataInfo(app string, instance
registry.ServiceInstance, revision str
return metadataInfo.(*info.MetadataInfo), nil
}
- var metadataStorageType string
var metadataInfo *info.MetadataInfo
var err error
- if instance.GetMetadata() == nil {
- // No metadata map at all; treat as default (local/RPC) storage
type.
- metadataStorageType = constant.DefaultMetadataStorageType
+ if getMetadataStorageType(instance) ==
constant.RemoteMetadataStorageType {
+ metadataInfo, err = getRemoteMetadataInfo(ctx, app, instance,
revision, registryId)
} else {
- metadataStorageType =
instance.GetMetadata()[constant.MetadataStorageTypePropertyName]
- if metadataStorageType == "" {
- // MetadataStorageTypePropertyName absent (e.g. old
Java provider); default to local storage type.
- logger.Warnf("[Metadata] MetadataStorageType not set
for instance %s, defaulting to local", instance.GetID())
- metadataStorageType =
constant.DefaultMetadataStorageType
- }
+ metadataInfo, err = getMetadataInfoFromRPC(ctx, app, instance,
revision, registryId)
+ }
+ if err != nil {
+ return nil, err
}
+ metaCache.Set(cacheKey, metadataInfo)
+ return metadataInfo, nil
+}
- if metadataStorageType == constant.RemoteMetadataStorageType {
- var reportErr error
- metadataInfo, reportErr =
metadata.GetMetadataFromMetadataReport(revision, instance, registryId)
- if reportErr != nil {
- logger.Errorf("[Metadata][Fallback] report failed,
fallback to RPC app=%s registry=%s revision=%s err=%v",
- app, registryId, revision, reportErr)
- } else if metadataInfo == nil {
- logger.Warnf("[Metadata][Fallback] report returned nil
metadata, fallback to RPC app=%s registry=%s revision=%s",
- app, registryId, revision)
- } else {
- metaCache.Set(cacheKey, metadataInfo)
- return metadataInfo, nil
- }
+func getMetadataStorageType(instance registry.ServiceInstance) string {
+ instanceMetadata := instance.GetMetadata()
+ if instanceMetadata == nil {
+ return constant.DefaultMetadataStorageType
+ }
- metadataInfo, err = metadata.GetMetadataFromRpc(revision,
instance)
- if err != nil {
- if reportErr != nil {
- // Wrap rpcErr so callers can use errors.Is/As
on the primary failure;
- // reportErr is annotated as context since it
triggered the fallback.
- return nil, perrors.Wrapf(err,
- "both paths failed, reportErr: %v",
reportErr)
- }
- // reportErr was nil — the report returned nil metadata
and RPC also failed.
- return nil, perrors.Wrapf(err,
- "RPC fallback failed after report returned nil
metadata")
- }
- if metadataInfo == nil {
- return nil, perrors.Errorf("got nil metadata from RPC
app=%s registry=%s revision=%s",
- app, registryId, revision)
- }
- metaCache.Set(cacheKey, metadataInfo)
- return metadataInfo, nil
+ storageType :=
instanceMetadata[constant.MetadataStorageTypePropertyName]
+ if storageType == "" {
+ logger.Warnf("[Metadata] MetadataStorageType not set for
instance %s, defaulting to local", instance.GetID())
+ return constant.DefaultMetadataStorageType
}
+ return storageType
+}
- // Non-remote storage type ("local" or absent): fetch metadata via RPC
directly.
- metadataInfo, err = metadata.GetMetadataFromRpc(revision, instance)
+func getMetadataInfoFromRPC(ctx context.Context, app string, instance
registry.ServiceInstance, revision string, registryId string)
(*info.MetadataInfo, error) {
+ metadataInfo, err := metadata.GetMetadataFromRpcWithContext(ctx,
revision, instance)
if err != nil {
return nil, perrors.Wrapf(err,
"failed app=%s registry=%s revision=%s", app,
registryId, revision)
}
+ return requireMetadataInfo(metadataInfo, app, registryId, revision)
+}
+
+func getRemoteMetadataInfo(ctx context.Context, app string, instance
registry.ServiceInstance, revision string, registryId string)
(*info.MetadataInfo, error) {
+ metadataInfo, reportErr :=
metadata.GetMetadataFromMetadataReport(revision, instance, registryId)
+ if reportErr == nil && metadataInfo != nil {
+ return metadataInfo, nil
+ }
+ logMetadataReportFallback(app, registryId, revision, reportErr)
+
+ metadataInfo, rpcErr := metadata.GetMetadataFromRpcWithContext(ctx,
revision, instance)
+ if rpcErr != nil {
+ return nil, wrapMetadataRPCFallbackError(rpcErr, reportErr)
+ }
+ return requireMetadataInfo(metadataInfo, app, registryId, revision)
+}
+
+func logMetadataReportFallback(app, registryId, revision string, reportErr
error) {
+ if reportErr != nil {
+ logger.Errorf("[Metadata][Fallback] report failed, fallback to
RPC app=%s registry=%s revision=%s err=%v",
+ app, registryId, revision, reportErr)
+ return
+ }
+ logger.Warnf("[Metadata][Fallback] report returned nil metadata,
fallback to RPC app=%s registry=%s revision=%s",
+ app, registryId, revision)
+}
+
+func wrapMetadataRPCFallbackError(rpcErr, reportErr error) error {
+ if reportErr != nil {
+ // Wrap rpcErr so callers can use errors.Is/As on the primary
failure;
+ // reportErr is annotated as context since it triggered the
fallback.
+ return perrors.Wrapf(rpcErr, "both paths failed, reportErr:
%v", reportErr)
+ }
+ return perrors.Wrapf(rpcErr, "RPC fallback failed after report returned
nil metadata")
+}
+
+func requireMetadataInfo(metadataInfo *info.MetadataInfo, app, registryId,
revision string) (*info.MetadataInfo, error) {
if metadataInfo == nil {
return nil, perrors.Errorf("got nil metadata from RPC app=%s
registry=%s revision=%s",
app, registryId, revision)
}
- metaCache.Set(cacheKey, metadataInfo)
return metadataInfo, nil
}
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
index f8dcb5367..071627eac 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
@@ -18,6 +18,7 @@
package servicediscovery
import (
+ "context"
"fmt"
"testing"
)
@@ -38,6 +39,8 @@ import (
"dubbo.apache.org/dubbo-go/v3/metadata/info"
"dubbo.apache.org/dubbo-go/v3/metadata/mapping"
metadatareport "dubbo.apache.org/dubbo-go/v3/metadata/report"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ "dubbo.apache.org/dubbo-go/v3/protocol/result"
"dubbo.apache.org/dubbo-go/v3/registry"
)
@@ -431,6 +434,67 @@ func TestGetMetadataInfo_CacheKeyFormat(t *testing.T) {
assert.Equal(t, expectedMeta, meta)
}
+func TestServiceInstancesChangedListenerPropagatesLifecycleContext(t
*testing.T) {
+ const (
+ protocolName = "metadata-context-test"
+ providerApp = "metadata-context-provider"
+ revision = "metadata-context-revision"
+ )
+ captured := &listenerContextInvoker{
+ BaseInvoker:
*base.NewBaseInvoker(common.NewURLWithOptions(common.WithProtocol(protocolName))),
+ }
+ extension.SetProtocol(protocolName, func() base.Protocol {
+ return &listenerContextProtocol{invoker: captured}
+ })
+
+ cacheKey := metadataCacheKey(providerApp, constant.DefaultKey, revision)
+ t.Cleanup(func() { metaCache.Delete(cacheKey) })
+ ctx, cancel := context.WithCancel(context.Background())
+ listener := NewServiceInstancesChangedListenerWithContext(ctx, testApp,
constant.DefaultKey, gxset.NewSet(providerApp))
+ instance := ®istry.DefaultServiceInstance{
+ ID: "127.0.0.1:20099",
+ ServiceName: providerApp,
+ Host: "127.0.0.1",
+ Port: 20099,
+ Metadata: map[string]string{
+ constant.ExportedServicesRevisionPropertyName: revision,
+ constant.MetadataServiceURLParamsPropertyName:
`{"protocol":"` + protocolName + `","port":"20880"}`,
+ },
+ }
+
+ require.NoError(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(providerApp,
[]registry.ServiceInstance{instance})))
+ require.Same(t, ctx, captured.ctx)
+ cancel()
+ require.ErrorIs(t, captured.ctx.Err(), context.Canceled)
+}
+
+type listenerContextProtocol struct {
+ invoker base.Invoker
+}
+
+func (p *listenerContextProtocol) Export(base.Invoker) base.Exporter {
+ return nil
+}
+
+func (p *listenerContextProtocol) Refer(*common.URL) base.Invoker {
+ return p.invoker
+}
+
+func (p *listenerContextProtocol) Destroy() {}
+
+type listenerContextInvoker struct {
+ base.BaseInvoker
+ ctx context.Context
+}
+
+func (i *listenerContextInvoker) Invoke(ctx context.Context, invocation
base.Invocation) result.Result {
+ i.ctx = ctx
+ if reply, ok := invocation.Reply().(*any); ok {
+ *reply = &info.MetadataInfo{App: "metadata-context-provider"}
+ }
+ return &result.RPCResult{}
+}
+
func TestGetMetadataInfo_LocalStorageGoesDirectlyToRPC(t *testing.T) {
// Ensure cache is initialized
_ = NewServiceInstancesChangedListener(testApp, constant.DefaultKey,
gxset.NewSet(testApp))
diff --git a/remoting/exchange_client.go b/remoting/exchange_client.go
index 6de4833e3..979ad1260 100644
--- a/remoting/exchange_client.go
+++ b/remoting/exchange_client.go
@@ -18,6 +18,7 @@
package remoting
import (
+ "context"
"errors"
"time"
)
@@ -56,6 +57,10 @@ type Client interface {
IsAvailable() bool
}
+type contextRequester interface {
+ RequestContext(ctx context.Context, request *Request, timeout
time.Duration, response *PendingResponse) error
+}
+
// ExchangeClient is abstraction level. it is like facade.
type ExchangeClient struct {
ConnectTimeout time.Duration // timeout for connecting server
@@ -116,6 +121,19 @@ func (client *ExchangeClient) GetActiveNumber() uint32 {
// Request means two way request.
func (client *ExchangeClient) Request(invocation *base.Invocation, url
*common.URL, timeout time.Duration,
res *result.RPCResult) error {
+ return client.RequestContext(context.Background(), invocation, url,
timeout, res)
+}
+
+// RequestContext sends a two-way request and stops waiting when ctx is
canceled.
+func (client *ExchangeClient) RequestContext(ctx context.Context, invocation
*base.Invocation, url *common.URL, timeout time.Duration,
+ res *result.RPCResult) error {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if err := ctx.Err(); err != nil {
+ res.Err = err
+ return err
+ }
if er := client.doInit(url); er != nil {
return er
}
@@ -129,7 +147,7 @@ func (client *ExchangeClient) Request(invocation
*base.Invocation, url *common.U
rsp.Reply = (*invocation).Reply()
AddPendingResponse(rsp)
- err := client.client.Request(request, timeout, rsp)
+ err := client.requestContext(ctx, request, timeout, rsp)
// request error
if err != nil {
RemovePendingResponse(SequenceType(request.ID))
@@ -147,9 +165,29 @@ func (client *ExchangeClient) Request(invocation
*base.Invocation, url *common.U
return nil
}
+func (client *ExchangeClient) requestContext(ctx context.Context, request
*Request, timeout time.Duration, response *PendingResponse) error {
+ if requester, ok := client.client.(contextRequester); ok {
+ return requester.RequestContext(ctx, request, timeout, response)
+ }
+ return client.client.Request(request, timeout, response)
+}
+
// AsyncRequest async two way request.
func (client *ExchangeClient) AsyncRequest(invocation *base.Invocation, url
*common.URL, timeout time.Duration,
callback common.AsyncCallback, result *result.RPCResult) error {
+ return client.AsyncRequestContext(context.Background(), invocation,
url, timeout, callback, result)
+}
+
+// AsyncRequestContext sends an asynchronous two-way request with ctx.
+func (client *ExchangeClient) AsyncRequestContext(ctx context.Context,
invocation *base.Invocation, url *common.URL, timeout time.Duration,
+ callback common.AsyncCallback, result *result.RPCResult) error {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if err := ctx.Err(); err != nil {
+ result.Err = err
+ return err
+ }
if er := client.doInit(url); er != nil {
return er
}
@@ -164,7 +202,7 @@ func (client *ExchangeClient) AsyncRequest(invocation
*base.Invocation, url *com
rsp.Reply = (*invocation).Reply()
AddPendingResponse(rsp)
- err := client.client.Request(request, timeout, rsp)
+ err := client.requestContext(ctx, request, timeout, rsp)
if err != nil {
RemovePendingResponse(SequenceType(request.ID))
result.Err = err
@@ -176,6 +214,17 @@ func (client *ExchangeClient) AsyncRequest(invocation
*base.Invocation, url *com
// Send sends oneway request.
func (client *ExchangeClient) Send(invocation *base.Invocation, url
*common.URL, timeout time.Duration) error {
+ return client.SendContext(context.Background(), invocation, url,
timeout)
+}
+
+// SendContext sends a one-way request with ctx.
+func (client *ExchangeClient) SendContext(ctx context.Context, invocation
*base.Invocation, url *common.URL, timeout time.Duration) error {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
if er := client.doInit(url); er != nil {
return er
}
@@ -187,7 +236,7 @@ func (client *ExchangeClient) Send(invocation
*base.Invocation, url *common.URL,
rsp := NewPendingResponse(request.ID)
rsp.response = NewResponse(request.ID, "2.0.2")
- err := client.client.Request(request, timeout, rsp)
+ err := client.requestContext(ctx, request, timeout, rsp)
if err != nil {
return err
}
diff --git a/remoting/exchange_client_test.go b/remoting/exchange_client_test.go
index 4a829c5d9..57ac9604a 100644
--- a/remoting/exchange_client_test.go
+++ b/remoting/exchange_client_test.go
@@ -18,6 +18,7 @@
package remoting
import (
+ "context"
"errors"
"sync"
"testing"
@@ -37,11 +38,13 @@ import (
)
type mockClient struct {
- mu sync.Mutex
- available bool
- connectErr error
- connCount int
- requestErr error
+ mu sync.Mutex
+ available bool
+ connectErr error
+ connCount int
+ requestErr error
+ contextRequestStarted chan struct{}
+ blockContextRequest bool
}
func (m *mockClient) SetExchangeClient(client *ExchangeClient) {}
@@ -54,6 +57,18 @@ func (m *mockClient) Request(request *Request, timeout
time.Duration, response *
return m.requestErr
}
+func (m *mockClient) RequestContext(ctx context.Context, request *Request,
timeout time.Duration, response *PendingResponse) error {
+ if m.contextRequestStarted != nil {
+ close(m.contextRequestStarted)
+ m.contextRequestStarted = nil
+ }
+ if m.blockContextRequest {
+ <-ctx.Done()
+ return ctx.Err()
+ }
+ return m.Request(request, timeout, response)
+}
+
func (m *mockClient) Connect(url *common.URL) error {
m.mu.Lock()
defer m.mu.Unlock()
@@ -150,6 +165,41 @@ func TestExchangeClientAsyncRequestErrorCleanup(t
*testing.T) {
assert.Equal(t, before, countPendingResponses(), "pendingResponses
leaked on AsyncRequest error path")
}
+func TestExchangeClientRequestContextCancellationCleanup(t *testing.T) {
+ requestStarted := make(chan struct{})
+ m := &mockClient{
+ available: true,
+ contextRequestStarted: requestStarted,
+ blockContextRequest: true,
+ }
+ ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+ before := countPendingResponses()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ res := &result.RPCResult{}
+ errCh := make(chan error, 1)
+ go func() {
+ errCh <- ec.RequestContext(ctx, newTestInvocation(), testURL(),
time.Minute, res)
+ }()
+
+ select {
+ case <-requestStarted:
+ case <-time.After(time.Second):
+ t.Fatal("request was not sent to the transport")
+ }
+ cancel()
+
+ select {
+ case err := <-errCh:
+ require.ErrorIs(t, err, context.Canceled)
+ require.ErrorIs(t, res.Err, context.Canceled)
+ case <-time.After(time.Second):
+ t.Fatal("request did not stop after context cancellation")
+ }
+ assert.Equal(t, before, countPendingResponses(), "pendingResponses
leaked after context cancellation")
+}
+
// countPendingResponses returns the number of entries currently held in the
global
// pendingResponses map.
func countPendingResponses() int {
diff --git a/remoting/getty/getty_client.go b/remoting/getty/getty_client.go
index 8fd452101..ee6432d36 100644
--- a/remoting/getty/getty_client.go
+++ b/remoting/getty/getty_client.go
@@ -18,6 +18,7 @@
package getty
import (
+ "context"
"math/rand"
"sync"
"time"
@@ -28,7 +29,6 @@ import (
"github.com/dubbogo/gost/log/logger"
gxsync "github.com/dubbogo/gost/sync"
- gxtime "github.com/dubbogo/gost/time"
perrors "github.com/pkg/errors"
@@ -215,6 +215,17 @@ func (c *Client) Close() {
// Request send request
func (c *Client) Request(request *remoting.Request, timeout time.Duration,
response *remoting.PendingResponse) error {
+ return c.RequestContext(context.Background(), request, timeout,
response)
+}
+
+// RequestContext sends a request and stops waiting when ctx is canceled.
+func (c *Client) RequestContext(ctx context.Context, request
*remoting.Request, timeout time.Duration, response *remoting.PendingResponse)
error {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
if timeout <= 0 {
timeout = c.opts.RequestTimeout
}
@@ -241,11 +252,15 @@ func (c *Client) Request(request *remoting.Request,
timeout time.Duration, respo
return nil
}
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
select {
- case <-gxtime.After(timeout):
+ case <-timer.C:
return perrors.WithStack(errClientReadTimeout)
case <-response.Done:
err = response.Err
+ case <-ctx.Done():
+ return perrors.WithStack(ctx.Err())
}
return perrors.WithStack(err)
