Merge pull request #237 from flycash/feature/TpsLimitFilter

Ftr: TpsLimitSupport

common/constant/default.go

@@ -46,7 +46,7 @@ const (
 const (
 	DEFAULT_KEY               = "default"
 	PREFIX_DEFAULT_KEY        = "default."
-	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog"
+	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps"
 	DEFAULT_REFERENCE_FILTERS = ""
 	GENERIC_REFERENCE_FILTERS = "generic"
 	GENERIC                   = "$invoke"

common/constant/key.go

@@ -48,19 +48,26 @@ const (
 )
 
 const (
-	TIMESTAMP_KEY        = "timestamp"
-	REMOTE_TIMESTAMP_KEY = "remote.timestamp"
-	CLUSTER_KEY          = "cluster"
-	LOADBALANCE_KEY      = "loadbalance"
-	WEIGHT_KEY           = "weight"
-	WARMUP_KEY           = "warmup"
-	RETRIES_KEY          = "retries"
-	BEAN_NAME            = "bean.name"
-	FAIL_BACK_TASKS_KEY  = "failbacktasks"
-	FORKS_KEY            = "forks"
-	DEFAULT_FORKS        = 2
-	DEFAULT_TIMEOUT      = 1000
-	ACCESS_LOG_KEY       = "accesslog"
+	TIMESTAMP_KEY                      = "timestamp"
+	REMOTE_TIMESTAMP_KEY               = "remote.timestamp"
+	CLUSTER_KEY                        = "cluster"
+	LOADBALANCE_KEY                    = "loadbalance"
+	WEIGHT_KEY                         = "weight"
+	WARMUP_KEY                         = "warmup"
+	RETRIES_KEY                        = "retries"
+	BEAN_NAME                          = "bean.name"
+	FAIL_BACK_TASKS_KEY                = "failbacktasks"
+	FORKS_KEY                          = "forks"
+	DEFAULT_FORKS                      = 2
+	DEFAULT_TIMEOUT                    = 1000
+	ACCESS_LOG_KEY                     = "accesslog"
+	TPS_LIMITER_KEY                    = "tps.limiter"
+	TPS_REJECTED_EXECUTION_HANDLER_KEY = "tps.limit.rejected.handler"
+	TPS_LIMIT_RATE_KEY                 = "tps.limit.rate"
+	DEFAULT_TPS_LIMIT_RATE             = "-1"
+	TPS_LIMIT_INTERVAL_KEY             = "tps.limit.interval"
+	DEFAULT_TPS_LIMIT_INTERVAL         = "60000"
+	TPS_LIMIT_STRATEGY_KEY             = "tps.limit.strategy"
 )
 
 const (

common/extension/tps_limit.go

+/*
+ * 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 extension
+
+import (
+	"github.com/apache/dubbo-go/filter/impl/tps"
+)
+
+var (
+	tpsLimitStrategy            = make(map[string]func(rate int, interval int) tps.TpsLimitStrategy)
+	tpsLimiter                  = make(map[string]func() tps.TpsLimiter)
+	tpsRejectedExecutionHandler = make(map[string]func() tps.RejectedExecutionHandler)
+)
+
+func SetTpsLimiter(name string, creator func() tps.TpsLimiter) {
+	tpsLimiter[name] = creator
+}
+
+func GetTpsLimiter(name string) tps.TpsLimiter {
+	creator, ok := tpsLimiter[name]
+	if !ok {
+		panic("TpsLimiter for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetTpsLimiter.")
+	}
+	return creator()
+}
+
+func SetTpsLimitStrategy(name string, creator func(rate int, interval int) tps.TpsLimitStrategy) {
+	tpsLimitStrategy[name] = creator
+}
+
+func GetTpsLimitStrategyCreator(name string) func(rate int, interval int) tps.TpsLimitStrategy {
+	creator, ok := tpsLimitStrategy[name]
+	if !ok {
+		panic("TpsLimitStrategy for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetTpsLimitStrategy.")
+	}
+	return creator
+}
+
+func SetTpsRejectedExecutionHandler(name string, creator func() tps.RejectedExecutionHandler) {
+	tpsRejectedExecutionHandler[name] = creator
+}
+
+func GetTpsRejectedExecutionHandler(name string) tps.RejectedExecutionHandler {
+	creator, ok := tpsRejectedExecutionHandler[name]
+	if !ok {
+		panic("TpsRejectedExecutionHandler for " + name + " is not existing, make sure you have import the package " +
+			"and you have register it by invoking extension.SetTpsRejectedExecutionHandler.")
+	}
+	return creator()
+}

config/method_config.go

@@ -25,12 +25,15 @@ import (
 )
 
 type MethodConfig struct {
-	InterfaceId   string
-	InterfaceName string
-	Name          string `yaml:"name"  json:"name,omitempty" property:"name"`
-	Retries       string `yaml:"retries"  json:"retries,omitempty" property:"retries"`
-	Loadbalance   string `yaml:"loadbalance"  json:"loadbalance,omitempty" property:"loadbalance"`
-	Weight        int64  `yaml:"weight"  json:"weight,omitempty" property:"weight"`
+	InterfaceId      string
+	InterfaceName    string
+	Name             string `yaml:"name"  json:"name,omitempty" property:"name"`
+	Retries          string `yaml:"retries"  json:"retries,omitempty" property:"retries"`
+	Loadbalance      string `yaml:"loadbalance"  json:"loadbalance,omitempty" property:"loadbalance"`
+	Weight           int64  `yaml:"weight"  json:"weight,omitempty" property:"weight"`
+	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"`
 }
 
 func (c *MethodConfig) Prefix() string {

config/service_config.go

@@ -43,27 +43,32 @@ import (
 )
 
 type ServiceConfig struct {
-	context       context.Context
-	id            string
-	Filter        string            `yaml:"filter" json:"filter,omitempty" property:"filter"`
-	Protocol      string            `default:"dubbo"  required:"true"  yaml:"protocol"  json:"protocol,omitempty" property:"protocol"` //multi protocol support, split by ','
-	InterfaceName string            `required:"true"  yaml:"interface"  json:"interface,omitempty" property:"interface"`
-	Registry      string            `yaml:"registry"  json:"registry,omitempty"  property:"registry"`
-	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"`
-	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"`
-	unexported    *atomic.Bool
-	exported      *atomic.Bool
-	rpcService    common.RPCService
-	cacheProtocol protocol.Protocol
-	cacheMutex    sync.Mutex
+	context                 context.Context
+	id                      string
+	Filter                  string            `yaml:"filter" json:"filter,omitempty" property:"filter"`
+	Protocol                string            `default:"dubbo"  required:"true"  yaml:"protocol"  json:"protocol,omitempty" property:"protocol"` // multi protocol support, split by ','
+	InterfaceName           string            `required:"true"  yaml:"interface"  json:"interface,omitempty" property:"interface"`
+	Registry                string            `yaml:"registry"  json:"registry,omitempty"  property:"registry"`
+	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"`
+	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"`
+	unexported              *atomic.Bool
+	exported                *atomic.Bool
+	rpcService              common.RPCService
+	cacheProtocol           protocol.Protocol
+	cacheMutex              sync.Mutex
 }
 
 func (c *ServiceConfig) Prefix() string {
@@ -94,9 +99,9 @@ func NewServiceConfig(id string, context context.Context) *ServiceConfig {
 }
 
 func (srvconfig *ServiceConfig) Export() error {
-	//TODO: config center start here
+	// TODO: config center start here
 
-	//TODO:delay export
+	// TODO:delay export
 	if srvconfig.unexported != nil && srvconfig.unexported.Load() {
 		err := perrors.Errorf("The service %v has already unexported! ", srvconfig.InterfaceName)
 		logger.Errorf(err.Error())
@@ -111,7 +116,7 @@ func (srvconfig *ServiceConfig) Export() error {
 	urlMap := srvconfig.getUrlMap()
 
 	for _, proto := range loadProtocol(srvconfig.Protocol, providerConfig.Protocols) {
-		//registry the service reflect
+		// registry the service reflect
 		methods, err := common.ServiceMap.Register(proto.Name, srvconfig.rpcService)
 		if err != nil {
 			err := perrors.Errorf("The service %v  export the protocol %v error! Error message is %v .", srvconfig.InterfaceName, proto.Name, err.Error())
@@ -164,7 +169,7 @@ func (srvconfig *ServiceConfig) Implement(s common.RPCService) {
 
 func (srvconfig *ServiceConfig) getUrlMap() url.Values {
 	urlMap := url.Values{}
-	//first set user params
+	// first set user params
 	for k, v := range srvconfig.Params {
 		urlMap.Set(k, v)
 	}
@@ -177,7 +182,7 @@ func (srvconfig *ServiceConfig) getUrlMap() url.Values {
 	urlMap.Set(constant.GROUP_KEY, srvconfig.Group)
 	urlMap.Set(constant.VERSION_KEY, srvconfig.Version)
 	urlMap.Set(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER))
-	//application info
+	// application info
 	urlMap.Set(constant.APPLICATION_KEY, providerConfig.ApplicationConfig.Name)
 	urlMap.Set(constant.ORGANIZATION_KEY, providerConfig.ApplicationConfig.Organization)
 	urlMap.Set(constant.NAME_KEY, providerConfig.ApplicationConfig.Name)
@@ -186,16 +191,27 @@ func (srvconfig *ServiceConfig) getUrlMap() url.Values {
 	urlMap.Set(constant.OWNER_KEY, providerConfig.ApplicationConfig.Owner)
 	urlMap.Set(constant.ENVIRONMENT_KEY, providerConfig.ApplicationConfig.Environment)
 
-	//filter
+	// filter
 	urlMap.Set(constant.SERVICE_FILTER_KEY, mergeValue(providerConfig.Filter, srvconfig.Filter, constant.DEFAULT_SERVICE_FILTERS))
 
-	//filter special config
+	// filter special config
 	urlMap.Set(constant.ACCESS_LOG_KEY, srvconfig.AccessLog)
+	// tps limiter
+	urlMap.Set(constant.TPS_LIMIT_STRATEGY_KEY, srvconfig.TpsLimitStrategy)
+	urlMap.Set(constant.TPS_LIMIT_INTERVAL_KEY, srvconfig.TpsLimitInterval)
+	urlMap.Set(constant.TPS_LIMIT_RATE_KEY, srvconfig.TpsLimitRate)
+	urlMap.Set(constant.TPS_LIMITER_KEY, srvconfig.TpsLimiter)
+	urlMap.Set(constant.TPS_REJECTED_EXECUTION_HANDLER_KEY, srvconfig.TpsLimitRejectedHandler)
 
 	for _, v := range srvconfig.Methods {
-		urlMap.Set("methods."+v.Name+"."+constant.LOADBALANCE_KEY, v.Loadbalance)
-		urlMap.Set("methods."+v.Name+"."+constant.RETRIES_KEY, v.Retries)
-		urlMap.Set("methods."+v.Name+"."+constant.WEIGHT_KEY, strconv.FormatInt(v.Weight, 10))
+		prefix := "methods." + v.Name + "."
+		urlMap.Set(prefix+constant.LOADBALANCE_KEY, v.Loadbalance)
+		urlMap.Set(prefix+constant.RETRIES_KEY, v.Retries)
+		urlMap.Set(prefix+constant.WEIGHT_KEY, strconv.FormatInt(v.Weight, 10))
+
+		urlMap.Set(prefix+constant.TPS_LIMIT_STRATEGY_KEY, v.TpsLimitStrategy)
+		urlMap.Set(prefix+constant.TPS_LIMIT_INTERVAL_KEY, v.TpsLimitInterval)
+		urlMap.Set(prefix+constant.TPS_LIMIT_RATE_KEY, v.TpsLimitRate)
 	}
 
 	return urlMap

config/testdata/provider_config.yml

@@ -29,6 +29,15 @@ services:
   "UserProvider":
     registry: "hangzhouzk,shanghaizk"
     filter: ""
+    # the name of limiter
+    tps.limiter: "default"
+    # the time unit of interval is ms
+    tps.limit.interval: 60000
+    tps.limit.rate: 200
+    # the name of strategy
+    tps.limit.strategy: "slidingWindow"
+    # the name of RejectedExecutionHandler
+    tps.limit.rejected.handler: "default"
     protocol : "dubbo"
     # equivalent to interface of dubbo.xml
     interface : "com.ikurento.user.UserProvider"

filter/impl/access_log_filter.go

@@ -132,7 +132,7 @@ func (ef *AccessLogFilter) writeLogToFile(data AccessLogData) {
 
 	logFile, err := ef.openLogFile(accessLog)
 	if err != nil {
-		logger.Warnf("Can not open 	the access log file: %s, %v", accessLog, err)
+		logger.Warnf("Can not open the access log file: %s, %v", accessLog, err)
 		return
 	}
 	logger.Debugf("Append log to %s", accessLog)

filter/impl/tps/impl/rejected_execution_handler_mock.go

+//  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.
+//
+
+// Code generated by MockGen. DO NOT EDIT.
+// Source: rejected_execution_handler.go
+
+// Package filter is a generated GoMock package.
+package impl
+
+import (
+	common "github.com/apache/dubbo-go/common"
+	protocol "github.com/apache/dubbo-go/protocol"
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+)
+
+// MockRejectedExecutionHandler is a mock of RejectedExecutionHandler interface
+type MockRejectedExecutionHandler struct {
+	ctrl     *gomock.Controller
+	recorder *MockRejectedExecutionHandlerMockRecorder
+}
+
+// MockRejectedExecutionHandlerMockRecorder is the mock recorder for MockRejectedExecutionHandler
+type MockRejectedExecutionHandlerMockRecorder struct {
+	mock *MockRejectedExecutionHandler
+}
+
+// NewMockRejectedExecutionHandler creates a new mock instance
+func NewMockRejectedExecutionHandler(ctrl *gomock.Controller) *MockRejectedExecutionHandler {
+	mock := &MockRejectedExecutionHandler{ctrl: ctrl}
+	mock.recorder = &MockRejectedExecutionHandlerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockRejectedExecutionHandler) EXPECT() *MockRejectedExecutionHandlerMockRecorder {
+	return m.recorder
+}
+
+// RejectedExecution mocks base method
+func (m *MockRejectedExecutionHandler) RejectedExecution(url common.URL, invocation protocol.Invocation) protocol.Result {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RejectedExecution", url, invocation)
+	ret0, _ := ret[0].(protocol.Result)
+	return ret0
+}
+
+// RejectedExecution indicates an expected call of RejectedExecution
+func (mr *MockRejectedExecutionHandlerMockRecorder) RejectedExecution(url, invocation interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RejectedExecution", reflect.TypeOf((*MockRejectedExecutionHandler)(nil).RejectedExecution), url, invocation)
+}

filter/impl/tps/impl/rejected_execution_handler_only_log.go

+/*
+ * 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 impl
+
+import (
+	"sync"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const HandlerName = "log"
+
+func init() {
+	extension.SetTpsRejectedExecutionHandler(HandlerName, GetOnlyLogRejectedExecutionHandler)
+	extension.SetTpsRejectedExecutionHandler(constant.DEFAULT_KEY, GetOnlyLogRejectedExecutionHandler)
+}
+
+var onlyLogHandlerInstance *OnlyLogRejectedExecutionHandler
+var onlyLogHandlerOnce sync.Once
+
+/**
+ * This implementation only logs the invocation info.
+ * it always return en error inside the result.
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" # the name of limiter
+ *   tps.limit.rejected.handler: "default" or "log"
+ *   methods:
+ *    - name: "GetUser"
+ */
+type OnlyLogRejectedExecutionHandler struct {
+}
+
+func (handler *OnlyLogRejectedExecutionHandler) RejectedExecution(url common.URL, invocation protocol.Invocation) protocol.Result {
+	logger.Errorf("The invocation was rejected due to over rate limitation. url: %s", url.String())
+	return &protocol.RPCResult{}
+}
+
+func GetOnlyLogRejectedExecutionHandler() tps.RejectedExecutionHandler {
+	onlyLogHandlerOnce.Do(func() {
+		onlyLogHandlerInstance = &OnlyLogRejectedExecutionHandler{}
+	})
+	return onlyLogHandlerInstance
+}

filter/impl/tps/impl/rejected_execution_handler_only_log_test.go

+/*
+ * 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 impl
+
+import (
+	"net/url"
+	"testing"
+)
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+)
+
+func TestOnlyLogRejectedExecutionHandler_RejectedExecution(t *testing.T) {
+	handler := GetOnlyLogRejectedExecutionHandler()
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.INTERFACE_KEY, "methodName"))
+	handler.RejectedExecution(*invokeUrl, nil)
+}

filter/impl/tps/impl/tps_limit_fix_window_strategy.go

+/*
+ * 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 impl
+
+import (
+	"sync/atomic"
+	"time"
+)
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+)
+
+const (
+	FixedWindowKey = "fixedWindow"
+)
+
+func init() {
+	extension.SetTpsLimitStrategy(FixedWindowKey, NewFixedWindowTpsLimitStrategyImpl)
+	extension.SetTpsLimitStrategy(constant.DEFAULT_KEY, NewFixedWindowTpsLimitStrategyImpl)
+}
+
+/**
+ * It's the same as default implementation in Java
+ * It's not a thread-safe implementation.
+ * It you want to use the thread-safe implementation, please use ThreadSafeFixedWindowTpsLimitStrategyImpl
+ * This is the default implementation.
+ *
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" # the name of limiter
+ *   tps.limit.strategy: "default" or "fixedWindow" # service-level
+ *   methods:
+ *    - name: "GetUser"
+ *      tps.interval: 3000
+ *      tps.limit.strategy: "default" or "fixedWindow" # method-level
+ */
+type FixedWindowTpsLimitStrategyImpl struct {
+	rate      int32
+	interval  int64
+	count     int32
+	timestamp int64
+}
+
+func (impl *FixedWindowTpsLimitStrategyImpl) IsAllowable() bool {
+
+	current := time.Now().UnixNano()
+	if impl.timestamp+impl.interval < current {
+		// it's a new window
+		// if a lot of threads come here, the count will be set to 0 several times.
+		// so the return statement will be wrong.
+		impl.timestamp = current
+		impl.count = 0
+	}
+	// this operation is thread-safe, but count + 1 may be overflow
+	return atomic.AddInt32(&impl.count, 1) <= impl.rate
+}
+
+func NewFixedWindowTpsLimitStrategyImpl(rate int, interval int) tps.TpsLimitStrategy {
+	return &FixedWindowTpsLimitStrategyImpl{
+		rate:      int32(rate),
+		interval:  int64(interval * 1000), // convert to ns
+		count:     0,
+		timestamp: time.Now().UnixNano(),
+	}
+}

filter/impl/tps/impl/tps_limit_fix_window_strategy_test.go

+/*
+ * 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 impl
+
+import (
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFixedWindowTpsLimitStrategyImpl_IsAllowable(t *testing.T) {
+	strategy := NewFixedWindowTpsLimitStrategyImpl(2, 60000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+
+	strategy = NewFixedWindowTpsLimitStrategyImpl(2, 2000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+	time.Sleep(time.Duration(2100 * 1000))
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+}

filter/impl/tps/impl/tps_limit_sliding_window_strategy.go

+/*
+ * 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 impl
+
+import (
+	"container/list"
+	"sync"
+	"time"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+)
+
+func init() {
+	extension.SetTpsLimitStrategy("slidingWindow", NewSlidingWindowTpsLimitStrategyImpl)
+}
+
+/**
+ * it's thread-safe.
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" # the name of limiter
+ *   tps.limit.strategy: "slidingWindow" # service-level
+ *   methods:
+ *    - name: "GetUser"
+ *      tps.interval: 3000
+ *      tps.limit.strategy: "slidingWindow" # method-level
+ */
+type SlidingWindowTpsLimitStrategyImpl struct {
+	rate     int
+	interval int64
+	mutex    *sync.Mutex
+	queue    *list.List
+}
+
+func (impl *SlidingWindowTpsLimitStrategyImpl) IsAllowable() bool {
+	impl.mutex.Lock()
+	defer impl.mutex.Unlock()
+	// quick path
+	size := impl.queue.Len()
+	current := time.Now().UnixNano()
+	if size < impl.rate {
+		impl.queue.PushBack(current)
+		return true
+	}
+
+	// slow path
+	boundary := current - impl.interval
+
+	timestamp := impl.queue.Front()
+	// remove the element that out of the window
+	for timestamp != nil && timestamp.Value.(int64) < boundary {
+		impl.queue.Remove(timestamp)
+		timestamp = impl.queue.Front()
+	}
+	if impl.queue.Len() < impl.rate {
+		impl.queue.PushBack(current)
+		return true
+	}
+	return false
+}
+
+func NewSlidingWindowTpsLimitStrategyImpl(rate int, interval int) tps.TpsLimitStrategy {
+	return &SlidingWindowTpsLimitStrategyImpl{
+		rate:     rate,
+		interval: int64(interval * 1000),
+		mutex:    &sync.Mutex{},
+		queue:    list.New(),
+	}
+}

filter/impl/tps/impl/tps_limit_sliding_window_strategy_test.go

+/*
+ * 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 impl
+
+import (
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSlidingWindowTpsLimitStrategyImpl_IsAllowable(t *testing.T) {
+	strategy := NewSlidingWindowTpsLimitStrategyImpl(2, 60000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+	time.Sleep(time.Duration(2100 * 1000))
+	assert.False(t, strategy.IsAllowable())
+
+	strategy = NewSlidingWindowTpsLimitStrategyImpl(2, 2000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+	time.Sleep(time.Duration(2100 * 1000))
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+}

filter/impl/tps/impl/tps_limit_strategy_mock.go

+//  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.
+//
+
+// Code generated by MockGen. DO NOT EDIT.
+// Source: tps_limit_strategy.go
+
+// Package filter is a generated GoMock package.
+package impl
+
+import (
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+)
+
+// MockTpsLimitStrategy is a mock of TpsLimitStrategy interface
+type MockTpsLimitStrategy struct {
+	ctrl     *gomock.Controller
+	recorder *MockTpsLimitStrategyMockRecorder
+}
+
+// MockTpsLimitStrategyMockRecorder is the mock recorder for MockTpsLimitStrategy
+type MockTpsLimitStrategyMockRecorder struct {
+	mock *MockTpsLimitStrategy
+}
+
+// NewMockTpsLimitStrategy creates a new mock instance
+func NewMockTpsLimitStrategy(ctrl *gomock.Controller) *MockTpsLimitStrategy {
+	mock := &MockTpsLimitStrategy{ctrl: ctrl}
+	mock.recorder = &MockTpsLimitStrategyMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockTpsLimitStrategy) EXPECT() *MockTpsLimitStrategyMockRecorder {
+	return m.recorder
+}
+
+// IsAllowable mocks base method
+func (m *MockTpsLimitStrategy) IsAllowable() bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsAllowable")
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// IsAllowable indicates an expected call of IsAllowable
+func (mr *MockTpsLimitStrategyMockRecorder) IsAllowable() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsAllowable", reflect.TypeOf((*MockTpsLimitStrategy)(nil).IsAllowable))
+}

filter/impl/tps/impl/tps_limit_thread_safe_fix_window_strategy.go

+/*
+ * 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 impl
+
+import (
+	"sync"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+)
+
+func init() {
+	extension.SetTpsLimitStrategy("threadSafeFixedWindow", NewThreadSafeFixedWindowTpsLimitStrategyImpl)
+}
+
+/**
+ * it's the thread-safe implementation.
+ * Also, it's a thread-safe decorator of FixedWindowTpsLimitStrategyImpl
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" # the name of limiter
+ *   tps.limit.strategy: "threadSafeFixedWindow" # service-level
+ *   methods:
+ *    - name: "GetUser"
+ *      tps.interval: 3000
+ *      tps.limit.strategy: "threadSafeFixedWindow" # method-level
+ */
+type ThreadSafeFixedWindowTpsLimitStrategyImpl struct {
+	mutex       *sync.Mutex
+	fixedWindow *FixedWindowTpsLimitStrategyImpl
+}
+
+func (impl *ThreadSafeFixedWindowTpsLimitStrategyImpl) IsAllowable() bool {
+	impl.mutex.Lock()
+	defer impl.mutex.Unlock()
+	return impl.fixedWindow.IsAllowable()
+}
+
+func NewThreadSafeFixedWindowTpsLimitStrategyImpl(rate int, interval int) tps.TpsLimitStrategy {
+	fixedWindowStrategy := NewFixedWindowTpsLimitStrategyImpl(rate, interval).(*FixedWindowTpsLimitStrategyImpl)
+	return &ThreadSafeFixedWindowTpsLimitStrategyImpl{
+		fixedWindow: fixedWindowStrategy,
+		mutex:       &sync.Mutex{},
+	}
+}

filter/impl/tps/impl/tps_limit_thread_safe_fix_window_strategy_test.go

+/*
+ * 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 impl
+
+import (
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+func TestThreadSafeFixedWindowTpsLimitStrategyImpl_IsAllowable(t *testing.T) {
+	strategy := NewThreadSafeFixedWindowTpsLimitStrategyImpl(2, 60000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+
+	strategy = NewThreadSafeFixedWindowTpsLimitStrategyImpl(2, 2000)
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+	time.Sleep(time.Duration(2100 * 1000))
+	assert.True(t, strategy.IsAllowable())
+	assert.True(t, strategy.IsAllowable())
+	assert.False(t, strategy.IsAllowable())
+}

filter/impl/tps/impl/tps_limiter_method_service.go

+/*
+ * 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 impl
+
+import (
+	"fmt"
+	"strconv"
+	"sync"
+)
+
+import (
+	"github.com/modern-go/concurrent"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const name = "method-service"
+
+func init() {
+	extension.SetTpsLimiter(constant.DEFAULT_KEY, GetMethodServiceTpsLimiter)
+	extension.SetTpsLimiter(name, GetMethodServiceTpsLimiter)
+}
+
+/**
+ * This implementation allows developer to config both method-level and service-level tps limiter.
+ * for example:
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" or "default" # the name of MethodServiceTpsLimiterImpl. It's the default limiter too.
+ *   tps.limit.interval: 5000 # interval, the time unit is ms
+ *   tps.limit.rate: 300 # the max value in the interval. <0 means that the service will not be limited.
+ *   methods:
+ *    - name: "GetUser"
+ *      tps.interval: 3000
+ *      tps.limit.rate: 20, # in this case, this configuration in service-level will be ignored.
+ *    - name: "UpdateUser"
+ *      tps.limit.rate: -1, # If the rate<0, the method will be ignored
+ *
+ *
+ * More examples:
+ * case1:
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" or "default" # the name of MethodServiceTpsLimiterImpl. It's the default limiter too.
+ *   tps.limit.interval: 5000 # interval, the time unit is ms
+ *   tps.limit.rate: 300 # the max value in the interval. <0 means that the service will not be limited.
+ *   methods:
+ *    - name: "GetUser"
+ *    - name: "UpdateUser"
+ *      tps.limit.rate: -1,
+ * in this case, the method UpdateUser will be ignored,
+ * which means that only GetUser will be limited by service-level configuration.
+ *
+ * case2:
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" or "default" # the name of MethodServiceTpsLimiterImpl. It's the default limiter too.
+ *   tps.limit.interval: 5000 # interval, the time unit is ms
+ *   tps.limit.rate: 300 # the max value in the interval. <0 means that the service will not be limited.
+ *   methods:
+ *    - name: "GetUser"
+ *    - name: "UpdateUser"
+ *      tps.limit.rate: 30,
+ * In this case, the GetUser will be limited by service-level configuration(300 times in 5000ms),
+ * but UpdateUser will be limited by its configuration (30 times in 60000ms)
+ *
+ * case3:
+ * "UserProvider":
+ *   registry: "hangzhouzk"
+ *   protocol : "dubbo"
+ *   interface : "com.ikurento.user.UserProvider"
+ *   ... # other configuration
+ *   tps.limiter: "method-service" or "default" # the name of MethodServiceTpsLimiterImpl. It's the default limiter too.
+ *   methods:
+ *    - name: "GetUser"
+ *    - name: "UpdateUser"
+ *      tps.limit.rate: 70,
+ *      tps.limit.interval: 40000
+ * In this case, only UpdateUser will be limited by its configuration (70 times in 40000ms)
+ */
+type MethodServiceTpsLimiterImpl struct {
+	tpsState *concurrent.Map
+}
+
+func (limiter MethodServiceTpsLimiterImpl) IsAllowable(url common.URL, invocation protocol.Invocation) bool {
+
+	methodConfigPrefix := "methods." + invocation.MethodName() + "."
+
+	methodLimitRateConfig := url.GetParam(methodConfigPrefix+constant.TPS_LIMIT_RATE_KEY, "")
+	methodIntervalConfig := url.GetParam(methodConfigPrefix+constant.TPS_LIMIT_INTERVAL_KEY, "")
+
+	limitTarget := url.ServiceKey()
+
+	// method-level tps limit
+	if len(methodIntervalConfig) > 0 || len(methodLimitRateConfig) > 0 {
+		limitTarget = limitTarget + "#" + invocation.MethodName()
+	}
+
+	limitState, found := limiter.tpsState.Load(limitTarget)
+	if found {
+		return limitState.(tps.TpsLimitStrategy).IsAllowable()
+	}
+
+	limitRate := getLimitConfig(methodLimitRateConfig, url, invocation,
+		constant.TPS_LIMIT_RATE_KEY,
+		constant.DEFAULT_TPS_LIMIT_RATE)
+
+	if limitRate < 0 {
+		return true
+	}
+
+	limitInterval := getLimitConfig(methodIntervalConfig, url, invocation,
+		constant.TPS_LIMIT_INTERVAL_KEY,
+		constant.DEFAULT_TPS_LIMIT_INTERVAL)
+	if limitInterval <= 0 {
+		panic(fmt.Sprintf("The interval must be positive, please check your configuration! url: %s", url.String()))
+	}
+
+	limitStrategyConfig := url.GetParam(methodConfigPrefix+constant.TPS_LIMIT_STRATEGY_KEY,
+		url.GetParam(constant.TPS_LIMIT_STRATEGY_KEY, constant.DEFAULT_KEY))
+	limitStateCreator := extension.GetTpsLimitStrategyCreator(limitStrategyConfig)
+	limitState, _ = limiter.tpsState.LoadOrStore(limitTarget, limitStateCreator(int(limitRate), int(limitInterval)))
+	return limitState.(tps.TpsLimitStrategy).IsAllowable()
+}
+
+func getLimitConfig(methodLevelConfig string,
+	url common.URL,
+	invocation protocol.Invocation,
+	configKey string,
+	defaultVal string) int64 {
+
+	if len(methodLevelConfig) > 0 {
+		result, err := strconv.ParseInt(methodLevelConfig, 0, 0)
+		if err != nil {
+			panic(fmt.Sprintf("The %s for invocation %s # %s must be positive, please check your configuration!",
+				configKey, url.ServiceKey(), invocation.MethodName()))
+		}
+		return result
+	}
+
+	result, err := strconv.ParseInt(url.GetParam(configKey, defaultVal), 0, 0)
+
+	if err != nil {
+		panic(fmt.Sprintf("Cannot parse the configuration %s, please check your configuration!", configKey))
+	}
+	return result
+}
+
+var methodServiceTpsLimiterInstance *MethodServiceTpsLimiterImpl
+var methodServiceTpsLimiterOnce sync.Once
+
+func GetMethodServiceTpsLimiter() tps.TpsLimiter {
+	methodServiceTpsLimiterOnce.Do(func() {
+		methodServiceTpsLimiterInstance = &MethodServiceTpsLimiterImpl{
+			tpsState: concurrent.NewMap(),
+		}
+	})
+	return methodServiceTpsLimiterInstance
+}

filter/impl/tps/impl/tps_limiter_method_service_test.go

+/*
+ * 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 impl
+
+import (
+	"net/url"
+	"testing"
+)
+import (
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter/impl/tps"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestMethodServiceTpsLimiterImpl_IsAllowable_Only_Service_Level(t *testing.T) {
+	methodName := "hello"
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.INTERFACE_KEY, methodName),
+		common.WithParamsValue(constant.TPS_LIMIT_RATE_KEY, "20"))
+
+	mockStrategyImpl := NewMockTpsLimitStrategy(ctrl)
+	mockStrategyImpl.EXPECT().IsAllowable().Return(true).Times(1)
+	extension.SetTpsLimitStrategy(constant.DEFAULT_KEY, func(rate int, interval int) tps.TpsLimitStrategy {
+		assert.Equal(t, 20, rate)
+		assert.Equal(t, 60000, interval)
+		return mockStrategyImpl
+	})
+
+	limiter := GetMethodServiceTpsLimiter()
+	result := limiter.IsAllowable(*invokeUrl, invoc)
+	assert.True(t, result)
+}
+
+func TestMethodServiceTpsLimiterImpl_IsAllowable_No_Config(t *testing.T) {
+	methodName := "hello1"
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	// ctrl := gomock.NewController(t)
+	// defer ctrl.Finish()
+
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.INTERFACE_KEY, methodName),
+		common.WithParamsValue(constant.TPS_LIMIT_RATE_KEY, ""))
+
+	limiter := GetMethodServiceTpsLimiter()
+	result := limiter.IsAllowable(*invokeUrl, invoc)
+	assert.True(t, result)
+}
+
+func TestMethodServiceTpsLimiterImpl_IsAllowable_Method_Level_Override(t *testing.T) {
+	methodName := "hello2"
+	methodConfigPrefix := "methods." + methodName + "."
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.INTERFACE_KEY, methodName),
+		common.WithParamsValue(constant.TPS_LIMIT_RATE_KEY, "20"),
+		common.WithParamsValue(constant.TPS_LIMIT_INTERVAL_KEY, "3000"),
+		common.WithParamsValue(constant.TPS_LIMIT_STRATEGY_KEY, "invalid"),
+		common.WithParamsValue(methodConfigPrefix+constant.TPS_LIMIT_RATE_KEY, "40"),
+		common.WithParamsValue(methodConfigPrefix+constant.TPS_LIMIT_INTERVAL_KEY, "7000"),
+		common.WithParamsValue(methodConfigPrefix+constant.TPS_LIMIT_STRATEGY_KEY, "default"),
+	)
+
+	mockStrategyImpl := NewMockTpsLimitStrategy(ctrl)
+	mockStrategyImpl.EXPECT().IsAllowable().Return(true).Times(1)
+	extension.SetTpsLimitStrategy(constant.DEFAULT_KEY, func(rate int, interval int) tps.TpsLimitStrategy {
+		assert.Equal(t, 40, rate)
+		assert.Equal(t, 7000, interval)
+		return mockStrategyImpl
+	})
+
+	limiter := GetMethodServiceTpsLimiter()
+	result := limiter.IsAllowable(*invokeUrl, invoc)
+	assert.True(t, result)
+}
+
+func TestMethodServiceTpsLimiterImpl_IsAllowable_Both_Method_And_Service(t *testing.T) {
+	methodName := "hello3"
+	methodConfigPrefix := "methods." + methodName + "."
+	invoc := invocation.NewRPCInvocation(methodName, []interface{}{"OK"}, make(map[string]string, 0))
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.INTERFACE_KEY, methodName),
+		common.WithParamsValue(constant.TPS_LIMIT_RATE_KEY, "20"),
+		common.WithParamsValue(constant.TPS_LIMIT_INTERVAL_KEY, "3000"),
+		common.WithParamsValue(methodConfigPrefix+constant.TPS_LIMIT_RATE_KEY, "40"),
+	)
+
+	mockStrategyImpl := NewMockTpsLimitStrategy(ctrl)
+	mockStrategyImpl.EXPECT().IsAllowable().Return(true).Times(1)
+	extension.SetTpsLimitStrategy(constant.DEFAULT_KEY, func(rate int, interval int) tps.TpsLimitStrategy {
+		assert.Equal(t, 40, rate)
+		assert.Equal(t, 3000, interval)
+		return mockStrategyImpl
+	})
+
+	limiter := GetMethodServiceTpsLimiter()
+	result := limiter.IsAllowable(*invokeUrl, invoc)
+	assert.True(t, result)
+}

filter/impl/tps/impl/tps_limiter_mock.go

+//  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.
+//
+
+// Code generated by MockGen. DO NOT EDIT.
+// Source: tps_limiter.go
+
+// Package filter is a generated GoMock package.
+package impl
+
+import (
+	common "github.com/apache/dubbo-go/common"
+	protocol "github.com/apache/dubbo-go/protocol"
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+)
+
+// MockTpsLimiter is a mock of TpsLimiter interface
+type MockTpsLimiter struct {
+	ctrl     *gomock.Controller
+	recorder *MockTpsLimiterMockRecorder
+}
+
+// MockTpsLimiterMockRecorder is the mock recorder for MockTpsLimiter
+type MockTpsLimiterMockRecorder struct {
+	mock *MockTpsLimiter
+}
+
+// NewMockTpsLimiter creates a new mock instance
+func NewMockTpsLimiter(ctrl *gomock.Controller) *MockTpsLimiter {
+	mock := &MockTpsLimiter{ctrl: ctrl}
+	mock.recorder = &MockTpsLimiterMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockTpsLimiter) EXPECT() *MockTpsLimiterMockRecorder {
+	return m.recorder
+}
+
+// IsAllowable mocks base method
+func (m *MockTpsLimiter) IsAllowable(arg0 common.URL, arg1 protocol.Invocation) bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsAllowable", arg0, arg1)
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// IsAllowable indicates an expected call of IsAllowable
+func (mr *MockTpsLimiterMockRecorder) IsAllowable(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsAllowable", reflect.TypeOf((*MockTpsLimiter)(nil).IsAllowable), arg0, arg1)
+}