You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2026/05/21 06:06:28 UTC
(dubbo-go) 07/14: feat: support mounting existing HTTP handlers on the Triple listener (#3301)
This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
commit 67ec99f307cfd65625aaa9deb9285dc2051461e9
Author: Zecheng Zhu <16...@qq.com>
AuthorDate: Sat May 9 13:54:31 2026 +0800
feat: support mounting existing HTTP handlers on the Triple listener (#3301)
* set http service fallback handler into mux
* feat: add HTTP handler mounting for triple servers
* set mounted http handler int server and refactor server start
* mount first and http only server
* fix: validate transport settings when reusing listeners
* fix: determine whether the HTTP handler has already been mounted
* refactor: clarify attach/host/fallback naming
* fix: keep health status and attached HTTP startup consistent
---
protocol/base/http_host.go | 33 ++
protocol/triple/server.go | 341 +++++++++++++++------
protocol/triple/server_test.go | 82 +++++
protocol/triple/triple.go | 82 +++--
.../triple/triple_protocol/method_route_mux.go | 19 +-
.../triple_protocol/method_route_mux_test.go | 52 ++++
protocol/triple/triple_protocol/server.go | 7 +
protocol/triple/triple_protocol/server_test.go | 16 +
protocol/triple/triple_test.go | 97 ++++++
server/http_attach.go | 115 +++++++
server/http_attach_test.go | 162 ++++++++++
server/server.go | 6 +
server/triple_case_route_integration_test.go | 164 ++++++++++
13 files changed, 1069 insertions(+), 107 deletions(-)
diff --git a/protocol/base/http_host.go b/protocol/base/http_host.go
new file mode 100644
index 000000000..da35b7795
--- /dev/null
+++ b/protocol/base/http_host.go
@@ -0,0 +1,33 @@
+/*
+ * 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 base
+
+import (
+ "net/http"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+)
+
+// HTTPHandlerHost is an optional protocol capability for protocols backed by
+// net/http that can host an existing HTTP handler on the same listener as
+// framework-managed routes.
+type HTTPHandlerHost interface {
+ HostHTTPHandler(url *common.URL, handler http.Handler) error
+}
diff --git a/protocol/triple/server.go b/protocol/triple/server.go
index 90e72bd6d..b3faa1326 100644
--- a/protocol/triple/server.go
+++ b/protocol/triple/server.go
@@ -54,10 +54,21 @@ import (
// Server is TRIPLE adaptation layer representation. It makes use of tri.Server to
// provide functionality.
type Server struct {
- triServer *tri.Server
- cfg *global.TripleConfig
- mu sync.RWMutex
- services map[string]grpc.ServiceInfo
+ triServer *tri.Server
+ cfg *global.TripleConfig
+ mu sync.RWMutex
+ services map[string]grpc.ServiceInfo
+ transportStarted bool
+ attachedHTTPHandler http.Handler
+ transportSettings *transportSettings
+}
+
+type transportSettings struct {
+ location string
+ callProtocol string
+ tripleConfig *global.TripleConfig
+ tlsConfig *tls.Config
+ rawTLSConfig *global.TLSConfig
}
// NewServer creates a new TRIPLE server.
@@ -68,64 +79,173 @@ func NewServer(cfg *global.TripleConfig) *Server {
}
}
-// Start TRIPLE server
-func (s *Server) Start(invoker base.Invoker, info *common.ServiceInfo) {
+// Start prepares the Triple transport for one exported service. It resolves the
+// listener configuration, ensures the underlying tri.Server exists, registers
+// the service handlers, and finally starts the shared transport.
+func (s *Server) Start(invoker base.Invoker, info *common.ServiceInfo) error {
url := invoker.GetURL()
- addr := url.Location
- var tripleConf *global.TripleConfig
+ settings, err := resolveServerTransport(url)
+ if err != nil {
+ return fmt.Errorf("TRIPLE server init failed: %w", err)
+ }
- tripleConfRaw, ok := url.GetAttribute(constant.TripleConfigKey)
- if ok {
- tripleConf = tripleConfRaw.(*global.TripleConfig)
+ if err := s.ensureTriServer(settings); err != nil {
+ return err
+ }
+ if err := s.refreshService(invoker, info); err != nil {
+ return err
}
+ s.startTransport(settings.callProtocol, settings.tlsConfig)
+ return nil
+}
- var callProtocol string
- if tripleConf != nil && tripleConf.Http3 != nil && tripleConf.Http3.Enable {
- callProtocol = constant.CallHTTP2AndHTTP3
- } else {
- // HTTP default type is HTTP/2.
- callProtocol = constant.CallHTTP2
+// resolveServerTransport extracts transport-level Triple configuration from the
+// export URL and converts it into the concrete listener settings used by the
+// adaptation layer.
+func resolveServerTransport(url *common.URL) (*transportSettings, error) {
+ if url == nil {
+ return nil, fmt.Errorf("triple server url must not be nil")
}
- // initialize tri.Server
- s.triServer = tri.NewServer(addr, tripleConf)
+ tripleConf, err := resolveTripleConfig(url)
+ if err != nil {
+ return nil, err
+ }
- serialization := url.GetParam(constant.SerializationKey, constant.ProtobufSerialization)
- switch serialization {
- case constant.ProtobufSerialization:
- case constant.JSONSerialization:
- case constant.Hessian2Serialization:
- case constant.MsgpackSerialization:
- default:
- panic(fmt.Sprintf("Unsupported serialization: %s", serialization))
+ callProtocol := constant.CallHTTP2
+ if tripleConf != nil && tripleConf.Http3 != nil && tripleConf.Http3.Enable {
+ callProtocol = constant.CallHTTP2AndHTTP3
}
+
// todo: support opentracing interceptor
// TODO: move tls config to handleService
- var globalTlsConf *global.TLSConfig
- var tlsConf *tls.Config
- var err error
-
- // handle tls
- tlsConfRaw, ok := url.GetAttribute(constant.TLSConfigKey)
- if ok {
- globalTlsConf, ok = tlsConfRaw.(*global.TLSConfig)
- if !ok {
- logger.Errorf("TRIPLE Server initialized the TLSConfig configuration failed")
- return
- }
+ rawTLSConfig, err := resolveRawTLSConfig(url)
+ if err != nil {
+ return nil, err
}
- if dubbotls.IsServerTLSValid(globalTlsConf) {
- tlsConf, err = dubbotls.GetServerTlSConfig(globalTlsConf)
+
+ var tlsConf *tls.Config
+ if dubbotls.IsServerTLSValid(rawTLSConfig) {
+ tlsConf, err = dubbotls.GetServerTlSConfig(rawTLSConfig)
if err != nil {
- logger.Errorf("TRIPLE Server initialized the TLSConfig configuration failed. err: %v", err)
- return
+ return nil, fmt.Errorf("TRIPLE server initialized the TLSConfig configuration failed: %w", err)
}
logger.Infof("TRIPLE Server initialized the TLSConfig configuration")
}
+ return &transportSettings{
+ location: url.Location,
+ callProtocol: callProtocol,
+ tripleConfig: tripleConf,
+ tlsConfig: tlsConf,
+ rawTLSConfig: rawTLSConfig,
+ }, nil
+}
+
+// ensureTriServer lazily creates the shared tri.Server for one listener
+// address. Service export and mount-first startup both converge here so they
+// share the same transport lifecycle and reflection registration.
+func (s *Server) ensureTriServer(settings *transportSettings) error {
+ if err := s.validateTransportSettings(settings); err != nil {
+ return err
+ }
+
+ if s.triServer != nil {
+ return nil
+ }
+
+ s.mu.Lock()
+ s.transportSettings = settings
+ s.cfg = settings.tripleConfig
+ s.mu.Unlock()
+
+ s.triServer = tri.NewServer(settings.location, settings.tripleConfig)
+ s.wireAttachedHTTPFallback()
+ internal.ReflectionRegister(s)
+ return nil
+}
+
+// wireAttachedHTTPFallback keeps the tri.Server route mux as the main listener
+// handler and connects an attached external HTTP handler through the
+// transport-level fallback slot so Triple routes keep higher priority.
+func (s *Server) wireAttachedHTTPFallback() {
+ s.mu.RLock()
+ handler := s.attachedHTTPHandler
+ s.mu.RUnlock()
+ if handler != nil && s.triServer != nil {
+ s.triServer.SetFallbackHTTPHandler(handler)
+ }
+}
+
+// AttachHTTPHandler attaches a single root HTTP handler to the Triple server.
+// The attached handler follows the same single-root semantics as server.Server:
+// once a handler has been attached for one listener, later attaches must fail
+// fast instead of silently replacing the existing handler.
+func (s *Server) AttachHTTPHandler(handler http.Handler) error {
+ if handler == nil {
+ return fmt.Errorf("attached HTTP handler must not be nil")
+ }
+
+ s.mu.Lock()
+ if s.attachedHTTPHandler != nil {
+ s.mu.Unlock()
+ return fmt.Errorf("an HTTP handler has already been attached")
+ }
+ s.attachedHTTPHandler = handler
+ triServer := s.triServer
+ s.mu.Unlock()
+
+ if triServer != nil {
+ triServer.SetFallbackHTTPHandler(handler)
+ }
+ return nil
+}
+
+func (s *Server) ValidateTransportURL(url *common.URL) error {
+ settings, err := resolveServerTransport(url)
+ if err != nil {
+ return err
+ }
+ return s.validateTransportSettings(settings)
+}
+
+// StartHTTPTransport starts the shared Triple listener without requiring any
+// exported service handlers first. This is used by mount-first and HTTP-only
+// flows to bring up the listener before service export happens later.
+func (s *Server) StartHTTPTransport(url *common.URL) error {
+ settings, err := resolveServerTransport(url)
+ if err != nil {
+ return err
+ }
+
+ // Reuse the same transport bootstrap path as normal service export so mount-
+ // first and export-first flows converge on one listener lifecycle.
+ if err := s.ensureTriServer(settings); err != nil {
+ return err
+ }
+ s.startTransport(settings.callProtocol, settings.tlsConfig)
+ return nil
+}
+
+func (s *Server) startTransport(callProtocol string, tlsConf *tls.Config) {
+ if s.transportStarted {
+ return
+ }
+ s.transportStarted = true
+
+ go func() {
+ if runErr := s.triServer.Run(callProtocol, tlsConf); runErr != nil {
+ logger.Errorf("server serve failed with err: %v", runErr)
+ }
+ }()
+}
+
+func (s *Server) registerServiceHandlers(invoker base.Invoker, info *common.ServiceInfo, hanOpts []tri.HandlerOption) {
+ url := invoker.GetURL()
+
// IDLMode means that this will only be set when
// the new triple is started in non-IDL mode.
// TODO: remove IDLMode when config package is removed
@@ -136,13 +256,7 @@ func (s *Server) Start(invoker base.Invoker, info *common.ServiceInfo) {
service, _ = url.GetAttribute(constant.RpcServiceKey)
}
- hanOpts := getHanOpts(url, tripleConf)
-
- //Set expected codec name from serviceinfo
- hanOpts = append(hanOpts, tri.WithExpectedCodecName(serialization))
-
intfName := url.Interface()
-
//OpenAPI group
var openapiGroup string
if g, ok := url.GetAttribute(constant.OpenAPIMetaKeyOpenAPIGroup); ok {
@@ -163,56 +277,111 @@ func (s *Server) Start(invoker base.Invoker, info *common.ServiceInfo) {
} else {
s.compatHandleService(url, intfName, url.Group(), url.Version(), hanOpts...)
}
- internal.ReflectionRegister(s)
-
- go func() {
- if runErr := s.triServer.Run(callProtocol, tlsConf); runErr != nil {
- logger.Errorf("server serve failed with err: %v", runErr)
- }
- }()
}
-// todo(DMwangnima): extract a common function
-// RefreshService refreshes Triple Service
+// RefreshService refreshes Triple service.
+// The exported wrapper preserves the historical panic-on-invalid-export behavior
+// for direct callers, while internal paths use refreshService to surface errors
+// so export and mount flows can share one explicit failure path.
func (s *Server) RefreshService(invoker base.Invoker, info *common.ServiceInfo) {
- URL := invoker.GetURL()
- serialization := URL.GetParam(constant.SerializationKey, constant.ProtobufSerialization)
+ if err := s.refreshService(invoker, info); err != nil {
+ panic(err)
+ }
+}
+
+func (s *Server) refreshService(invoker base.Invoker, info *common.ServiceInfo) error {
+ url := invoker.GetURL()
+
+ hanOpts, err := resolveHandlerOptions(url)
+ if err != nil {
+ return err
+ }
+
+ s.registerServiceHandlers(invoker, info, hanOpts)
+ return nil
+}
+
+// validateTransportSettings only guards listener-fixed settings. Handler-level
+// Triple options may still vary per exported service, but protocol and TLS
+// configuration must remain stable once one listener has been created.
+func (s *Server) validateTransportSettings(next *transportSettings) error {
+ if next == nil {
+ return nil
+ }
+
+ s.mu.RLock()
+ current := s.transportSettings
+ s.mu.RUnlock()
+ if current == nil {
+ return nil
+ }
+
+ if current.callProtocol != next.callProtocol {
+ return fmt.Errorf("triple transport at %s already uses protocol %s, cannot switch to %s", next.location, current.callProtocol, next.callProtocol)
+ }
+ if !reflect.DeepEqual(current.rawTLSConfig, next.rawTLSConfig) {
+ return fmt.Errorf("triple transport at %s already uses different TLS settings", next.location)
+ }
+ return nil
+}
+
+func resolveTripleConfig(url *common.URL) (*global.TripleConfig, error) {
+ if url == nil {
+ return nil, fmt.Errorf("triple server url must not be nil")
+ }
+
+ tripleConfRaw, ok := url.GetAttribute(constant.TripleConfigKey)
+ if !ok || tripleConfRaw == nil {
+ return nil, nil
+ }
+
+ tripleConf, ok := tripleConfRaw.(*global.TripleConfig)
+ if !ok {
+ return nil, fmt.Errorf("invalid triple config type %T", tripleConfRaw)
+ }
+ return tripleConf.Clone(), nil
+}
+
+func resolveRawTLSConfig(url *common.URL) (*global.TLSConfig, error) {
+ if url == nil {
+ return nil, fmt.Errorf("triple server url must not be nil")
+ }
+
+ tlsConfRaw, ok := url.GetAttribute(constant.TLSConfigKey)
+ if !ok || tlsConfRaw == nil {
+ return nil, nil
+ }
+
+ tlsConf, ok := tlsConfRaw.(*global.TLSConfig)
+ if !ok {
+ return nil, fmt.Errorf("invalid tls config type %T", tlsConfRaw)
+ }
+ return tlsConf.Clone(), nil
+}
+
+func resolveHandlerOptions(url *common.URL) ([]tri.HandlerOption, error) {
+ if url == nil {
+ return nil, fmt.Errorf("triple server url must not be nil")
+ }
+
+ serialization := url.GetParam(constant.SerializationKey, constant.ProtobufSerialization)
switch serialization {
case constant.ProtobufSerialization:
case constant.JSONSerialization:
case constant.Hessian2Serialization:
case constant.MsgpackSerialization:
default:
- panic(fmt.Sprintf("Unsupported serialization: %s", serialization))
- }
- hanOpts := getHanOpts(URL, s.cfg)
- //Set expected codec name from serviceinfo
- hanOpts = append(hanOpts, tri.WithExpectedCodecName(serialization))
- intfName := URL.Interface()
-
- IDLMode := URL.GetParam(constant.IDLMode, "")
- var service common.RPCService
- if IDLMode == constant.NONIDL {
- service, _ = URL.GetAttribute(constant.RpcServiceKey)
+ return nil, fmt.Errorf("unsupported serialization: %s", serialization)
}
- var openapiGroup string
- if g, ok := URL.GetAttribute(constant.OpenAPIMetaKeyOpenAPIGroup); ok {
- if gs, ok := g.(string); ok && gs != "" {
- openapiGroup = gs
- }
+ tripleConf, err := resolveTripleConfig(url)
+ if err != nil {
+ return nil, err
}
- if info != nil {
- s.handleServiceWithInfo(intfName, invoker, info, hanOpts...)
- s.saveServiceInfo(intfName, info, openapiGroup, URL.Group(), URL.Version())
- } else if IDLMode == constant.NONIDL {
- reflectInfo := createServiceInfoWithReflection(service)
- s.handleServiceWithInfo(intfName, invoker, reflectInfo, hanOpts...)
- s.saveServiceInfo(intfName, reflectInfo, openapiGroup, URL.Group(), URL.Version())
- } else {
- s.compatHandleService(URL, intfName, URL.Group(), URL.Version(), hanOpts...)
- }
+ hanOpts := getHanOpts(url, tripleConf)
+ hanOpts = append(hanOpts, tri.WithExpectedCodecName(serialization))
+ return hanOpts, nil
}
func getHanOpts(url *common.URL, tripleConf *global.TripleConfig) (hanOpts []tri.HandlerOption) {
diff --git a/protocol/triple/server_test.go b/protocol/triple/server_test.go
index 508b21118..186a5d28d 100644
--- a/protocol/triple/server_test.go
+++ b/protocol/triple/server_test.go
@@ -577,10 +577,14 @@ func TestHandleServiceWithInfoSaveServiceInfoOnlyOriginalMethods(t *testing.T) {
}
type tripleServerTestInvoker struct {
+ url *common.URL
invokeFn func(context.Context, base.Invocation) result.Result
}
func (m *tripleServerTestInvoker) GetURL() *common.URL {
+ if m.url != nil {
+ return m.url
+ }
return common.NewURLWithOptions()
}
@@ -658,6 +662,84 @@ func (c *tripleServerTestConn) ResponseTrailer() http.Header {
return c.respTrailer
}
+func TestServerValidateTransportSettingsAllowsCompatibleSettings(t *testing.T) {
+ server := NewServer(nil)
+ server.transportSettings = &transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ }
+
+ err := server.validateTransportSettings(&transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ })
+ require.NoError(t, err)
+}
+
+func TestServerValidateTransportSettingsRejectsConflictingProtocol(t *testing.T) {
+ server := NewServer(nil)
+ server.transportSettings = &transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ }
+
+ err := server.validateTransportSettings(&transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2AndHTTP3,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ })
+ require.ErrorContains(t, err, "already uses protocol")
+}
+
+func TestServerValidateTransportSettingsRejectsConflictingTLS(t *testing.T) {
+ server := NewServer(nil)
+ server.transportSettings = &transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ }
+
+ err := server.validateTransportSettings(&transportSettings{
+ location: "127.0.0.1:20000",
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: &global.TLSConfig{TLSCertFile: "server.crt", TLSKeyFile: "server.key"},
+ })
+ require.ErrorContains(t, err, "different TLS settings")
+}
+
+func TestServerAttachHTTPHandlerRejectsDuplicate(t *testing.T) {
+ server := NewServer(nil)
+
+ require.NoError(t, server.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})))
+
+ err := server.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+ require.ErrorContains(t, err, "already been attached")
+}
+
+func TestResolveHandlerOptionsRejectsUnsupportedSerialization(t *testing.T) {
+ url := common.NewURLWithOptions(
+ common.WithParamsValue(constant.SerializationKey, "yaml"),
+ )
+
+ _, err := resolveHandlerOptions(url)
+ require.ErrorContains(t, err, "unsupported serialization: yaml")
+}
+
+func TestServerRefreshServiceRejectsInvalidHandlerTripleConfig(t *testing.T) {
+ server := NewServer(nil)
+ invoker := &tripleServerTestInvoker{
+ url: common.NewURLWithOptions(
+ common.WithAttribute(constant.TripleConfigKey, "invalid"),
+ ),
+ }
+
+ err := server.refreshService(invoker, nil)
+ require.ErrorContains(t, err, "invalid triple config type string")
+}
+
func TestServerRegisterUnaryMethodHandler(t *testing.T) {
server := newServerForMethodHandlerTest()
invoker := &tripleServerTestInvoker{
diff --git a/protocol/triple/triple.go b/protocol/triple/triple.go
index c90ef65d9..caf3f8ad2 100644
--- a/protocol/triple/triple.go
+++ b/protocol/triple/triple.go
@@ -19,6 +19,8 @@ package triple
import (
"context"
+ "fmt"
+ "net/http"
"strings"
"sync"
)
@@ -31,7 +33,6 @@ import (
"dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
- "dubbo.apache.org/dubbo-go/v3/global"
"dubbo.apache.org/dubbo-go/v3/internal"
"dubbo.apache.org/dubbo-go/v3/protocol/base"
)
@@ -96,40 +97,43 @@ func (tp *TripleProtocol) Export(invoker base.Invoker) base.Exporter {
exporter := NewTripleExporter(serviceKey, invoker, tp.ExporterMap())
tp.SetExporterMap(serviceKey, exporter)
logger.Infof("[TRIPLE Protocol] Export service: %s", url.String())
- tp.openServer(invoker, info)
+ if err := tp.openServer(invoker, info); err != nil {
+ logger.Errorf("[TRIPLE Protocol] Export service failed: %s, err: %v", url.String(), err)
+ exporter.UnExport()
+ panic(err)
+ }
internal.HealthSetServingStatusServing(serviceKey)
return exporter
}
-func (tp *TripleProtocol) openServer(invoker base.Invoker, info *common.ServiceInfo) {
+func (tp *TripleProtocol) openServer(invoker base.Invoker, info *common.ServiceInfo) error {
url := invoker.GetURL()
tp.serverLock.Lock()
defer tp.serverLock.Unlock()
- if _, ok := tp.serverMap[url.Location]; ok {
- tp.serverMap[url.Location].RefreshService(invoker, info)
- return
+ if srv, ok := tp.serverMap[url.Location]; ok {
+ if err := srv.ValidateTransportURL(url); err != nil {
+ return err
+ }
+ if err := srv.refreshService(invoker, info); err != nil {
+ return err
+ }
+ return nil
}
if _, ok := tp.ExporterMap().Load(url.ServiceKey()); !ok {
panic("[TRIPLE Protocol]" + url.Key() + "is not existing")
}
- tripleConfRaw, ok := url.GetAttribute(constant.TripleConfigKey)
- if !ok {
- // NOTE: sometimes happened on old triple
- logger.Warnf("Triple config is not found for url: %s", url.Key())
- }
-
- tripleConf, ok := tripleConfRaw.(*global.TripleConfig)
- if !ok || tripleConf == nil {
- // NOTE: sometimes happened on old triple
- logger.Warnf("Triple config obtained from url: %s is not of type *global.TripleConfig or is nil", url.Key())
+ // Do not freeze listener configuration at construction time. The shared
+ // transport must be resolved from the current export URL so mount-first and
+ // export-first flows converge on the same bootstrap path.
+ srv := NewServer(nil)
+ if err := srv.Start(invoker, info); err != nil {
+ return err
}
-
- srv := NewServer(tripleConf)
- srv.Start(invoker, info)
tp.serverMap[url.Location] = srv
+ return nil
}
// Refer a remote triple service
@@ -187,6 +191,46 @@ func (tp *TripleProtocol) drainServers() []*Server {
return servers
}
+// HostHTTPHandler ensures the Triple listener at the given location can host an
+// attached root HTTP handler. If the listener already exists, the handler is
+// attached to the existing server; otherwise the Triple transport is started
+// eagerly so later service export can reuse the same listener.
+func (tp *TripleProtocol) HostHTTPHandler(url *common.URL, handler http.Handler) error {
+ if url == nil {
+ return fmt.Errorf("triple server url must not be nil")
+ }
+ if handler == nil {
+ return fmt.Errorf("attached HTTP handler must not be nil")
+ }
+
+ tp.serverLock.Lock()
+ defer tp.serverLock.Unlock()
+
+ if srv, ok := tp.serverMap[url.Location]; ok {
+ if err := srv.ValidateTransportURL(url); err != nil {
+ return err
+ }
+ // If service export already created the listener, hosting only needs to
+ // attach the root handler to the existing server instance.
+ return srv.AttachHTTPHandler(handler)
+ }
+
+ // Keep construction side-effect free and let StartHTTPTransport resolve
+ // the actual listener settings from the host URL.
+ srv := NewServer(nil)
+ if err := srv.AttachHTTPHandler(handler); err != nil {
+ return err
+ }
+ // Boot transport eagerly so HTTP-only and attach-before-export scenarios
+ // reuse the same Triple listener once services are exported later.
+ if err := srv.StartHTTPTransport(url); err != nil {
+ return err
+ }
+
+ tp.serverMap[url.Location] = srv
+ return nil
+}
+
// isGenericCall checks if the generic parameter indicates a generic call
func isGenericCall(generic string) bool {
if generic == "" {
diff --git a/protocol/triple/triple_protocol/method_route_mux.go b/protocol/triple/triple_protocol/method_route_mux.go
index b4be12669..655677d2a 100644
--- a/protocol/triple/triple_protocol/method_route_mux.go
+++ b/protocol/triple/triple_protocol/method_route_mux.go
@@ -33,8 +33,9 @@ import (
type methodRouteMux struct {
exact *http.ServeMux
- mu sync.RWMutex
- lower map[string]methodRouteEntry
+ mu sync.RWMutex
+ lower map[string]methodRouteEntry
+ fallback http.Handler
}
type methodRouteEntry struct {
@@ -92,9 +93,23 @@ func (m *methodRouteMux) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
+ m.mu.RLock()
+ fallback := m.fallback
+ m.mu.RUnlock()
+ if fallback != nil {
+ fallback.ServeHTTP(w, r)
+ return
+ }
+
http.NotFound(w, r)
}
+func (m *methodRouteMux) SetFallbackHandler(handler http.Handler) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.fallback = handler
+}
+
func normalizeMethodRouteKey(path string) string {
lastSlash := -1
for i := len(path) - 1; i >= 0; i-- {
diff --git a/protocol/triple/triple_protocol/method_route_mux_test.go b/protocol/triple/triple_protocol/method_route_mux_test.go
index 331d7a8c9..c713f1b19 100644
--- a/protocol/triple/triple_protocol/method_route_mux_test.go
+++ b/protocol/triple/triple_protocol/method_route_mux_test.go
@@ -30,6 +30,7 @@ import (
const (
methodRouteMuxGetUserPath = "/Service/GetUser"
methodRouteMuxGetUserLowerPath = "/Service/getUser"
+ methodRouteMuxHealthPath = "/healthz"
methodRouteMuxNotFoundBody = "404 page not found\n"
)
@@ -206,3 +207,54 @@ func TestMethodRouteMuxFallbackCollisionFirstRegistrationWins(t *testing.T) {
entry.handler.ServeHTTP(resp, req)
assert.Equal(t, "first", resp.Body.String())
}
+
+func TestMethodRouteMuxUsesFallbackHandlerOnMiss(t *testing.T) {
+ mux := newMethodRouteMux()
+ mux.SetFallbackHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("http-fallback"))
+ }))
+
+ req := httptest.NewRequest(http.MethodGet, methodRouteMuxHealthPath, nil)
+ resp := httptest.NewRecorder()
+
+ mux.ServeHTTP(resp, req)
+
+ assert.Equal(t, http.StatusOK, resp.Code)
+ assert.Equal(t, "http-fallback", resp.Body.String())
+}
+
+func TestMethodRouteMuxTripleMatchWinsOverFallback(t *testing.T) {
+ mux := newMethodRouteMux()
+ mux.Handle(methodRouteMuxGetUserPath, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("triple"))
+ }))
+ mux.SetFallbackHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("http-fallback"))
+ }))
+
+ req := httptest.NewRequest(http.MethodPost, methodRouteMuxGetUserPath, nil)
+ resp := httptest.NewRecorder()
+
+ mux.ServeHTTP(resp, req)
+
+ assert.Equal(t, http.StatusOK, resp.Code)
+ assert.Equal(t, "triple", resp.Body.String())
+}
+
+func TestMethodRouteMuxCaseFallbackWinsOverHTTPFallback(t *testing.T) {
+ mux := newMethodRouteMux()
+ mux.Handle(methodRouteMuxGetUserPath, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("triple-case-fallback"))
+ }))
+ mux.SetFallbackHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("http-fallback"))
+ }))
+
+ req := httptest.NewRequest(http.MethodPost, methodRouteMuxGetUserLowerPath, nil)
+ resp := httptest.NewRecorder()
+
+ mux.ServeHTTP(resp, req)
+
+ assert.Equal(t, http.StatusOK, resp.Code)
+ assert.Equal(t, "triple-case-fallback", resp.Body.String())
+}
diff --git a/protocol/triple/triple_protocol/server.go b/protocol/triple/triple_protocol/server.go
index 9859ed38b..95121c06a 100644
--- a/protocol/triple/triple_protocol/server.go
+++ b/protocol/triple/triple_protocol/server.go
@@ -175,6 +175,13 @@ func (s *Server) RegisterCompatStreamHandler(
return nil
}
+func (s *Server) SetFallbackHTTPHandler(h http.Handler) {
+ if s.mux == nil {
+ return
+ }
+ s.mux.SetFallbackHandler(h)
+}
+
func (s *Server) Run(callProtocol string, tlsConf *tls.Config) error {
// Support for starting HTTP/2 and HTTP/3 servers simultaneously.
switch callProtocol {
diff --git a/protocol/triple/triple_protocol/server_test.go b/protocol/triple/triple_protocol/server_test.go
index 33e024236..d2693c98d 100644
--- a/protocol/triple/triple_protocol/server_test.go
+++ b/protocol/triple/triple_protocol/server_test.go
@@ -19,6 +19,7 @@ package triple_protocol
import (
"net/http"
+ "net/http/httptest"
"net/url"
"testing"
)
@@ -99,3 +100,18 @@ func TestServer_RegisterMuxHandle(t *testing.T) {
assert.Equal(t, test.path, pattern)
}
}
+
+func TestServerSetFallbackHTTPHandler(t *testing.T) {
+ srv := NewServer("127.0.0.1:0", nil)
+ srv.SetFallbackHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _, _ = w.Write([]byte("ok"))
+ }))
+
+ req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
+ resp := httptest.NewRecorder()
+
+ srv.mux.ServeHTTP(resp, req)
+
+ assert.Equal(t, http.StatusOK, resp.Code)
+ assert.Equal(t, "ok", resp.Body.String())
+}
diff --git a/protocol/triple/triple_test.go b/protocol/triple/triple_test.go
index 6af85ff40..a499d6712 100644
--- a/protocol/triple/triple_test.go
+++ b/protocol/triple/triple_test.go
@@ -19,6 +19,7 @@ package triple
import (
"context"
+ "net/http"
"sync"
"sync/atomic"
"testing"
@@ -27,10 +28,15 @@ import (
import (
"github.com/stretchr/testify/assert"
+ "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/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/internal"
"dubbo.apache.org/dubbo-go/v3/protocol/base"
tri "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
)
@@ -128,6 +134,97 @@ func TestTripleGracefulShutdownCallbackRegistration(t *testing.T) {
})
}
+func TestTripleProtocolOpenServerRejectsConflictingTransportSettings(t *testing.T) {
+ tp := NewTripleProtocol()
+ location := "127.0.0.1:20000"
+ tp.serverMap[location] = &Server{
+ transportSettings: &transportSettings{
+ location: location,
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ },
+ }
+
+ invoker := &tripleServerTestInvoker{
+ url: common.NewURLWithOptions(
+ common.WithProtocol(TRIPLE),
+ common.WithIp("127.0.0.1"),
+ common.WithPort("20000"),
+ common.WithAttribute(constant.TripleConfigKey, &global.TripleConfig{
+ Http3: &global.Http3Config{Enable: true},
+ }),
+ common.WithAttribute(constant.TLSConfigKey, global.DefaultTLSConfig()),
+ ),
+ }
+
+ err := tp.openServer(invoker, nil)
+ require.ErrorContains(t, err, "already uses protocol")
+}
+
+func TestTripleProtocolExportPublishesServingStatusByServiceKey(t *testing.T) {
+ tp := NewTripleProtocol()
+ url := common.NewURLWithOptions(
+ common.WithProtocol(TRIPLE),
+ common.WithIp("127.0.0.1"),
+ common.WithPort("20000"),
+ common.WithPath("com.example.GreetService"),
+ common.WithParamsValue(constant.InterfaceKey, "com.example.GreetService"),
+ common.WithParamsValue(constant.GroupKey, "test-group"),
+ common.WithParamsValue(constant.VersionKey, "1.0.0"),
+ )
+ tp.serverMap[url.Location] = &Server{
+ triServer: tri.NewServer(url.Location, nil),
+ transportSettings: &transportSettings{
+ location: url.Location,
+ callProtocol: constant.CallHTTP2,
+ },
+ }
+
+ originalSetServing := internal.HealthSetServingStatusServing
+ t.Cleanup(func() {
+ internal.HealthSetServingStatusServing = originalSetServing
+ })
+
+ published := make(chan string, 1)
+ internal.HealthSetServingStatusServing = func(service string) {
+ published <- service
+ }
+
+ exporter := tp.Export(&tripleServerTestInvoker{url: url})
+ require.NotNil(t, exporter)
+
+ select {
+ case got := <-published:
+ assert.Equal(t, url.ServiceKey(), got)
+ assert.NotEqual(t, url.Service(), got)
+ case <-time.After(time.Second):
+ t.Fatal("export did not publish serving status")
+ }
+}
+
+func TestTripleProtocolHostHTTPHandlerRejectsDuplicateOnExistingListener(t *testing.T) {
+ tp := NewTripleProtocol()
+ location := "127.0.0.1:20000"
+ tp.serverMap[location] = &Server{
+ attachedHTTPHandler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}),
+ transportSettings: &transportSettings{
+ location: location,
+ callProtocol: constant.CallHTTP2,
+ rawTLSConfig: global.DefaultTLSConfig(),
+ },
+ }
+
+ url := common.NewURLWithOptions(
+ common.WithProtocol(TRIPLE),
+ common.WithIp("127.0.0.1"),
+ common.WithPort("20000"),
+ common.WithAttribute(constant.TLSConfigKey, global.DefaultTLSConfig()),
+ )
+
+ err := tp.HostHTTPHandler(url, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+ require.ErrorContains(t, err, "already been attached")
+}
+
func TestTripleProtocol_Destroy_EmptyServerMap(t *testing.T) {
tp := NewTripleProtocol()
diff --git a/server/http_attach.go b/server/http_attach.go
new file mode 100644
index 000000000..0603ac628
--- /dev/null
+++ b/server/http_attach.go
@@ -0,0 +1,115 @@
+/*
+ * 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 (
+ "fmt"
+ "net/http"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+)
+
+// AttachHTTPHandler records an existing HTTP root handler on the server before
+// Serve starts. The attached handler is later hosted on the selected protocol
+// listener and acts as the transport-level fallback after Triple route lookup,
+// so callers should aggregate any HTTP sub-services behind their own mux/router
+// before attaching.
+func (s *Server) AttachHTTPHandler(handler http.Handler) error {
+ if handler == nil {
+ return fmt.Errorf("attached HTTP handler must not be nil")
+ }
+
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ if s.serve {
+ return fmt.Errorf("attached HTTP handler must be configured before Serve")
+ }
+ // Server-level HTTP integration is a single root handler. Callers that need
+ // multiple HTTP services should compose them behind their own mux/router.
+ if s.attachedHTTPHandler != nil {
+ return fmt.Errorf("an HTTP handler has already been attached")
+ }
+
+ s.attachedHTTPHandler = handler
+ return nil
+}
+
+// hostAttachedHTTPHandler asks an HTTP-capable protocol to host the attached
+// root handler on the same listener as framework-managed routes. Today that
+// means Triple running on an explicit, user-selected port.
+func (s *Server) hostAttachedHTTPHandler() error {
+ s.mu.RLock()
+ handler := s.attachedHTTPHandler
+ s.mu.RUnlock()
+ if handler == nil {
+ return nil
+ }
+
+ protocolConfigs := loadProtocol(s.cfg.Provider.ProtocolIDs, s.cfg.Protocols)
+ if len(protocolConfigs) == 0 {
+ // WithServerProtocol populates Server.Protocols, but HTTP mounting can run
+ // before any service export wires Provider.ProtocolIDs. Fall back to the
+ // declared server protocols so mount-only startup still works.
+ protocolConfigs = make([]*global.ProtocolConfig, 0, len(s.cfg.Protocols))
+ for _, protocolConf := range s.cfg.Protocols {
+ protocolConfigs = append(protocolConfigs, protocolConf)
+ }
+ }
+ mounted := false
+
+ for _, protocolConf := range protocolConfigs {
+ if protocolConf == nil || protocolConf.Name != constant.TriProtocol {
+ continue
+ }
+ if protocolConf.Port == "" {
+ // Hosting needs a stable listener up front; unlike normal service
+ // export we should not silently start on an implicit/random port here.
+ return fmt.Errorf("hosting an attached HTTP handler requires an explicit triple port")
+ }
+
+ proto := extension.GetProtocol(protocolConf.Name)
+ host, ok := proto.(base.HTTPHandlerHost)
+ if !ok {
+ return fmt.Errorf("protocol %s does not support hosting attached HTTP handlers", protocolConf.Name)
+ }
+
+ u := common.NewURLWithOptions(
+ common.WithProtocol(protocolConf.Name),
+ common.WithIp(protocolConf.Ip),
+ common.WithPort(protocolConf.Port),
+ common.WithAttribute(constant.TripleConfigKey, protocolConf.TripleConfig),
+ common.WithAttribute(constant.TLSConfigKey, s.cfg.TLS),
+ )
+ if err := host.HostHTTPHandler(u, handler); err != nil {
+ return err
+ }
+ mounted = true
+ }
+
+ if !mounted {
+ return fmt.Errorf("attached HTTP handler requires at least one triple protocol")
+ }
+ return nil
+}
diff --git a/server/http_attach_test.go b/server/http_attach_test.go
new file mode 100644
index 000000000..f51995c2a
--- /dev/null
+++ b/server/http_attach_test.go
@@ -0,0 +1,162 @@
+/*
+ * 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 (
+ "net/http"
+ "strconv"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/protocol"
+ _ "dubbo.apache.org/dubbo-go/v3/protocol/triple"
+)
+
+type httpMountTestHandler struct{}
+
+func (h *httpMountTestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {}
+
+func TestServerAttachHTTPHandler(t *testing.T) {
+ srv, err := NewServer()
+ require.NoError(t, err)
+
+ handler := &httpMountTestHandler{}
+ err = srv.AttachHTTPHandler(handler)
+ require.NoError(t, err)
+
+ assert.Same(t, handler, srv.attachedHTTPHandler)
+}
+
+func TestServerAttachHTTPHandlerRejectsNil(t *testing.T) {
+ srv, err := NewServer()
+ require.NoError(t, err)
+
+ err = srv.AttachHTTPHandler(nil)
+ require.ErrorContains(t, err, "must not be nil")
+}
+
+func TestServerAttachHTTPHandlerRejectsDuplicate(t *testing.T) {
+ srv, err := NewServer()
+ require.NoError(t, err)
+
+ first := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
+ second := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
+
+ require.NoError(t, srv.AttachHTTPHandler(first))
+
+ err = srv.AttachHTTPHandler(second)
+ require.ErrorContains(t, err, "already been attached")
+}
+
+func TestServerAttachHTTPHandlerRejectsAfterServe(t *testing.T) {
+ srv, err := NewServer()
+ require.NoError(t, err)
+
+ srv.serve = true
+ err = srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+ require.ErrorContains(t, err, "before Serve")
+}
+
+func TestHostAttachedHTTPHandlerRequiresTripleProtocol(t *testing.T) {
+ srv, err := NewServer(
+ WithServerProtocol(
+ protocol.WithDubbo(),
+ ),
+ )
+ require.NoError(t, err)
+
+ require.NoError(t, srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})))
+
+ err = srv.hostAttachedHTTPHandler()
+ require.ErrorContains(t, err, "requires at least one triple protocol")
+}
+
+func TestHostAttachedHTTPHandlerRequiresExplicitPort(t *testing.T) {
+ srv, err := NewServer(
+ WithServerProtocol(
+ protocol.WithTriple(),
+ ),
+ )
+ require.NoError(t, err)
+
+ srv.cfg.Protocols[constant.TriProtocol].Port = ""
+ require.NoError(t, srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})))
+
+ err = srv.hostAttachedHTTPHandler()
+ require.ErrorContains(t, err, "requires an explicit triple port")
+}
+
+func TestServeDoesNotHostAttachedHTTPHandlerBeforeLaterStartupFailures(t *testing.T) {
+ port := testFreePort(t)
+ srv, err := NewServer(
+ WithServerProtocol(
+ protocol.WithTriple(),
+ protocol.WithIp("127.0.0.1"),
+ protocol.WithPort(port),
+ ),
+ )
+ require.NoError(t, err)
+
+ require.NoError(t, srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/healthz":
+ _, _ = w.Write([]byte("healthy"))
+ default:
+ http.NotFound(w, r)
+ }
+ })))
+
+ internalProLock.Lock()
+ originalInternalServices := internalProServices
+ internalProServices = []*InternalService{
+ {Name: "broken-internal-service"},
+ }
+ internalProLock.Unlock()
+ t.Cleanup(func() {
+ internalProLock.Lock()
+ internalProServices = originalInternalServices
+ internalProLock.Unlock()
+ })
+
+ t.Cleanup(func() {
+ extension.GetProtocol(constant.TriProtocol).Destroy()
+ })
+
+ err = srv.Serve()
+ require.ErrorContains(t, err, "internal service init func is empty")
+
+ client := &http.Client{Timeout: 100 * time.Millisecond}
+ baseURL := "http://127.0.0.1:" + strconv.Itoa(port)
+ deadline := time.Now().Add(time.Second)
+ for time.Now().Before(deadline) {
+ status, body, reqErr := simpleHTTPRequest(client, http.MethodGet, baseURL+"/healthz")
+ if reqErr == nil {
+ t.Fatalf("attached HTTP handler started after Serve failure: status=%d body=%q", status, body)
+ }
+ time.Sleep(50 * time.Millisecond)
+ }
+}
diff --git a/server/server.go b/server/server.go
index 29c605acb..c3316223a 100644
--- a/server/server.go
+++ b/server/server.go
@@ -21,6 +21,7 @@ package server
import (
"context"
stderrors "errors"
+ "net/http"
"reflect"
"sort"
"strconv"
@@ -59,6 +60,8 @@ type Server struct {
interfaceNameServices map[string]*ServiceOptions
// indicate whether the server is already started
serve bool
+
+ attachedHTTPHandler http.Handler
}
// ServiceInfo Deprecated： common.ServiceInfo type alias, just for compatible with old generate pb.go file
@@ -390,6 +393,9 @@ func (s *Server) ServeContext(ctx context.Context) error {
return s.rollbackServeStartWithCause(err, serviceInstanceRegistered)
}
serviceInstanceRegistered = true
+ if err := s.hostAttachedHTTPHandler(); err != nil {
+ return s.rollbackServeStartWithCause(err, serviceInstanceRegistered)
+ }
// k8s probe ready
probe.SetStartupComplete(true)
diff --git a/server/triple_case_route_integration_test.go b/server/triple_case_route_integration_test.go
index af0332949..5cc891c35 100644
--- a/server/triple_case_route_integration_test.go
+++ b/server/triple_case_route_integration_test.go
@@ -236,6 +236,27 @@ func tripleRouteRequest(client *http.Client, url string) (int, string, error) {
return resp.StatusCode, string(body), nil
}
+func simpleHTTPRequest(client *http.Client, method string, url string) (int, string, error) {
+ req, err := http.NewRequest(method, url, nil)
+ if err != nil {
+ return 0, "", err
+ }
+
+ resp, err := client.Do(req)
+ if err != nil {
+ return 0, "", err
+ }
+ defer func() {
+ _ = resp.Body.Close()
+ }()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return 0, "", err
+ }
+ return resp.StatusCode, string(body), nil
+}
+
func waitTripleRouteReady(t *testing.T, client *http.Client, url string) {
t.Helper()
@@ -254,3 +275,146 @@ func waitTripleRouteReady(t *testing.T, client *http.Client, url string) {
t.Fatalf("triple route not ready: status=%d body=%q err=%v", lastStatus, lastBody, lastErr)
}
+
+func waitSimpleHTTPReady(t *testing.T, client *http.Client, method string, url string, wantBody string) {
+ t.Helper()
+
+ deadline := time.Now().Add(5 * time.Second)
+ var lastStatus int
+ var lastBody string
+ var lastErr error
+
+ for time.Now().Before(deadline) {
+ lastStatus, lastBody, lastErr = simpleHTTPRequest(client, method, url)
+ if lastErr == nil && lastStatus == http.StatusOK && lastBody == wantBody {
+ return
+ }
+ time.Sleep(50 * time.Millisecond)
+ }
+
+ t.Fatalf("http handler not ready: status=%d body=%q err=%v", lastStatus, lastBody, lastErr)
+}
+
+// TestAttachedHTTPHandlerStartsTransportWithoutTripleServices covers the
+// attach-first flow where the shared listener is started for plain HTTP traffic
+// before any Triple service has been exported.
+func TestAttachedHTTPHandlerStartsTransportWithoutTripleServices(t *testing.T) {
+ port := testFreePort(t)
+
+ srv, err := NewServer(
+ WithServerProtocol(
+ protocol.WithTriple(),
+ protocol.WithIp("127.0.0.1"),
+ protocol.WithPort(port),
+ ),
+ )
+ require.NoError(t, err)
+
+ // Attach-only startup should bring up the shared listener even before any
+ // Triple service has been exported.
+ err = srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/healthz":
+ _, _ = w.Write([]byte("healthy"))
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ require.NoError(t, err)
+
+ err = srv.hostAttachedHTTPHandler()
+ require.NoError(t, err)
+
+ t.Cleanup(func() {
+ extension.GetProtocol(constant.TriProtocol).Destroy()
+ })
+
+ client := &http.Client{Timeout: 2 * time.Second}
+ baseURL := "http://127.0.0.1:" + strconv.Itoa(port)
+
+ waitSimpleHTTPReady(t, client, http.MethodGet, baseURL+"/healthz", "healthy")
+
+ status, body, reqErr := simpleHTTPRequest(client, http.MethodGet, baseURL+"/unknown")
+ require.NoError(t, reqErr)
+ assert.Equal(t, http.StatusNotFound, status)
+ assert.Equal(t, tripleCaseRouteNotFoundBody, body)
+}
+
+// TestAttachedHTTPHandlerCoexistsWithTripleRoutes verifies that the attached HTTP
+// handler remains a transport-level fallback and does not preempt registered
+// Triple procedures on the same listener.
+func TestAttachedHTTPHandlerCoexistsWithTripleRoutes(t *testing.T) {
+ port := testFreePort(t)
+
+ srv, err := NewServer(
+ WithServerProtocol(
+ protocol.WithTriple(),
+ protocol.WithIp("127.0.0.1"),
+ protocol.WithPort(port),
+ ),
+ )
+ require.NoError(t, err)
+
+ // The attached handler is the transport-level fallback; Triple routes should
+ // still win whenever the request matches a registered procedure.
+ err = srv.AttachHTTPHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/healthz":
+ _, _ = w.Write([]byte("healthy"))
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ require.NoError(t, err)
+
+ service := &TripleCaseRouteService{}
+ info := &common.ServiceInfo{
+ InterfaceName: "com.example.GreetService",
+ ServiceType: service,
+ Methods: []common.MethodInfo{
+ {
+ Name: "SayHello",
+ Type: constant.CallUnary,
+ ReqInitFunc: func() any {
+ return &emptypb.Empty{}
+ },
+ MethodFunc: func(ctx context.Context, args []any, handler any) (any, error) {
+ req := args[0].(*emptypb.Empty)
+ res, callErr := handler.(*TripleCaseRouteService).SayHello(ctx, req)
+ if callErr != nil {
+ return nil, callErr
+ }
+ return tri.NewResponse(res), nil
+ },
+ },
+ },
+ }
+
+ err = srv.Register(service, info, WithInterface(info.InterfaceName), WithNotRegister())
+ require.NoError(t, err)
+
+ err = srv.hostAttachedHTTPHandler()
+ require.NoError(t, err)
+
+ err = srv.exportServices(context.Background())
+ require.NoError(t, err)
+
+ t.Cleanup(func() {
+ extension.GetProtocol(constant.TriProtocol).Destroy()
+ })
+
+ client := &http.Client{Timeout: 2 * time.Second}
+ baseURL := "http://127.0.0.1:" + strconv.Itoa(port)
+
+ waitTripleRouteReady(t, client, baseURL+"/com.example.GreetService/SayHello")
+
+ status, body, reqErr := simpleHTTPRequest(client, http.MethodGet, baseURL+"/healthz")
+ require.NoError(t, reqErr)
+ assert.Equal(t, http.StatusOK, status)
+ assert.Equal(t, "healthy", body)
+
+ status, body, reqErr = tripleRouteRequest(client, baseURL+"/com.example.GreetService/sayHello")
+ require.NoError(t, reqErr)
+ assert.Equal(t, http.StatusOK, status)
+ assert.Equal(t, tripleCaseRouteHelloBody, body)
+}