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 910b3895d Refactor/triple comments helpers (#3643)
910b3895d is described below
commit 910b3895d32540d094e2243e267da3b23acf3426
Author: 翎 <[email protected]>
AuthorDate: Tue Aug 11 23:13:18 2026 +0800
Refactor/triple comments helpers (#3643)
* refactor(triple): clarify client keepalive helpers
* refactor(triple): rename server handler option helper
* docs(triple): formalize handler flow comments
* refactor(triple): align client KeepAlive naming
* docs(triple): restore TODO comments
---
protocol/triple/client.go | 52 +++++++++++++++---------------
protocol/triple/client_test.go | 4 +--
protocol/triple/server.go | 48 +++++++++++++--------------
protocol/triple/server_test.go | 4 +--
protocol/triple/triple_protocol/handler.go | 39 +++++++++++-----------
5 files changed, 72 insertions(+), 75 deletions(-)
diff --git a/protocol/triple/client.go b/protocol/triple/client.go
index 8d7491d75..382b3fa79 100644
--- a/protocol/triple/client.go
+++ b/protocol/triple/client.go
@@ -54,9 +54,8 @@ const (
httpsPrefix string = "https://";
)
-// clientManager wraps triple clients and is responsible for find concrete
triple client to invoke
-// callUnary, callClientStream, callServerStream, callBidiStream.
-// A Reference has a clientManager.
+// clientManager owns the service-level Triple clients used by unary,
streaming,
+// and health-check calls for a Reference.
type clientManager struct {
isIDL bool
triClient *tri.Client
@@ -107,12 +106,13 @@ func (cm *clientManager) close() error {
return nil
}
-// newClientManager extracts configurations from url and builds clientManager
+// newClientManager resolves URL and global Triple settings, then builds the
+// service and health clients that share one HTTP transport.
func newClientManager(url *common.URL) (*clientManager, error) {
var cliOpts []tri.ClientOption
var isIDL bool
- // Set serialization
+ // Resolve codec options before constructing the transport-backed
client.
serialization := url.GetParam(constant.SerializationKey,
constant.ProtobufSerialization)
switch serialization {
case constant.ProtobufSerialization:
@@ -128,18 +128,18 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
panic(fmt.Sprintf("Unsupported serialization: %s",
serialization))
}
- // Set timeout
+ // Apply the call timeout configured on the reference URL.
timeout := url.GetParamDuration(constant.TimeoutKey, "")
cliOpts = append(cliOpts, tri.WithTimeout(timeout))
- // Set service group and version
+ // Pass group and version through request headers for service selection.
group := url.GetParam(constant.GroupKey, "")
version := url.GetParam(constant.VersionKey, "")
cliOpts = append(cliOpts, tri.WithGroup(group),
tri.WithVersion(version))
// TODO(DMwangnima): support OpenTracing
- // Handle TLS
+ // Resolve TLS first because HTTP/2 and HTTP/3 transport setup depends
on it.
var (
tlsFlag bool
tlsConf *global.TLSConfig
@@ -172,15 +172,15 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
tripleConf = tripleConfRaw.(*global.TripleConfig)
}
- // Handle keepalive options
- cliKeepAliveOpts, keepAliveInterval, keepAliveTimeout,
genKeepAliveOptsErr := genKeepAliveOptions(url, tripleConf)
- if genKeepAliveOptsErr != nil {
- logger.Errorf("[Triple][Client] genKeepAliveOpts failed,
err=%v", genKeepAliveOptsErr)
- return nil, genKeepAliveOptsErr
+ // Resolve keepalive and size-limit options before choosing the
transport.
+ clientKeepAliveOpts, keepAliveInterval, keepAliveTimeout, keepAliveErr
:= resolveClientKeepAliveOptions(url, tripleConf)
+ if keepAliveErr != nil {
+ logger.Errorf("[Triple][Client] genKeepAliveOpts failed,
err=%v", keepAliveErr)
+ return nil, keepAliveErr
}
- cliOpts = append(cliOpts, cliKeepAliveOpts...)
+ cliOpts = append(cliOpts, clientKeepAliveOpts...)
- // Handle HTTP transport of triple protocol
+ // Build the HTTP transport used by the Triple client.
var transport http.RoundTripper
var callProtocol string
@@ -192,10 +192,9 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
}
switch callProtocol {
- // This case might be for backward compatibility,
- // It's not useful for the Triple protocol, HTTP/1 lacks trailer
functionality.
- // Triple protocol only supports HTTP/2 and HTTP/3.
case constant.CallHTTP:
+ // Backward compatibility path for callers that still request
HTTP/1.1.
+ // Triple itself requires HTTP/2 or HTTP/3 trailer support.
transport = &http.Transport{
TLSClientConfig: cfg,
}
@@ -245,7 +244,8 @@ func newClientManager(url *common.URL) (*clientManager,
error) {
return nil, fmt.Errorf("TRIPLE HTTP/2 and HTTP/3 client
must have TLS config, but TLS config is nil")
}
- // Create a dual transport that can handle both HTTP/2 and
HTTP/3
+ // Dual transport lets the client negotiate HTTP/2 or HTTP/3
with the
+ // same URL and keepalive settings.
transport = newDualTransport(cfg, keepAliveInterval,
keepAliveTimeout)
logger.Info("[Triple][Client] triple HTTP/2 and HTTP/3 client
transport init successfully")
default:
@@ -296,20 +296,20 @@ func (cm *clientManager) callHealthWatch(ctx
context.Context, service string) (*
return stream, nil
}
-func genKeepAliveOptions(url *common.URL, tripleConf *global.TripleConfig)
([]tri.ClientOption, time.Duration, time.Duration, error) {
- var cliKeepAliveOpts []tri.ClientOption
+func resolveClientKeepAliveOptions(url *common.URL, tripleConf
*global.TripleConfig) ([]tri.ClientOption, time.Duration, time.Duration, error)
{
+ var clientKeepAliveOpts []tri.ClientOption
- // Set max send and recv msg size
+ // Apply client message-size limits from URL compatibility parameters.
maxCallRecvMsgSize := constant.DefaultMaxCallRecvMsgSize
if recvMsgSize, err :=
humanize.ParseBytes(url.GetParam(constant.MaxCallRecvMsgSize, "")); err == nil
&& recvMsgSize > 0 {
maxCallRecvMsgSize = int(recvMsgSize)
}
- cliKeepAliveOpts = append(cliKeepAliveOpts,
tri.WithReadMaxBytes(maxCallRecvMsgSize))
+ clientKeepAliveOpts = append(clientKeepAliveOpts,
tri.WithReadMaxBytes(maxCallRecvMsgSize))
maxCallSendMsgSize := constant.DefaultMaxCallSendMsgSize
if sendMsgSize, err :=
humanize.ParseBytes(url.GetParam(constant.MaxCallSendMsgSize, "")); err == nil
&& sendMsgSize > 0 {
maxCallSendMsgSize = int(sendMsgSize)
}
- cliKeepAliveOpts = append(cliKeepAliveOpts,
tri.WithSendMaxBytes(maxCallSendMsgSize))
+ clientKeepAliveOpts = append(clientKeepAliveOpts,
tri.WithSendMaxBytes(maxCallSendMsgSize))
// Set keepalive interval and keepalive timeout
// Compatibility: read the legacy URL keepalive parameters.
@@ -318,7 +318,7 @@ func genKeepAliveOptions(url *common.URL, tripleConf
*global.TripleConfig) ([]tr
keepAliveTimeout := url.GetParamDuration(constant.KeepAliveTimeout,
constant.DefaultKeepAliveTimeout)
if tripleConf == nil {
- return cliKeepAliveOpts, keepAliveInterval, keepAliveTimeout,
nil
+ return clientKeepAliveOpts, keepAliveInterval,
keepAliveTimeout, nil
}
var parseErr error
@@ -336,5 +336,5 @@ func genKeepAliveOptions(url *common.URL, tripleConf
*global.TripleConfig) ([]tr
}
}
- return cliKeepAliveOpts, keepAliveInterval, keepAliveTimeout, nil
+ return clientKeepAliveOpts, keepAliveInterval, keepAliveTimeout, nil
}
diff --git a/protocol/triple/client_test.go b/protocol/triple/client_test.go
index 277180bc5..bebf97518 100644
--- a/protocol/triple/client_test.go
+++ b/protocol/triple/client_test.go
@@ -195,7 +195,7 @@ func
TestClientManagerCallUnaryCopiesErrorResponseMetadata(t *testing.T) {
// TestClientManager_CallMethods_MissingClient removed - no longer applicable
// in the service-level client architecture where all methods share a single
triClient.
-func Test_genKeepAliveOptions(t *testing.T) {
+func Test_resolveClientKeepAliveOptions(t *testing.T) {
defaultInterval, _ :=
time.ParseDuration(constant.DefaultKeepAliveInterval)
defaultTimeout, _ :=
time.ParseDuration(constant.DefaultKeepAliveTimeout)
@@ -282,7 +282,7 @@ func Test_genKeepAliveOptions(t *testing.T) {
for _, test := range tests {
t.Run(test.desc, func(t *testing.T) {
- opts, interval, timeout, err :=
genKeepAliveOptions(test.url, test.tripleConf)
+ opts, interval, timeout, err :=
resolveClientKeepAliveOptions(test.url, test.tripleConf)
if test.expectErr {
require.Error(t, err)
} else {
diff --git a/protocol/triple/server.go b/protocol/triple/server.go
index b1963cccf..9200263a3 100644
--- a/protocol/triple/server.go
+++ b/protocol/triple/server.go
@@ -240,7 +240,7 @@ func (s *Server) startTransport(callProtocol string,
tlsConf *tls.Config) {
}()
}
-func (s *Server) registerServiceHandlers(invoker base.Invoker, info
*common.ServiceInfo, hanOpts []tri.HandlerOption) {
+func (s *Server) registerServiceHandlers(invoker base.Invoker, info
*common.ServiceInfo, handlerOpts []tri.HandlerOption) {
url := invoker.GetURL()
// IDLMode means that this will only be set when
@@ -264,15 +264,15 @@ func (s *Server) registerServiceHandlers(invoker
base.Invoker, info *common.Serv
if info != nil {
// New triple IDL mode
- s.handleServiceWithInfo(intfName, invoker, info, hanOpts...)
+ s.handleServiceWithInfo(intfName, invoker, info, handlerOpts...)
s.saveServiceInfo(intfName, info, openapiGroup, url.Group(),
url.Version())
} else if IDLMode == constant.NONIDL {
// New triple non-IDL mode
reflectInfo := createServiceInfoWithReflection(service)
- s.handleServiceWithInfo(intfName, invoker, reflectInfo,
hanOpts...)
+ s.handleServiceWithInfo(intfName, invoker, reflectInfo,
handlerOpts...)
s.saveServiceInfo(intfName, reflectInfo, openapiGroup,
url.Group(), url.Version())
} else {
- s.compatHandleService(url, intfName, url.Group(),
url.Version(), hanOpts...)
+ s.compatHandleService(url, intfName, url.Group(),
url.Version(), handlerOpts...)
}
}
@@ -289,12 +289,12 @@ func (s *Server) RefreshService(invoker base.Invoker,
info *common.ServiceInfo)
func (s *Server) refreshService(invoker base.Invoker, info
*common.ServiceInfo) error {
url := invoker.GetURL()
- hanOpts, err := resolveHandlerOptions(url)
+ handlerOpts, err := resolveHandlerOptions(url)
if err != nil {
return err
}
- s.registerServiceHandlers(invoker, info, hanOpts)
+ s.registerServiceHandlers(invoker, info, handlerOpts)
return nil
}
@@ -376,15 +376,15 @@ func resolveHandlerOptions(url *common.URL)
([]tri.HandlerOption, error) {
return nil, err
}
- hanOpts := getHanOpts(url, tripleConf)
- hanOpts = append(hanOpts, tri.WithExpectedCodecName(serialization))
- return hanOpts, nil
+ handlerOpts := buildServerHandlerOptions(url, tripleConf)
+ handlerOpts = append(handlerOpts,
tri.WithExpectedCodecName(serialization))
+ return handlerOpts, nil
}
-func getHanOpts(url *common.URL, tripleConf *global.TripleConfig) (hanOpts
[]tri.HandlerOption) {
+func buildServerHandlerOptions(url *common.URL, tripleConf
*global.TripleConfig) (handlerOpts []tri.HandlerOption) {
group := url.GetParam(constant.GroupKey, "")
version := url.GetParam(constant.VersionKey, "")
- hanOpts = append(hanOpts, tri.WithGroup(group),
tri.WithVersion(version))
+ handlerOpts = append(handlerOpts, tri.WithGroup(group),
tri.WithVersion(version))
// Compatibility: read the legacy URL receive-size parameter.
// TODO: remove MaxServerRecvMsgSize in version 4.0.0.
@@ -392,7 +392,7 @@ func getHanOpts(url *common.URL, tripleConf
*global.TripleConfig) (hanOpts []tri
if recvMsgSize, convertErr :=
humanize.ParseBytes(url.GetParam(constant.MaxServerRecvMsgSize, ""));
convertErr == nil && recvMsgSize != 0 {
maxServerRecvMsgSize = int(recvMsgSize)
}
- hanOpts = append(hanOpts, tri.WithReadMaxBytes(maxServerRecvMsgSize))
+ handlerOpts = append(handlerOpts,
tri.WithReadMaxBytes(maxServerRecvMsgSize))
// Compatibility: read the legacy URL send-size parameter.
// TODO: remove MaxServerSendMsgSize in version 4.0.0.
@@ -400,10 +400,10 @@ func getHanOpts(url *common.URL, tripleConf
*global.TripleConfig) (hanOpts []tri
if sendMsgSize, convertErr :=
humanize.ParseBytes(url.GetParam(constant.MaxServerSendMsgSize, ""));
convertErr == nil && sendMsgSize != 0 {
maxServerSendMsgSize = int(sendMsgSize)
}
- hanOpts = append(hanOpts, tri.WithSendMaxBytes(maxServerSendMsgSize))
+ handlerOpts = append(handlerOpts,
tri.WithSendMaxBytes(maxServerSendMsgSize))
if tripleConf == nil {
- return hanOpts
+ return handlerOpts
}
if tripleConf.MaxServerRecvMsgSize != "" {
@@ -411,7 +411,7 @@ func getHanOpts(url *common.URL, tripleConf
*global.TripleConfig) (hanOpts []tri
if recvMsgSize, convertErr :=
humanize.ParseBytes(tripleConf.MaxServerRecvMsgSize); convertErr == nil &&
recvMsgSize != 0 {
maxServerRecvMsgSize = int(recvMsgSize)
}
- hanOpts = append(hanOpts,
tri.WithReadMaxBytes(maxServerRecvMsgSize))
+ handlerOpts = append(handlerOpts,
tri.WithReadMaxBytes(maxServerRecvMsgSize))
}
if tripleConf.MaxServerSendMsgSize != "" {
@@ -419,14 +419,14 @@ func getHanOpts(url *common.URL, tripleConf
*global.TripleConfig) (hanOpts []tri
if sendMsgSize, convertErr :=
humanize.ParseBytes(tripleConf.MaxServerSendMsgSize); convertErr == nil &&
sendMsgSize != 0 {
maxServerSendMsgSize = int(sendMsgSize)
}
- hanOpts = append(hanOpts,
tri.WithSendMaxBytes(maxServerSendMsgSize))
+ handlerOpts = append(handlerOpts,
tri.WithSendMaxBytes(maxServerSendMsgSize))
}
// TODO: support OpenTracing
// CORS configuration
if tripleConf.Cors != nil && len(tripleConf.Cors.AllowOrigins) > 0 {
- hanOpts = append(hanOpts, tri.WithCORS(&tri.CorsConfig{
+ handlerOpts = append(handlerOpts, tri.WithCORS(&tri.CorsConfig{
AllowOrigins: tripleConf.Cors.AllowOrigins,
AllowMethods: tripleConf.Cors.AllowMethods,
AllowHeaders: tripleConf.Cors.AllowHeaders,
@@ -436,7 +436,7 @@ func getHanOpts(url *common.URL, tripleConf
*global.TripleConfig) (hanOpts []tri
}))
}
- return hanOpts
+ return handlerOpts
}
// *Important*, this function is responsible for being compatible with old
triple-gen code and non-IDL code
@@ -487,7 +487,7 @@ func (s *Server) compatHandleService(url *common.URL,
interfaceName string, grou
func (s *Server) compatRegisterHandler(interfaceName string, svc
dubbo3.Dubbo3GrpcService, opts ...tri.HandlerOption) {
desc := svc.XXX_ServiceDesc()
- // Init unary handlers
+ // Register compat unary handlers from generated descriptors.
for _, method := range desc.Methods {
// Please refer to
protocol/triple/internal/proto/triple_gen/greettriple for procedure examples
// Error could be ignored because base is empty string
@@ -497,7 +497,7 @@ func (s *Server) compatRegisterHandler(interfaceName
string, svc dubbo3.Dubbo3Gr
}
}
- // Init stream handlers
+ // Register compat stream handlers from generated descriptors.
for _, stream := range desc.Streams {
// Please refer to
protocol/triple/internal/proto/triple_gen/greettriple for procedure examples
// Error could be ignored because base is empty string
@@ -549,7 +549,7 @@ func (s *Server) registerUnaryMethodHandler(procedure
string, m common.MethodInf
func(ctx context.Context, req *tri.Request) (*tri.Response,
error) {
args := extractUnaryInvocationArgs(req.Msg)
attachments := generateAttachments(req.Header())
- // Inject attachments
+ // Make incoming attachments available to invocation
filters and user code.
ctx = context.WithValue(ctx, constant.AttachmentKey,
attachments)
invo := invocation.NewRPCInvocation(m.Name, args,
attachments)
res := invoker.Invoke(ctx, invo)
@@ -572,7 +572,7 @@ func (s *Server)
registerClientStreamMethodHandler(procedure string, m common.Me
func(ctx context.Context, stream *tri.ClientStream)
(*tri.Response, error) {
args := []any{m.StreamInitFunc(stream)}
attachments :=
generateAttachments(stream.RequestHeader())
- // Inject attachments
+ // Make incoming attachments available to invocation
filters and user code.
ctx = context.WithValue(ctx, constant.AttachmentKey,
attachments)
invo := invocation.NewRPCInvocation(m.Name, args,
attachments)
res := invoker.Invoke(ctx, invo)
@@ -592,7 +592,7 @@ func (s *Server)
registerServerStreamMethodHandler(procedure string, m common.Me
func(ctx context.Context, req *tri.Request, stream
*tri.ServerStream) error {
args := []any{req.Msg, m.StreamInitFunc(stream)}
attachments := generateAttachments(req.Header())
- // Inject attachments
+ // Make incoming attachments available to invocation
filters and user code.
ctx = context.WithValue(ctx, constant.AttachmentKey,
attachments)
invo := invocation.NewRPCInvocation(m.Name, args,
attachments)
res := invoker.Invoke(ctx, invo)
@@ -611,7 +611,7 @@ func (s *Server) registerBidiStreamMethodHandler(procedure
string, m common.Meth
func(ctx context.Context, stream *tri.BidiStream) error {
args := []any{m.StreamInitFunc(stream)}
attachments :=
generateAttachments(stream.RequestHeader())
- // Inject attachments
+ // Make incoming attachments available to invocation
filters and user code.
ctx = context.WithValue(ctx, constant.AttachmentKey,
attachments)
invo := invocation.NewRPCInvocation(m.Name, args,
attachments)
res := invoker.Invoke(ctx, invo)
diff --git a/protocol/triple/server_test.go b/protocol/triple/server_test.go
index eba6ca459..537e01a50 100644
--- a/protocol/triple/server_test.go
+++ b/protocol/triple/server_test.go
@@ -358,7 +358,7 @@ func TestServer_SaveServiceInfo_Concurrent(t *testing.T) {
assert.Len(t, server.GetServiceInfo(), concurrency)
}
-func Test_getHanOpts(t *testing.T) {
+func Test_buildServerHandlerOptions(t *testing.T) {
tests := []struct {
desc string
url *common.URL
@@ -402,7 +402,7 @@ func Test_getHanOpts(t *testing.T) {
for _, test := range tests {
t.Run(test.desc, func(t *testing.T) {
- opts := getHanOpts(test.url, test.tripleConf)
+ opts := buildServerHandlerOptions(test.url,
test.tripleConf)
assert.Len(t, opts, test.expectLen)
})
}
diff --git a/protocol/triple/triple_protocol/handler.go
b/protocol/triple/triple_protocol/handler.go
index 8b8abceb3..73e0fdc30 100644
--- a/protocol/triple/triple_protocol/handler.go
+++ b/protocol/triple/triple_protocol/handler.go
@@ -74,7 +74,7 @@ func generateUnaryHandlerFunc(
) StreamingHandlerFunc {
// Wrap the strongly-typed implementation so we can apply interceptors.
untyped := UnaryHandlerFunc(func(ctx context.Context, request
AnyRequest) (AnyResponse, error) {
- // Verify err
+ // Honor cancellation before reading from the transport stream.
if err := ctx.Err(); err != nil {
return nil, err
}
@@ -94,26 +94,25 @@ func generateUnaryHandlerFunc(
if interceptor != nil {
untyped = interceptor.WrapUnaryHandler(untyped)
}
- // Receive and send
- // Conn should be responsible for marshal and unmarshal
- // Given a stream, how should we call the unary function?
+ // The transport stream owns framing, decoding, and encoding. This
adapter
+ // receives the request, invokes the unary handler, and writes the
response.
implementation := func(ctx context.Context, conn StreamingHandlerConn)
error {
req := reqInitFunc()
if err := conn.Receive(req); err != nil {
return err
}
- // Wrap the specific msg
+ // Build the request envelope with stream metadata for handlers
and interceptors.
request := NewRequest(req)
request.spec = conn.Spec()
request.peer = conn.Peer()
request.header = conn.RequestHeader()
- // Embed header in context so that user logic could process
them via FromIncomingContext
+ // Expose incoming headers through context for user logic via
FromIncomingContext.
ctx = newIncomingContext(ctx, conn.RequestHeader())
ctx = context.WithValue(ctx, handlerOutgoingKey{}, conn)
response, err := untyped(ctx, request)
- // Write the server-side return-attachment-data in the trailer
to send to the caller
+ // Propagate outgoing context values as response trailers.
if data := ExtractFromOutgoingContext(ctx); data != nil {
mergeHeaders(conn.ResponseTrailer(), data)
}
@@ -122,7 +121,7 @@ func generateUnaryHandlerFunc(
return err
}
- // Merge headers
+ // Propagate application headers and trailers before sending
the payload.
mergeHeaders(conn.ResponseHeader(), response.Header())
mergeHeaders(conn.ResponseTrailer(), response.Trailer())
return conn.Send(response.Any())
@@ -161,7 +160,7 @@ func generateClientStreamHandlerFunc(
) StreamingHandlerFunc {
implementation := func(ctx context.Context, conn StreamingHandlerConn)
error {
stream := &ClientStream{conn: conn}
- // Embed header in context so that user logic could process
them via FromIncomingContext
+ // Expose incoming headers through context for user logic via
FromIncomingContext.
ctx = newIncomingContext(ctx, conn.RequestHeader())
res, err := streamFunc(ctx, stream)
@@ -224,7 +223,7 @@ func generateServerStreamHandlerFunc(
if err := conn.Receive(req); err != nil {
return err
}
- // Embed header in context so that user logic could process
them via FromIncomingContext
+ // Expose incoming headers through context for user logic via
FromIncomingContext.
ctx = newIncomingContext(ctx, conn.RequestHeader())
err := streamFunc(
ctx,
@@ -280,7 +279,7 @@ func generateBidiStreamHandlerFunc(
interceptor Interceptor,
) StreamingHandlerFunc {
implementation := func(ctx context.Context, conn StreamingHandlerConn)
error {
- // Embed header in context so that user logic could process
them via FromIncomingContext
+ // Expose incoming headers through context for user logic via
FromIncomingContext.
ctx = newIncomingContext(ctx, conn.RequestHeader())
err := streamFunc(ctx, &BidiStream{conn: conn})
if err != nil {
@@ -318,14 +317,14 @@ func (h *Handler) ServeHTTP(responseWriter
http.ResponseWriter, request *http.Re
return
}
- // CORS handling
+ // Apply CORS policy before protocol negotiation.
if h.cors != nil {
if h.handleCORS(responseWriter, request) {
return
}
}
- // Inspect headers
+ // Filter protocol handlers by HTTP method before inspecting the
payload.
var protocolHandlers []protocolHandler
for _, handler := range h.protocolHandlers {
if _, ok := handler.Methods()[request.Method]; ok {
@@ -341,8 +340,7 @@ func (h *Handler) ServeHTTP(responseWriter
http.ResponseWriter, request *http.Re
contentType :=
canonicalizeContentType(getHeaderCanonical(request.Header, headerContentType))
- // Inspect contentType
- // Find our implementation of the RPC protocol in use.
+ // Select the protocol handler that can decode this content type.
var protocolHdl protocolHandler
for _, handler := range protocolHandlers {
if handler.CanHandlePayload(request, contentType) {
@@ -358,7 +356,7 @@ func (h *Handler) ServeHTTP(responseWriter
http.ResponseWriter, request *http.Re
// Establish a stream and serve the RPC.
setHeaderCanonical(request.Header, headerContentType, contentType)
- // Process context
+ // Derive the request context and deadline from the selected protocol
handler.
ctx, cancel, timeoutErr := protocolHdl.SetTimeout(request) //nolint:
contextcheck
if timeoutErr != nil {
ctx = request.Context()
@@ -366,7 +364,7 @@ func (h *Handler) ServeHTTP(responseWriter
http.ResponseWriter, request *http.Re
if cancel != nil {
defer cancel()
}
- // Create stream
+ // Create the protocol stream that owns request consumption and
response writes.
connCloser, ok := protocolHdl.NewConn(
responseWriter,
request.WithContext(ctx),
@@ -381,7 +379,7 @@ func (h *Handler) ServeHTTP(responseWriter
http.ResponseWriter, request *http.Re
return
}
- // Invoke implementation
+ // Select the group/version implementation and invoke it on the
protocol stream.
svcGroup := request.Header.Get(tripleServiceGroup)
svcVersion := request.Header.Get(tripleServiceVersion)
// TODO(DMwangnima): inspect ok
@@ -441,7 +439,7 @@ func (c *handlerConfig) newSpec(streamType StreamType) Spec
{
}
func (c *handlerConfig) newProtocolHandlers(streamType StreamType)
[]protocolHandler {
- // Initialize protocol
+ // Build protocol candidates for this stream type.
var protocols []protocol
if streamType == StreamTypeUnary {
protocols = append(protocols, &protocolTriple{})
@@ -449,9 +447,8 @@ func (c *handlerConfig) newProtocolHandlers(streamType
StreamType) []protocolHan
if c.HandleGRPC {
protocols = append(protocols, &protocolGRPC{})
}
- // Protocol -> protocolHandler
handlers := make([]protocolHandler, 0, len(protocols))
- // Initialize codec and compressor
+ // Create read-only codec and compressor views shared by protocol
handlers.
compressors := newReadOnlyCompressionPools(
c.CompressionPools,
c.CompressionNames,
