(dubbo-go) branch main updated: feat:add internal service extension (#2577)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

liujun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git


The following commit(s) were added to refs/heads/main by this push:
     new 2e54f26c9 feat:add internal service extension (#2577)
2e54f26c9 is described below

commit 2e54f26c98f6ddd1bac7026197b634cb3bafdf3c
Author: finalt <fi...@163.com>
AuthorDate: Fri Jan 26 16:05:39 2024 +0800

    feat:add internal service extension (#2577)
    
    fixes #2565, #2539
---
 common/constant/key.go                         |  10 +++
 protocol/triple/health/healthServer.go         |  18 ++--
 protocol/triple/reflection/serverreflection.go |  16 ++--
 server/server.go                               | 117 +++++++++++++++++++++----
 4 files changed, 132 insertions(+), 29 deletions(-)

diff --git a/common/constant/key.go b/common/constant/key.go
index e3dfa8760..5dd9dcb0c 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -17,6 +17,8 @@
 
 package constant
 
+import "math"
+
 type DubboCtxKey string
 
 const (
@@ -439,3 +441,11 @@ const (
 	DefaultMetaFileName  = "dubbo.metadata."
 	DefaultEntrySize     = 100
 )
+
+// priority
+const (
+	DefaultPriority = 0
+	HighestPriority = math.MinInt32
+	// LowestPriority for metadata service
+	LowestPriority = math.MaxInt32
+)
diff --git a/protocol/triple/health/healthServer.go b/protocol/triple/health/healthServer.go
index aca420d27..5bd89063e 100644
--- a/protocol/triple/health/healthServer.go
+++ b/protocol/triple/health/healthServer.go
@@ -172,14 +172,18 @@ func (srv *HealthTripleServer) Resume() {
 
 func init() {
 	healthServer = NewServer()
-
 	internal.HealthSetServingStatusServing = SetServingStatusServing
-
-	server.SetProServices(&server.ServiceDefinition{
-		Handler: healthServer,
-		Info:    &triple_health.Health_ServiceInfo,
-		Opts: []server.ServiceOption{server.WithNotRegister(),
-			server.WithInterface(constant.HealthCheckServiceInterface)},
+	server.SetProServices(&server.InternalService{
+		Name: "healthCheck",
+		Init: func(options *server.ServiceOptions) (*server.ServiceDefinition, bool) {
+			return &server.ServiceDefinition{
+				Handler: healthServer,
+				Info:    &triple_health.Health_ServiceInfo,
+				Opts: []server.ServiceOption{server.WithNotRegister(),
+					server.WithInterface(constant.HealthCheckServiceInterface)},
+			}, true
+		},
+		Priority: constant.DefaultPriority,
 	})
 
 	// In order to adapt config.Load
diff --git a/protocol/triple/reflection/serverreflection.go b/protocol/triple/reflection/serverreflection.go
index dff18213b..a3f6ac134 100644
--- a/protocol/triple/reflection/serverreflection.go
+++ b/protocol/triple/reflection/serverreflection.go
@@ -263,11 +263,17 @@ var reflectionServer *ReflectionServer
 func init() {
 	reflectionServer = NewServer()
 	internal.ReflectionRegister = Register
-	server.SetProServices(&server.ServiceDefinition{
-		Handler: reflectionServer,
-		Info:    &rpb.ServerReflection_ServiceInfo,
-		Opts: []server.ServiceOption{server.WithNotRegister(),
-			server.WithInterface(constant.ReflectionServiceInterface)},
+	server.SetProServices(&server.InternalService{
+		Name: "reflection",
+		Init: func(options *server.ServiceOptions) (*server.ServiceDefinition, bool) {
+			return &server.ServiceDefinition{
+				Handler: reflectionServer,
+				Info:    &rpb.ServerReflection_ServiceInfo,
+				Opts: []server.ServiceOption{server.WithNotRegister(),
+					server.WithInterface(constant.ReflectionServiceInterface)},
+			}, true
+		},
+		Priority: constant.DefaultPriority,
 	})
 	// In order to adapt config.Load
 	// Plans for future removal
diff --git a/server/server.go b/server/server.go
index a76d9cd59..0258a940d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -21,6 +21,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"sort"
 	"sync"
 )
 
@@ -38,7 +39,8 @@ import (
 )
 
 // proServices are for internal services
-var proServices = map[string]*ServiceDefinition{}
+var proServices = make([]*InternalService, 0, 16)
+var proLock sync.Mutex
 
 type Server struct {
 	invoker protocol.Invoker
@@ -123,8 +125,17 @@ func newInfoInvoker(url *common.URL, info *ServiceInfo, svc common.RPCService) p
 
 // Register assemble invoker chains like ProviderConfig.Load, init a service per call
 func (s *Server) Register(handler interface{}, info *ServiceInfo, opts ...ServiceOption) error {
+	newSvcOpts, err := s.genSvcOpts(handler, opts...)
+	if err != nil {
+		return err
+	}
+	s.svcOptsMap.Store(newSvcOpts, info)
+	return nil
+}
+
+func (s *Server) genSvcOpts(handler interface{}, opts ...ServiceOption) (*ServiceOptions, error) {
 	if s.cfg == nil {
-		return errors.New("Server has not been initialized, please use NewServer() to create Server")
+		return nil, errors.New("Server has not been initialized, please use NewServer() to create Server")
 	}
 	var svcOpts []ServiceOption
 	appCfg := s.cfg.Application
@@ -153,16 +164,13 @@ func (s *Server) Register(handler interface{}, info *ServiceInfo, opts ...Servic
 			SetRegistries(regsCfg),
 		)
 	}
-
 	// options passed by users have higher priority
 	svcOpts = append(svcOpts, opts...)
 	if err := newSvcOpts.init(s, svcOpts...); err != nil {
-		return err
+		return nil, err
 	}
 	newSvcOpts.Implement(handler)
-	s.svcOptsMap.Store(newSvcOpts, info)
-
-	return nil
+	return newSvcOpts, nil
 }
 
 func (s *Server) exportServices() (err error) {
@@ -187,11 +195,88 @@ func (s *Server) Serve() error {
 	if err := s.exportServices(); err != nil {
 		return err
 	}
+	if err := s.exportInternalServices(); err != nil {
+		return err
+	}
 	metadata.ExportMetadataService()
 	registry_exposed.RegisterServiceInstance(s.cfg.Application.Name, s.cfg.Application.Tag, s.cfg.Application.MetadataType)
 	select {}
 }
 
+// In order to expose internal services
+func (s *Server) exportInternalServices() error {
+	cfg := &ServiceOptions{}
+	cfg.Application = s.cfg.Application
+	cfg.Provider = s.cfg.Provider
+	cfg.Protocols = s.cfg.Protocols
+	cfg.Registries = s.cfg.Registries
+
+	services := make([]*InternalService, 0, len(proServices))
+
+	proLock.Lock()
+	defer proLock.Unlock()
+	for _, service := range proServices {
+		if service.Init == nil {
+			return errors.New("[internal service]internal service init func is empty, please set the init func correctly")
+		}
+		sd, ok := service.Init(cfg)
+		if !ok {
+			logger.Infof("[internal service]%s service will not expose", service.Name)
+			continue
+		}
+		newSvcOpts, err := s.genSvcOpts(sd.Handler, sd.Opts...)
+		if err != nil {
+			return err
+		}
+		service.svcOpts = newSvcOpts
+		service.info = sd.Info
+		services = append(services, service)
+	}
+
+	sort.Slice(services, func(i, j int) bool {
+		return services[i].Priority < services[j].Priority
+	})
+
+	for _, service := range services {
+		if service.BeforeExport != nil {
+			service.BeforeExport(service.svcOpts)
+		}
+		err := service.svcOpts.ExportWithInfo(service.info)
+		if service.AfterExport != nil {
+			service.AfterExport(service.svcOpts, err)
+		}
+		if err != nil {
+			logger.Errorf("[internal service]export %s service failed, err: %s", service.Name, err)
+			return err
+		}
+	}
+
+	return nil
+}
+
+// InternalService for dubbo internal services
+type InternalService struct {
+	// This is required
+	// internal service name
+	Name    string
+	svcOpts *ServiceOptions
+	info    *ServiceInfo
+	// This is required
+	// This options is service configuration
+	// Return serviceDefinition and bool, where bool indicates whether it is exported
+	Init func(options *ServiceOptions) (*ServiceDefinition, bool)
+	// This options is InternalService.svcOpts itself
+	BeforeExport func(options *ServiceOptions)
+	// This options is InternalService.svcOpts itself
+	AfterExport func(options *ServiceOptions, err error)
+	// Priority of service exposure
+	// Lower numbers have the higher priority
+	// The default priority is 0
+	// The metadata service is exposed at the end
+	// If you have no requirements for the order of service exposure, you can use the default priority or not set
+	Priority int
+}
+
 type MethodInfo struct {
 	Name           string
 	Type           string
@@ -210,17 +295,15 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	srv := &Server{
 		cfg: newSrvOpts,
 	}
-
-	for _, service := range proServices {
-		err := srv.Register(service.Handler, service.Info, service.Opts...)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	return srv, nil
 }
 
-func SetProServices(sd *ServiceDefinition) {
-	proServices[sd.Info.InterfaceName] = sd
+func SetProServices(sd *InternalService) {
+	if sd.Name == "" {
+		logger.Warnf("[internal service]internal name is empty, please set internal name")
+		return
+	}
+	proLock.Lock()
+	defer proLock.Unlock()
+	proServices = append(proServices, sd)
 }