chore: 封装通用http.server 启动/停止 函数

internal/biz/task/dash_initialize.go

@@ -34,8 +34,7 @@ func (i *DashInitialize) Identifier() string {
 	return "initialize"
 }
 
-func (i *DashInitialize) Run() error {
-	ctx := context.Background()
+func (i *DashInitialize) Run(ctx context.Context) error {
 	key := "did_init"
 	return dao.WithTxEx(ctx, i.db.Client, func(ctx context.Context, client *ent.Client) error {
 		exist, err := client.KeyValueStore.Query().Where(keyvaluestore.KeyEQ(key)).Exist(ctx)

internal/server/api/web.go

@@ -11,10 +11,12 @@ import (
 	"github.com/tbxark/sphere/pkg/log/logfields"
 	"github.com/tbxark/sphere/pkg/server/auth/authorizer"
 	"github.com/tbxark/sphere/pkg/server/auth/jwtauth"
+	"github.com/tbxark/sphere/pkg/server/ginx"
 	"github.com/tbxark/sphere/pkg/server/middleware/auth"
 	"github.com/tbxark/sphere/pkg/server/middleware/logger"
 	"github.com/tbxark/sphere/pkg/server/route/cors"
 	"net/http"
+	"time"
 )
 
 type Web struct {
@@ -34,7 +36,7 @@ func (w *Web) Identifier() string {
 	return "api"
 }
 
-func (w *Web) Run() error {
+func (w *Web) Run(ctx context.Context) error {
 	jwtAuthorizer := jwtauth.NewJwtAuth[authorizer.RBACClaims[int64]](w.config.JWT)
 
 	zapLogger := log.ZapLogger().With(logfields.String("module", "api"))
@@ -65,16 +67,9 @@ func (w *Web) Run() error {
 		Addr:    w.config.HTTP.Address,
 		Handler: engine.Handler(),
 	}
-	return w.server.ListenAndServe()
+	return ginx.Start(ctx, w.server, 30*time.Second)
 }
 
 func (w *Web) Close(ctx context.Context) error {
-	if w.server != nil {
-		err := w.server.Close()
-		if err != nil {
-			return err
-		}
-		w.server = nil
-	}
-	return nil
+	return ginx.Close(ctx, w.server)
 }

internal/server/bot/bot.go

@@ -28,8 +28,8 @@ func (b *Bot) Identifier() string {
 	return "bot"
 }
 
-func (b *Bot) Run() error {
-	return b.Bot.Run(func(bot *bot.Bot) error {
+func (b *Bot) Run(ctx context.Context) error {
+	return b.Bot.Run(ctx, func(bot *bot.Bot) error {
 
 		sfMid := telegram.NewSingleFlightMiddleware()
 

internal/server/dash/web.go

@@ -41,7 +41,7 @@ func (w *Web) Identifier() string {
 	return "dash"
 }
 
-func (w *Web) Run() error {
+func (w *Web) Run(ctx context.Context) error {
 
 	jwtAuthorizer := jwtauth.NewJwtAuth[authorizer.RBACClaims[int64]](w.config.AuthJWT)
 	jwtRefresher := jwtauth.NewJwtAuth[authorizer.RBACClaims[int64]](w.config.RefreshJWT)
@@ -109,18 +109,11 @@ func (w *Web) Run() error {
 		Addr:    w.config.HTTP.Address,
 		Handler: engine.Handler(),
 	}
-	return w.server.ListenAndServe()
+	return ginx.Start(ctx, w.server, 30*time.Second)
 }
 
 func (w *Web) Close(ctx context.Context) error {
-	if w.server != nil {
-		err := w.server.Close()
-		if err != nil {
-			return err
-		}
-		w.server = nil
-	}
-	return nil
+	return ginx.Close(ctx, w.server)
 }
 
 func (w *Web) withPermission(resource string) gin.HandlerFunc {

pkg/server/ginx/server.go

@@ -0,0 +1,50 @@
+package ginx
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/tbxark/sphere/pkg/log"
+	"github.com/tbxark/sphere/pkg/log/logfields"
+	"net/http"
+	"time"
+)
+
+func Start(ctx context.Context, server *http.Server, closeTimeout time.Duration) error {
+	errChan := make(chan error, 1)
+	closeChan := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), closeTimeout)
+			defer cancel()
+			if err := server.Shutdown(shutdownCtx); err != nil {
+				log.Errorw("close server error", logfields.Error(err))
+				errChan <- err
+			}
+		case <-closeChan: // 防止goroutine泄露
+			break
+		}
+	}()
+	defer close(closeChan)
+	if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
+		return fmt.Errorf("server error: %w", err)
+	}
+	select {
+	case err := <-errChan:
+		return err
+	default:
+		return nil
+	}
+}
+
+func Close(ctx context.Context, server *http.Server) error {
+	if server == nil {
+		return nil
+	}
+	err := server.Shutdown(ctx)
+	if err != nil {
+		return err
+	}
+	return nil
+}

pkg/server/service/docs/web.go

@@ -5,13 +5,15 @@ import (
 	"fmt"
 	"github.com/gin-gonic/gin"
 	"github.com/swaggo/swag"
+	"github.com/tbxark/sphere/pkg/server/ginx"
 	"github.com/tbxark/sphere/pkg/server/route/cors"
 	"github.com/tbxark/sphere/pkg/server/route/docs"
 	"html/template"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
 	"strings"
+	"time"
 )
 
 type Target struct {
@@ -39,7 +41,7 @@ func (w *Web) Identifier() string {
 	return "docs"
 }
 
-func (w *Web) Run() error {
+func (w *Web) Run(ctx context.Context) error {
 	engine := gin.Default()
 	cors.Setup(engine, []string{"*"})
 
@@ -57,18 +59,11 @@ func (w *Web) Run() error {
 		Addr:    w.config.Address,
 		Handler: engine.Handler(),
 	}
-	return w.server.ListenAndServe()
+	return ginx.Start(ctx, w.server, 30*time.Second)
 }
 
 func (w *Web) Close(ctx context.Context) error {
-	if w.server != nil {
-		err := w.server.Close()
-		if err != nil {
-			return err
-		}
-		w.server = nil
-	}
-	return nil
+	return ginx.Close(ctx, w.server)
 }
 
 func setup(spec *swag.Spec, router gin.IRouter, target string) error {

pkg/telegram/bot.go

@@ -5,8 +5,6 @@ import (
 	"github.com/go-telegram/bot"
 	"github.com/go-telegram/bot/models"
 	"github.com/tbxark/sphere/pkg/log"
-	"os"
-	"os/signal"
 )
 
 type Config struct {
@@ -25,10 +23,7 @@ func NewApp(config *Config) *Bot {
 	}
 }
 
-func (b *Bot) Run(options ...func(*bot.Bot) error) error {
-	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
-	defer cancel()
-
+func (b *Bot) Run(ctx context.Context, options ...func(*bot.Bot) error) error {
 	opts := []bot.Option{
 		bot.WithSkipGetMe(),
 		bot.WithErrorsHandler(func(err error) {
@@ -77,14 +72,12 @@ func (b *Bot) Run(options ...func(*bot.Bot) error) error {
 }
 
 func (b *Bot) Close(ctx context.Context) error {
-	if b.bot != nil {
-		_, err := b.bot.Close(ctx)
-		if err != nil {
-			return err
-		}
-		b.bot = nil
+	if b.bot == nil {
+		return nil
 	}
-	return nil
+	_, err := b.bot.Close(ctx)
+	b.bot = nil
+	return err
 }
 
 func (b *Bot) BindCommand(command string, handlerFunc HandlerFunc, middleware ...bot.Middleware) {

pkg/utils/boot/app.go

@@ -9,7 +9,7 @@ import (
 
 type Runnable interface {
 	Identifier() string
-	Run() error
+	Run(ctx context.Context) error
 }
 
 type Closeable interface {
@@ -30,34 +30,21 @@ func NewApplication(tasks []Runnable, cleaners []Closeable) *Application {
 }
 
 func (a *Application) Run(ctx context.Context) error {
-	wg, errCtx := errgroup.WithContext(ctx)
+	wg, ctx := errgroup.WithContext(ctx)
 	for _, item := range a.runner {
 		log.Infof("runner %s start", item.Identifier())
 		runner := item
 		wg.Go(func() error {
-			done := make(chan struct{})
-			defer close(done)
-			go func() {
-				select {
-				case <-errCtx.Done():
-					log.Infof("runner %s stopping due to context cancellation", runner.Identifier())
-					if stoppable, ok := runner.(Closeable); ok {
-						_ = stoppable.Close(context.Background())
-					}
-				case <-done:
-					// runner finished and closed this goroutine
-					return
-				}
-			}()
 			defer func() {
 				if r := recover(); r != nil {
 					log.Errorw(
 						"runner panic",
 						logfields.String("runner", runner.Identifier()),
+						logfields.Any("recover", r),
 					)
 				}
 			}()
-			if err := runner.Run(); err != nil {
+			if err := runner.Run(ctx); err != nil {
 				log.Errorw(
 					"runner error",
 					logfields.String("runner", runner.Identifier()),
@@ -74,14 +61,15 @@ func (a *Application) Run(ctx context.Context) error {
 func (a *Application) Close(ctx context.Context) error {
 	wg := errgroup.Group{}
 	for _, item := range a.closer {
-		log.Infof("closer %s start", item)
+		log.Infof("closer %s start", item.Identifier())
 		closer := item
 		wg.Go(func() error {
 			defer func() {
 				if r := recover(); r != nil {
 					log.Errorw(
 						"closer panic",
 						logfields.String("closer", closer.Identifier()),
+						logfields.Any("recover", r),
 					)
 				}
 			}()