You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2023/09/13 12:19:47 UTC
[dubbo-go] branch feature-triple updated: feat: add server side (#2421)
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch feature-triple
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/feature-triple by this push:
new 5d1552b10 feat: add server side (#2421)
5d1552b10 is described below
commit 5d1552b1065e5c65d836dc092f7979df95977ac8
Author: Xinhao Yuan <ll...@foxmail.com>
AuthorDate: Wed Sep 13 20:19:41 2023 +0800
feat: add server side (#2421)
---
common/url.go | 17 +-
global/service_config.go | 207 +++++++++++++++++++++
protocol/triple/triple.go | 5 +-
server/action.go | 465 ++++++++++++++++++++++++++++++++++++++++++++++
server/compat.go | 79 ++++++++
server/options.go | 173 ++++++++++++++++-
server/server.go | 52 ++++--
7 files changed, 980 insertions(+), 18 deletions(-)
diff --git a/common/url.go b/common/url.go
index 2824ec68d..3bc481bc9 100644
--- a/common/url.go
+++ b/common/url.go
@@ -44,6 +44,7 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/server"
)
// dubbo role type constant
@@ -110,10 +111,11 @@ type URL struct {
paramsLock sync.RWMutex
params url.Values
- Path string // like /com.ikurento.dubbo.UserProvider
- Username string
- Password string
- Methods []string
+ Path string // like /com.ikurento.dubbo.UserProvider
+ Username string
+ Password string
+ Methods []string
+ MethodInfo []server.MethodInfo
// special for registry
SubURL *URL
}
@@ -218,6 +220,13 @@ func WithToken(token string) Option {
}
}
+// WithMethodInfos sets methodInfos for URL
+func WithMethodInfos(methodInfos []server.MethodInfo) Option {
+ return func(url *URL) {
+ url.MethodInfo = methodInfos
+ }
+}
+
// NewURLWithOptions will create a new URL with options
func NewURLWithOptions(opts ...Option) *URL {
newURL := &URL{}
diff --git a/global/service_config.go b/global/service_config.go
new file mode 100644
index 000000000..a098e975b
--- /dev/null
+++ b/global/service_config.go
@@ -0,0 +1,207 @@
+package global
+
+// ServiceConfig is the configuration of the service provider
+type ServiceConfig struct {
+ Filter string `yaml:"filter" json:"filter,omitempty" property:"filter"`
+ ProtocolIDs []string `yaml:"protocol-ids" json:"protocol-ids,omitempty" property:"protocol-ids"` // multi protocolIDs support, split by ','
+ Interface string `yaml:"interface" json:"interface,omitempty" property:"interface"`
+ RegistryIDs []string `yaml:"registry-ids" json:"registry-ids,omitempty" property:"registry-ids"`
+ Cluster string `default:"failover" yaml:"cluster" json:"cluster,omitempty" property:"cluster"`
+ Loadbalance string `default:"random" yaml:"loadbalance" json:"loadbalance,omitempty" property:"loadbalance"`
+ Group string `yaml:"group" json:"group,omitempty" property:"group"`
+ Version string `yaml:"version" json:"version,omitempty" property:"version" `
+ Methods []*MethodConfig `yaml:"methods" json:"methods,omitempty" property:"methods"`
+ Warmup string `yaml:"warmup" json:"warmup,omitempty" property:"warmup"`
+ Retries string `yaml:"retries" json:"retries,omitempty" property:"retries"`
+ Serialization string `yaml:"serialization" json:"serialization" property:"serialization"`
+ Params map[string]string `yaml:"params" json:"params,omitempty" property:"params"`
+ Token string `yaml:"token" json:"token,omitempty" property:"token"`
+ AccessLog string `yaml:"accesslog" json:"accesslog,omitempty" property:"accesslog"`
+ TpsLimiter string `yaml:"tps.limiter" json:"tps.limiter,omitempty" property:"tps.limiter"`
+ TpsLimitInterval string `yaml:"tps.limit.interval" json:"tps.limit.interval,omitempty" property:"tps.limit.interval"`
+ TpsLimitRate string `yaml:"tps.limit.rate" json:"tps.limit.rate,omitempty" property:"tps.limit.rate"`
+ TpsLimitStrategy string `yaml:"tps.limit.strategy" json:"tps.limit.strategy,omitempty" property:"tps.limit.strategy"`
+ TpsLimitRejectedHandler string `yaml:"tps.limit.rejected.handler" json:"tps.limit.rejected.handler,omitempty" property:"tps.limit.rejected.handler"`
+ ExecuteLimit string `yaml:"execute.limit" json:"execute.limit,omitempty" property:"execute.limit"`
+ ExecuteLimitRejectedHandler string `yaml:"execute.limit.rejected.handler" json:"execute.limit.rejected.handler,omitempty" property:"execute.limit.rejected.handler"`
+ Auth string `yaml:"auth" json:"auth,omitempty" property:"auth"`
+ NotRegister bool `yaml:"not_register" json:"not_register,omitempty" property:"not_register"`
+ ParamSign string `yaml:"param.sign" json:"param.sign,omitempty" property:"param.sign"`
+ Tag string `yaml:"tag" json:"tag,omitempty" property:"tag"`
+ TracingKey string `yaml:"tracing-key" json:"tracing-key,omitempty" propertiy:"tracing-key"`
+}
+
+type ServiceOption func(*ServiceConfig)
+
+func DefaultServiceConfig() *ServiceConfig {
+ return &ServiceConfig{
+ Methods: make([]*MethodConfig, 0, 8),
+ Params: make(map[string]string, 8),
+ }
+}
+
+func WithService_Filter(filter string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Filter = filter
+ }
+}
+
+func WithService_ProtocolIDs(protocolIDs []string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ if len(protocolIDs) <= 0 {
+ cfg.ProtocolIDs = protocolIDs
+ }
+ }
+}
+
+func WithService_Interface(name string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Interface = name
+ }
+}
+
+func WithService_RegistryIDs(registryIDs []string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ if len(registryIDs) <= 0 {
+ cfg.RegistryIDs = registryIDs
+ }
+ }
+}
+
+func WithService_Cluster(cluster string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Cluster = cluster
+ }
+}
+
+func WithService_LoadBalance(loadBalance string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Loadbalance = loadBalance
+ }
+}
+
+func WithService_Group(group string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Group = group
+ }
+}
+
+func WithService_Version(version string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Version = version
+ }
+}
+
+func WithService_Methods(methods []*MethodConfig) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Methods = methods
+ }
+}
+
+func WithService_WarmUp(warmUp string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Warmup = warmUp
+ }
+}
+
+func WithService_Retries(retries string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Retries = retries
+ }
+}
+
+func WithService_Serialization(serialization string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Serialization = serialization
+ }
+}
+
+func WithService_Params(params map[string]string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Params = params
+ }
+}
+
+func WithService_Token(token string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Token = token
+ }
+}
+
+func WithService_AccessLog(accessLog string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.AccessLog = accessLog
+ }
+}
+
+func WithService_TpsLimiter(tpsLimiter string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TpsLimiter = tpsLimiter
+ }
+}
+
+func WithService_TpsLimitInterval(tpsLimitInterval string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TpsLimitInterval = tpsLimitInterval
+ }
+}
+
+func WithService_TpsLimitRate(tpsLimitRate string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TpsLimitRate = tpsLimitRate
+ }
+}
+
+func WithService_TpsLimitStrategy(tpsLimitStrategy string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TpsLimitStrategy = tpsLimitStrategy
+ }
+}
+
+func WithService_TpsLimitRejectedHandler(tpsLimitRejectedHandler string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TpsLimitRejectedHandler = tpsLimitRejectedHandler
+ }
+}
+
+func WithService_ExecuteLimit(executeLimit string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.ExecuteLimit = executeLimit
+ }
+}
+
+func WithService_ExecuteLimitRejectedHandler(executeLimitRejectedHandler string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.ExecuteLimitRejectedHandler = executeLimitRejectedHandler
+ }
+}
+
+func WithService_Auth(auth string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Auth = auth
+ }
+}
+
+func WithService_NotRegister(notRegister bool) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.NotRegister = notRegister
+ }
+}
+
+func WithService_ParamSign(paramSign string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.ParamSign = paramSign
+ }
+}
+
+func WithService_Tag(tag string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.Tag = tag
+ }
+}
+
+func WithService_TracingKey(tracingKey string) ServiceOption {
+ return func(cfg *ServiceConfig) {
+ cfg.TracingKey = tracingKey
+ }
+}
diff --git a/protocol/triple/triple.go b/protocol/triple/triple.go
index 0569f9d33..4ec0742bf 100644
--- a/protocol/triple/triple.go
+++ b/protocol/triple/triple.go
@@ -56,7 +56,10 @@ func (tp *TripleProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
url := invoker.GetURL()
serviceKey := url.ServiceKey()
// todo: retrieve this info from url
- info := &server.ServiceInfo{}
+ info := &server.ServiceInfo{
+ InterfaceName: url.Path,
+ Methods: url.MethodInfo,
+ }
exporter := NewTripleExporter(serviceKey, invoker, tp.ExporterMap())
tp.SetExporterMap(serviceKey, exporter)
logger.Infof("[TRIPLE Protocol] Export service: %s", url.String())
diff --git a/server/action.go b/server/action.go
new file mode 100644
index 000000000..84f9e07ec
--- /dev/null
+++ b/server/action.go
@@ -0,0 +1,465 @@
+/*
+ * 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 (
+ "container/list"
+
+ "fmt"
+ "net/url"
+ "os"
+ "strconv"
+ "strings"
+ "time"
+)
+
+import (
+ "github.com/creasty/defaults"
+
+ "github.com/dubbogo/gost/log/logger"
+ gxnet "github.com/dubbogo/gost/net"
+
+ perrors "github.com/pkg/errors"
+
+ "go.uber.org/atomic"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ commonCfg "dubbo.apache.org/dubbo-go/v3/common/config"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/config"
+ "dubbo.apache.org/dubbo-go/v3/protocol/protocolwrapper"
+)
+
+// Prefix returns dubbo.service.${InterfaceName}.
+func (s *ServerOptions) Prefix() string {
+ return strings.Join([]string{constant.ServiceConfigPrefix, s.Id}, ".")
+}
+
+func (s *ServerOptions) Init(opts ...ServerOption) error {
+ for _, opt := range opts {
+ opt(s)
+ }
+ if err := defaults.Set(s); err != nil {
+ return err
+ }
+ for _, opt := range opts {
+ opt(s)
+ }
+
+ srv := s.Server
+
+ s.exported = atomic.NewBool(false)
+
+ application := s.Application
+ if application != nil {
+ s.applicationCompat = compatApplicationConfig(application)
+ if err := s.applicationCompat.Init(); err != nil {
+ return err
+ }
+ s.metadataType = s.applicationCompat.MetadataType
+ if srv.Group == "" {
+ srv.Group = s.applicationCompat.Group
+ }
+ if srv.Version == "" {
+ srv.Version = s.applicationCompat.Version
+ }
+ }
+ s.unexported = atomic.NewBool(false)
+ err := s.check()
+ if err != nil {
+ panic(err)
+ }
+ s.Export = true
+ return commonCfg.Verify(s)
+}
+
+func (s *ServerOptions) check() error {
+ srv := s.Server
+ // check if the limiter has been imported
+ if srv.TpsLimiter != "" {
+ _, err := extension.GetTpsLimiter(srv.TpsLimiter)
+ if err != nil {
+ panic(err)
+ }
+ }
+ if srv.TpsLimitStrategy != "" {
+ _, err := extension.GetTpsLimitStrategyCreator(srv.TpsLimitStrategy)
+ if err != nil {
+ panic(err)
+ }
+ }
+ if srv.TpsLimitRejectedHandler != "" {
+ _, err := extension.GetRejectedExecutionHandler(srv.TpsLimitRejectedHandler)
+ if err != nil {
+ panic(err)
+ }
+ }
+
+ if srv.TpsLimitInterval != "" {
+ tpsLimitInterval, err := strconv.ParseInt(srv.TpsLimitInterval, 0, 0)
+ if err != nil {
+ return fmt.Errorf("[ServiceConfig] Cannot parse the configuration tps.limit.interval for service %s, please check your configuration", srv.Interface)
+ }
+ if tpsLimitInterval < 0 {
+ return fmt.Errorf("[ServiceConfig] The configuration tps.limit.interval for service %s must be positive, please check your configuration", srv.Interface)
+ }
+ }
+
+ if srv.TpsLimitRate != "" {
+ tpsLimitRate, err := strconv.ParseInt(srv.TpsLimitRate, 0, 0)
+ if err != nil {
+ return fmt.Errorf("[ServiceConfig] Cannot parse the configuration tps.limit.rate for service %s, please check your configuration", srv.Interface)
+ }
+ if tpsLimitRate < 0 {
+ return fmt.Errorf("[ServiceConfig] The configuration tps.limit.rate for service %s must be positive, please check your configuration", srv.Interface)
+ }
+ }
+ return nil
+}
+
+// InitExported will set exported as false atom bool
+func (s *ServerOptions) InitExported() {
+ s.exported = atomic.NewBool(false)
+}
+
+// IsExport will return whether the service config is exported or not
+func (s *ServerOptions) IsExport() bool {
+ return s.exported.Load()
+}
+
+// Get Random Port
+func getRandomPort(protocolConfigs []*config.ProtocolConfig) *list.List {
+ ports := list.New()
+ for _, proto := range protocolConfigs {
+ if len(proto.Port) > 0 {
+ continue
+ }
+
+ tcp, err := gxnet.ListenOnTCPRandomPort(proto.Ip)
+ if err != nil {
+ panic(perrors.New(fmt.Sprintf("Get tcp port error, err is {%v}", err)))
+ }
+ defer tcp.Close()
+ ports.PushBack(strings.Split(tcp.Addr().String(), ":")[1])
+ }
+ return ports
+}
+
+func (s *ServerOptions) ExportWithoutInfo() error {
+ return s.export(nil)
+}
+
+func (s *ServerOptions) ExportWithInfo(info *ServiceInfo) error {
+ return s.export(info)
+}
+
+func (s *ServerOptions) export(info *ServiceInfo) error {
+ srv := s.Server
+
+ var methodInfos []MethodInfo
+ var methods string
+ if info != nil {
+ srv.Interface = info.InterfaceName
+ methodInfos = info.Methods
+ s.Id = info.InterfaceName
+ s.info = info
+ }
+ // TODO: delay export
+ if s.unexported != nil && s.unexported.Load() {
+ err := perrors.Errorf("The service %v has already unexported!", srv.Interface)
+ logger.Errorf(err.Error())
+ return err
+ }
+ if s.exported != nil && s.exported.Load() {
+ logger.Warnf("The service %v has already exported!", srv.Interface)
+ return nil
+ }
+
+ regUrls := make([]*common.URL, 0)
+ if !srv.NotRegister {
+ regUrls = config.LoadRegistries(srv.RegistryIDs, s.registriesCompat, common.PROVIDER)
+ }
+
+ urlMap := s.getUrlMap()
+ protocolConfigs := loadProtocol(srv.ProtocolIDs, s.protocolCompat)
+ if len(protocolConfigs) == 0 {
+ logger.Warnf("The service %v's '%v' protocols don't has right protocolConfigs, Please check your configuration center and transfer protocol ", srv.Interface, srv.ProtocolIDs)
+ return nil
+ }
+
+ ports := getRandomPort(protocolConfigs)
+ nextPort := ports.Front()
+ proxyFactory := extension.GetProxyFactory(s.ProxyFactoryKey)
+ for _, proto := range protocolConfigs {
+ if info != nil {
+ for _, info := range methodInfos {
+ methods += info.Name + ","
+ }
+ } else {
+ // registry the service reflect
+ var err error
+ methods, err = common.ServiceMap.Register(srv.Interface, proto.Name, srv.Group, srv.Version, s.rpcService)
+ if err != nil {
+ formatErr := perrors.Errorf("The service %v export the protocol %v error! Error message is %v.",
+ srv.Interface, proto.Name, err.Error())
+ logger.Errorf(formatErr.Error())
+ return formatErr
+ }
+ }
+
+ port := proto.Port
+ if len(proto.Port) == 0 {
+ port = nextPort.Value.(string)
+ nextPort = nextPort.Next()
+ }
+ ivkURL := common.NewURLWithOptions(
+ common.WithPath(srv.Interface),
+ common.WithProtocol(proto.Name),
+ common.WithIp(proto.Ip),
+ common.WithPort(port),
+ common.WithParams(urlMap),
+ common.WithParamsValue(constant.BeanNameKey, s.Id),
+ //common.WithParamsValue(constant.SslEnabledKey, strconv.FormatBool(config.GetSslEnabled())),
+ common.WithMethods(strings.Split(methods, ",")),
+ common.WithMethodInfos(methodInfos),
+ common.WithToken(srv.Token),
+ common.WithParamsValue(constant.MetadataTypeKey, s.metadataType),
+ // fix https://github.com/apache/dubbo-go/issues/2176
+ common.WithParamsValue(constant.MaxServerSendMsgSize, proto.MaxServerSendMsgSize),
+ common.WithParamsValue(constant.MaxServerRecvMsgSize, proto.MaxServerRecvMsgSize),
+ )
+ if len(srv.Tag) > 0 {
+ ivkURL.AddParam(constant.Tagkey, srv.Tag)
+ }
+
+ // post process the URL to be exported
+ s.postProcessConfig(ivkURL)
+ // config post processor may set "export" to false
+ if !ivkURL.GetParamBool(constant.ExportKey, true) {
+ return nil
+ }
+
+ if len(regUrls) > 0 {
+ s.cacheMutex.Lock()
+ if s.cacheProtocol == nil {
+ logger.Debugf(fmt.Sprintf("First load the registry protocol, url is {%v}!", ivkURL))
+ s.cacheProtocol = extension.GetProtocol(constant.RegistryProtocol)
+ }
+ s.cacheMutex.Unlock()
+
+ for _, regUrl := range regUrls {
+ setRegistrySubURL(ivkURL, regUrl)
+ invoker := proxyFactory.GetInvoker(regUrl)
+ exporter := s.cacheProtocol.Export(invoker)
+ if exporter == nil {
+ return perrors.New(fmt.Sprintf("Registry protocol new exporter error, registry is {%v}, url is {%v}", regUrl, ivkURL))
+ }
+ s.exporters = append(s.exporters, exporter)
+ }
+ } else {
+ if ivkURL.GetParam(constant.InterfaceKey, "") == constant.MetadataServiceName {
+ ms, err := extension.GetLocalMetadataService("")
+ if err != nil {
+ logger.Warnf("export org.apache.dubbo.metadata.MetadataService failed beacause of %s ! pls check if you import _ \"dubbo.apache.org/dubbo-go/v3/metadata/service/local\"", err)
+ return nil
+ }
+ if err := ms.SetMetadataServiceURL(ivkURL); err != nil {
+ logger.Warnf("SetMetadataServiceURL error = %s", err)
+ }
+ }
+ invoker := proxyFactory.GetInvoker(ivkURL)
+ exporter := extension.GetProtocol(protocolwrapper.FILTER).Export(invoker)
+ if exporter == nil {
+ return perrors.New(fmt.Sprintf("Filter protocol without registry new exporter error, url is {%v}", ivkURL))
+ }
+ s.exporters = append(s.exporters, exporter)
+ }
+ publishServiceDefinition(ivkURL)
+ }
+ s.exported.Store(true)
+ return nil
+}
+
+// setRegistrySubURL set registry sub url is ivkURl
+func setRegistrySubURL(ivkURL *common.URL, regUrl *common.URL) {
+ ivkURL.AddParam(constant.RegistryKey, regUrl.GetParam(constant.RegistryKey, ""))
+ regUrl.SubURL = ivkURL
+}
+
+// loadProtocol filter protocols by ids
+func loadProtocol(protocolIds []string, protocols map[string]*config.ProtocolConfig) []*config.ProtocolConfig {
+ returnProtocols := make([]*config.ProtocolConfig, 0, len(protocols))
+ for _, v := range protocolIds {
+ for k, config := range protocols {
+ if v == k {
+ returnProtocols = append(returnProtocols, config)
+ }
+ }
+ }
+ return returnProtocols
+}
+
+// Unexport will call unexport of all exporters service config exported
+func (s *ServerOptions) Unexport() {
+ if !s.exported.Load() {
+ return
+ }
+ if s.unexported.Load() {
+ return
+ }
+
+ func() {
+ s.exportersLock.Lock()
+ defer s.exportersLock.Unlock()
+ for _, exporter := range s.exporters {
+ exporter.UnExport()
+ }
+ s.exporters = nil
+ }()
+
+ s.exported.Store(false)
+ s.unexported.Store(true)
+}
+
+// Implement only store the @s and return
+func (s *ServerOptions) Implement(rpcService common.RPCService) {
+ s.rpcService = rpcService
+}
+
+func (s *ServerOptions) getUrlMap() url.Values {
+ srv := s.Server
+ app := s.applicationCompat
+
+ urlMap := url.Values{}
+ // first set user params
+ for k, v := range srv.Params {
+ urlMap.Set(k, v)
+ }
+ urlMap.Set(constant.InterfaceKey, srv.Interface)
+ urlMap.Set(constant.TimestampKey, strconv.FormatInt(time.Now().Unix(), 10))
+ urlMap.Set(constant.ClusterKey, srv.Cluster)
+ urlMap.Set(constant.LoadbalanceKey, srv.Loadbalance)
+ urlMap.Set(constant.WarmupKey, srv.Warmup)
+ urlMap.Set(constant.RetriesKey, srv.Retries)
+ if srv.Group != "" {
+ urlMap.Set(constant.GroupKey, srv.Group)
+ }
+ if srv.Version != "" {
+ urlMap.Set(constant.VersionKey, srv.Version)
+ }
+ urlMap.Set(constant.RegistryRoleKey, strconv.Itoa(common.PROVIDER))
+ urlMap.Set(constant.ReleaseKey, "dubbo-golang-"+constant.Version)
+ urlMap.Set(constant.SideKey, (common.RoleType(common.PROVIDER)).Role())
+ // todo: move
+ urlMap.Set(constant.SerializationKey, srv.Serialization)
+ // application config info
+ urlMap.Set(constant.ApplicationKey, app.Name)
+ urlMap.Set(constant.OrganizationKey, app.Organization)
+ urlMap.Set(constant.NameKey, app.Name)
+ urlMap.Set(constant.ModuleKey, app.Module)
+ urlMap.Set(constant.AppVersionKey, app.Version)
+ urlMap.Set(constant.OwnerKey, app.Owner)
+ urlMap.Set(constant.EnvironmentKey, app.Environment)
+
+ // filter
+ var filters string
+ if srv.Filter == "" {
+ filters = constant.DefaultServiceFilters
+ } else {
+ filters = srv.Filter
+ }
+ if s.adaptiveService {
+ filters += fmt.Sprintf(",%s", constant.AdaptiveServiceProviderFilterKey)
+ }
+ urlMap.Set(constant.ServiceFilterKey, filters)
+
+ // filter special config
+ urlMap.Set(constant.AccessLogFilterKey, srv.AccessLog)
+ // tps limiter
+ urlMap.Set(constant.TPSLimitStrategyKey, srv.TpsLimitStrategy)
+ urlMap.Set(constant.TPSLimitIntervalKey, srv.TpsLimitInterval)
+ urlMap.Set(constant.TPSLimitRateKey, srv.TpsLimitRate)
+ urlMap.Set(constant.TPSLimiterKey, srv.TpsLimiter)
+ urlMap.Set(constant.TPSRejectedExecutionHandlerKey, srv.TpsLimitRejectedHandler)
+ urlMap.Set(constant.TracingConfigKey, srv.TracingKey)
+
+ // execute limit filter
+ urlMap.Set(constant.ExecuteLimitKey, srv.ExecuteLimit)
+ urlMap.Set(constant.ExecuteRejectedExecutionHandlerKey, srv.ExecuteLimitRejectedHandler)
+
+ // auth filter
+ urlMap.Set(constant.ServiceAuthKey, srv.Auth)
+ urlMap.Set(constant.ParameterSignatureEnableKey, srv.ParamSign)
+
+ // whether to export or not
+ urlMap.Set(constant.ExportKey, strconv.FormatBool(s.Export))
+ urlMap.Set(constant.PIDKey, fmt.Sprintf("%d", os.Getpid()))
+
+ for _, v := range srv.Methods {
+ prefix := "methods." + v.Name + "."
+ urlMap.Set(prefix+constant.LoadbalanceKey, v.LoadBalance)
+ urlMap.Set(prefix+constant.RetriesKey, v.Retries)
+ urlMap.Set(prefix+constant.WeightKey, strconv.FormatInt(v.Weight, 10))
+
+ urlMap.Set(prefix+constant.TPSLimitStrategyKey, v.TpsLimitStrategy)
+ urlMap.Set(prefix+constant.TPSLimitIntervalKey, v.TpsLimitInterval)
+ urlMap.Set(prefix+constant.TPSLimitRateKey, v.TpsLimitRate)
+
+ urlMap.Set(constant.ExecuteLimitKey, v.ExecuteLimit)
+ urlMap.Set(constant.ExecuteRejectedExecutionHandlerKey, v.ExecuteLimitRejectedHandler)
+ }
+
+ return urlMap
+}
+
+// GetExportedUrls will return the url in service config's exporter
+func (s *ServerOptions) GetExportedUrls() []*common.URL {
+ if s.exported.Load() {
+ var urls []*common.URL
+ for _, exporter := range s.exporters {
+ urls = append(urls, exporter.GetInvoker().GetURL())
+ }
+ return urls
+ }
+ return nil
+}
+
+// postProcessConfig asks registered ConfigPostProcessor to post-process the current ServiceConfig.
+func (s *ServerOptions) postProcessConfig(url *common.URL) {
+ for _, p := range extension.GetConfigPostProcessors() {
+ p.PostProcessServiceConfig(url)
+ }
+}
+
+func publishServiceDefinition(url *common.URL) {
+ localService, err := extension.GetLocalMetadataService(constant.DefaultKey)
+ if err != nil {
+ logger.Warnf("get local metadata service failed, please check if you have imported _ \"dubbo.apache.org/dubbo-go/v3/metadata/service/local\"")
+ return
+ }
+ localService.PublishServiceDefinition(url)
+ if url.GetParam(constant.MetadataTypeKey, "") != constant.RemoteMetadataStorageType {
+ return
+ }
+ if remoteMetadataService, err := extension.GetRemoteMetadataService(); err == nil && remoteMetadataService != nil {
+ remoteMetadataService.PublishServiceDefinition(url)
+ }
+}
diff --git a/server/compat.go b/server/compat.go
new file mode 100644
index 000000000..f22c8cc9f
--- /dev/null
+++ b/server/compat.go
@@ -0,0 +1,79 @@
+/*
+ * 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 (
+ "dubbo.apache.org/dubbo-go/v3/config"
+ "dubbo.apache.org/dubbo-go/v3/global"
+)
+
+// these functions are used to resolve circular dependencies temporarily.
+// please refer to issue(https://github.com/apache/dubbo-go/issues/2377)
+// todo(DMwangnima): remove these functions when refactoring dubbo-go
+func compatApplicationConfig(c *global.ApplicationConfig) *config.ApplicationConfig {
+ return &config.ApplicationConfig{
+ Organization: c.Organization,
+ Name: c.Name,
+ Module: c.Module,
+ Group: c.Group,
+ Version: c.Version,
+ Owner: c.Owner,
+ Environment: c.Environment,
+ MetadataType: c.MetadataType,
+ Tag: c.Tag,
+ }
+}
+
+func compatRegistryConfig(c *global.RegistryConfig) *config.RegistryConfig {
+ return &config.RegistryConfig{
+ Protocol: c.Protocol,
+ Timeout: c.Timeout,
+ Group: c.Group,
+ Namespace: c.Namespace,
+ TTL: c.TTL,
+ Address: c.Address,
+ Username: c.Username,
+ Password: c.Password,
+ Simplified: c.Simplified,
+ Preferred: c.Preferred,
+ Zone: c.Zone,
+ Weight: c.Weight,
+ Params: c.Params,
+ RegistryType: c.RegistryType,
+ UseAsMetaReport: c.UseAsMetaReport,
+ UseAsConfigCenter: c.UseAsConfigCenter,
+ }
+}
+
+func compatMethodConfig(c *global.MethodConfig) *config.MethodConfig {
+ return &config.MethodConfig{
+ InterfaceId: c.InterfaceId,
+ InterfaceName: c.InterfaceName,
+ Name: c.Name,
+ Retries: c.Retries,
+ LoadBalance: c.LoadBalance,
+ Weight: c.Weight,
+ TpsLimitInterval: c.TpsLimitInterval,
+ TpsLimitRate: c.TpsLimitRate,
+ TpsLimitStrategy: c.TpsLimitStrategy,
+ ExecuteLimit: c.ExecuteLimit,
+ ExecuteLimitRejectedHandler: c.ExecuteLimitRejectedHandler,
+ Sticky: c.Sticky,
+ RequestTimeout: c.RequestTimeout,
+ }
+}
diff --git a/server/options.go b/server/options.go
index 01c35d57f..36b75a8ac 100644
--- a/server/options.go
+++ b/server/options.go
@@ -1,6 +1,175 @@
+/*
+ * 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
-type Options struct {
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/config"
+ "dubbo.apache.org/dubbo-go/v3/global"
+ "dubbo.apache.org/dubbo-go/v3/protocol"
+ "go.uber.org/atomic"
+ "sync"
+)
+
+type ServerOptions struct {
+ Application *global.ApplicationConfig
+ Provider *global.ProviderConfig
+ Server *global.ServiceConfig
+ Registries map[string]*global.RegistryConfig
+
+ Id string
+ unexported *atomic.Bool
+ exported *atomic.Bool
+ Export bool
+ metadataType string
+ info *ServiceInfo
+ ProxyFactoryKey string
+ rpcService common.RPCService
+ cacheMutex sync.Mutex
+ cacheProtocol protocol.Protocol
+ exportersLock sync.Mutex
+ exporters []protocol.Exporter
+ adaptiveService bool
+
+ methodsCompat []*config.MethodConfig
+ applicationCompat *config.ApplicationConfig
+ registriesCompat map[string]*config.RegistryConfig
+ protocolCompat map[string]*config.ProtocolConfig
+}
+
+func defaultServerOptions() *ServerOptions {
+ return &ServerOptions{
+ Provider: global.DefaultProviderConfig(),
+ }
}
-type Option func(*Options)
+func (srvOpts *ServerOptions) init(opts ...ServerOption) error {
+ for _, opt := range opts {
+ opt(srvOpts)
+ }
+ return nil
+}
+
+type ServerOption func(*ServerOptions)
+
+// ---------- For user ----------
+
+func WithRegistryIDs(registryIDs []string) ServerOption {
+ return func(cfg *ServerOptions) {
+ if len(registryIDs) <= 0 {
+ cfg.Server.RegistryIDs = registryIDs
+ }
+ }
+}
+
+func WithFilter(filter string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Filter = filter
+ }
+}
+
+func WithProtocolIDs(protocolIDs []string) ServerOption {
+ return func(cfg *ServerOptions) {
+ if len(protocolIDs) <= 0 {
+ cfg.Server.ProtocolIDs = protocolIDs
+ }
+ }
+}
+
+func WithTracingKey(tracingKey string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.TracingKey = tracingKey
+ }
+}
+
+func WithLoadBalance(loadBalance string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Loadbalance = loadBalance
+ }
+}
+
+func WithWarmUp(warmUp string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Warmup = warmUp
+ }
+}
+
+func WithCluster(cluster string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Cluster = cluster
+ }
+}
+
+func WithGroup(group string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Group = group
+ }
+}
+
+func WithVersion(version string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Version = version
+ }
+}
+
+func WithSerialization(serialization string) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.Serialization = serialization
+ }
+}
+
+func WithNotRegister(notRegister bool) ServerOption {
+ return func(cfg *ServerOptions) {
+ cfg.Server.NotRegister = notRegister
+ }
+}
+
+// ----------From framework----------
+
+func WithApplicationConfig(opts ...global.ApplicationOption) ServerOption {
+ appCfg := new(global.ApplicationConfig)
+ for _, opt := range opts {
+ opt(appCfg)
+ }
+
+ return func(opts *ServerOptions) {
+ opts.Application = appCfg
+ }
+}
+
+func WithProviderConfig(opts ...global.ProviderOption) ServerOption {
+ providerCfg := new(global.ProviderConfig)
+ for _, opt := range opts {
+ opt(providerCfg)
+ }
+
+ return func(opts *ServerOptions) {
+ opts.Provider = providerCfg
+ }
+}
+
+func WithServiceConfig(opts ...global.ServiceOption) ServerOption {
+ serviceCfg := new(global.ServiceConfig)
+ for _, opt := range opts {
+ opt(serviceCfg)
+ }
+
+ return func(opts *ServerOptions) {
+ opts.Server = serviceCfg
+ }
+}
diff --git a/server/server.go b/server/server.go
index b860be890..763af3837 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1,22 +1,35 @@
+/*
+ * 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"
+ "dubbo.apache.org/dubbo-go/v3/protocol"
)
type Server struct {
-}
+ invoker protocol.Invoker
+ info *ServiceInfo
-// Register assemble invoker chains like ProviderConfig.Load, init a service per call
-func (pro *Server) Register(handler interface{}, info *ServiceInfo, opts ...Option) error {
- // put information from info to url
- // ProviderConfig.Load
-
- // url
- return nil
+ cfg *ServerOptions
}
-// meta
+// ServiceInfo is meta info of a service
type ServiceInfo struct {
InterfaceName string
ServiceType interface{}
@@ -24,6 +37,17 @@ type ServiceInfo struct {
Meta map[string]interface{}
}
+// Register assemble invoker chains like ProviderConfig.Load, init a service per call
+func (s *Server) Register(handler interface{}, info *ServiceInfo, opts ...ServerOption) error {
+ for _, opt := range opts {
+ opt(s.cfg)
+ }
+
+ // ProviderConfig.Load
+ // url
+ return nil
+}
+
type MethodInfo struct {
Name string
Type string
@@ -33,6 +57,12 @@ type MethodInfo struct {
Meta map[string]interface{}
}
-func NewServer() (*Server, error) {
- return nil, nil
+func NewServer(opts ...ServerOption) (*Server, error) {
+ newSrvOpts := defaultServerOptions()
+ if err := newSrvOpts.init(opts...); err != nil {
+ return nil, err
+ }
+ return &Server{
+ cfg: newSrvOpts,
+ }, nil
}