Add UT

config/graceful_shutdown.go

@@ -21,10 +21,11 @@ import (
 	"os"
 	"os/signal"
 	"runtime/debug"
-	"sync"
 	"syscall"
 	"time"
+)
 
+import (
 	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/common/logger"
@@ -45,7 +46,6 @@ import (
  * So this signal will be ignored.
  *
  */
-var gracefulShutdownOnce = sync.Once{}
 
 func GracefulShutdownInit() {
 

config/graceful_shutdown_config.go

@@ -20,10 +20,10 @@ package config
 import (
 	"strconv"
 	"time"
-
+)
+import (
 	"github.com/apache/dubbo-go/common/logger"
 )
-
 const (
 	defaultTimeout  = 60 * time.Second
 	defaultStepTimeout = 10 * time.Second

filter/impl/GracefulShutdownFilter.go → filter/impl/graceful_shutdown_filter.go

@@ -19,7 +19,9 @@ package impl
 
 import (
 	"sync/atomic"
+)
 
+import (
 	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/common/logger"
@@ -34,7 +36,9 @@ func init() {
 		activeCount:    0,
 		shutdownConfig: config.GetConsumerConfig().ShutdownConfig,
 	}
-	var providerFilter = &gracefulShutdownFilter{activeCount: 0}
+	var providerFilter = &gracefulShutdownFilter{activeCount: 0,
+		shutdownConfig: config.GetProviderConfig().ShutdownConfig,
+	}
 
 	extension.SetFilter(constant.CONSUMER_SHUTDOWN_FILTER, func() filter.Filter {
 		return consumerFiler
@@ -62,7 +66,7 @@ func (gf *gracefulShutdownFilter) Invoke(invoker protocol.Invoker, invocation pr
 func (gf *gracefulShutdownFilter) OnResponse(result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
 	atomic.AddInt32(&gf.activeCount, -1)
 	// although this isn't thread safe, it won't be a problem if the gf.rejectNewRequest() is true.
-	if gf.activeCount <= 0 {
+	if gf.shutdownConfig != nil && gf.activeCount <= 0 {
 		gf.shutdownConfig.RequestsFinished = true
 	}
 	return result

filter/impl/graceful_shutdown_filter_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/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/config"
+	filterCommon "github.com/apache/dubbo-go/filter/common"
+	"github.com/apache/dubbo-go/filter/common/impl"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestGenericFilter_Invoke(t *testing.T) {
+	invoc := invocation.NewRPCInvocation("GetUser", []interface{}{"OK"}, make(map[string]string, 0))
+
+	invokeUrl := common.NewURLWithOptions(
+		common.WithParams(url.Values{}), )
+
+	shutdownFilter := extension.GetFilter(constant.PROVIDER_SHUTDOWN_FILTER).(*gracefulShutdownFilter)
+
+	providerConfig := config.GetProviderConfig()
+
+	assert.False(t, shutdownFilter.rejectNewRequest())
+	assert.Nil(t, providerConfig.ShutdownConfig)
+
+	assert.Equal(t, extension.GetRejectedExecutionHandler(constant.DEFAULT_KEY),
+		shutdownFilter.getRejectHandler())
+
+	result := shutdownFilter.Invoke(protocol.NewBaseInvoker(*invokeUrl), invoc)
+	assert.NotNil(t, result)
+	assert.Nil(t, result.Error())
+
+	providerConfig.ShutdownConfig = &config.ShutdownConfig{
+		RejectRequest:        true,
+		RejectRequestHandler: "mock",
+	}
+	shutdownFilter.shutdownConfig = providerConfig.ShutdownConfig
+
+	assert.True(t, shutdownFilter.rejectNewRequest())
+	result = shutdownFilter.OnResponse(nil, protocol.NewBaseInvoker(*invokeUrl), invoc)
+
+	rejectHandler := &impl.OnlyLogRejectedExecutionHandler{}
+	extension.SetRejectedExecutionHandler("mock", func() filterCommon.RejectedExecutionHandler {
+		return rejectHandler
+	})
+	assert.True(t, providerConfig.ShutdownConfig.RequestsFinished)
+	assert.Equal(t, rejectHandler, shutdownFilter.getRejectHandler())
+
+}

protocol/rpc_status.go

@@ -73,17 +73,3 @@ func beginCount0(rpcStatus *RpcStatus) {
 func endCount0(rpcStatus *RpcStatus) {
 	atomic.AddInt32(&rpcStatus.active, -1)
 }
-
-func GetTotalActive() int32 {
-	var result int32 = 0
-	methodStatistics.Range(func(_, value interface{}) bool {
-		statics := value.(*sync.Map)
-		statics.Range(func(_, value interface{}) bool {
-			rpcStatus := value.(*RpcStatus)
-			result = result + rpcStatus.active
-			return true
-		})
-		return true
-	})
-	return result
-}