mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-12 18:26:36 +08:00
merge: replace init registration with bootstrap wiring
This commit is contained in:
@@ -15,8 +15,9 @@ import (
|
|||||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||||
)
|
)
|
||||||
|
|
||||||
func init() {
|
// RegisterTaskListeners subscribes push notification handlers to task completion events.
|
||||||
task.OnTaskCompleted = handleTaskCompleted
|
func RegisterTaskListeners() {
|
||||||
|
task.OnTaskCompleted(handleTaskCompleted)
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleTaskCompleted handles task completions and triggers appropriate push events.
|
// handleTaskCompleted handles task completions and triggers appropriate push events.
|
||||||
|
|||||||
@@ -13,7 +13,6 @@ import ("fmt"
|
|||||||
"github.com/Rain-kl/Wavelet/internal/apps/admin"
|
"github.com/Rain-kl/Wavelet/internal/apps/admin"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
"github.com/Rain-kl/Wavelet/internal/task"
|
"github.com/Rain-kl/Wavelet/internal/task"
|
||||||
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/task/scheduler"
|
"github.com/Rain-kl/Wavelet/internal/task/scheduler"
|
||||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
@@ -21,10 +20,6 @@ import ("fmt"
|
|||||||
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/common/response")
|
"github.com/Rain-kl/Wavelet/internal/common/response")
|
||||||
|
|
||||||
func init() {
|
|
||||||
taskhandlers.Register()
|
|
||||||
}
|
|
||||||
|
|
||||||
// ListTaskTypes 获取支持的任务类型列表
|
// ListTaskTypes 获取支持的任务类型列表
|
||||||
// @Summary 获取支持的任务类型
|
// @Summary 获取支持的任务类型
|
||||||
// @Description 返回系统支持的所有可调度任务类型列表,包括任务名称、描述、是否支持时间范围等元数据,需要管理员权限
|
// @Description 返回系统支持的所有可调度任务类型列表,包括任务名称、描述、是否支持时间范围等元数据,需要管理员权限
|
||||||
|
|||||||
@@ -0,0 +1,65 @@
|
|||||||
|
// Copyright 2026 Arctel.net
|
||||||
|
// SPDX-License-Identifier: Apache-2.0
|
||||||
|
|
||||||
|
// Package bootstrap wires cross-module integrations at the application composition root.
|
||||||
|
// All registrations use sync.Once so entry points can call them safely without import-order side effects.
|
||||||
|
package bootstrap
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
|
||||||
|
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
|
||||||
|
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
registerTasksOnce sync.Once
|
||||||
|
registerPushDomainEventsOnce sync.Once
|
||||||
|
registerTaskListenersOnce sync.Once
|
||||||
|
)
|
||||||
|
|
||||||
|
// RegisterTasks registers all built-in task handlers and metadata.
|
||||||
|
func RegisterTasks() {
|
||||||
|
registerTasksOnce.Do(func() {
|
||||||
|
taskhandlers.Register()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterPushDomainEvents wires push notification handlers for domain events.
|
||||||
|
func RegisterPushDomainEvents() {
|
||||||
|
registerPushDomainEventsOnce.Do(func() {
|
||||||
|
custom_events.Register()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterTaskListeners wires operational listeners to task framework hooks.
|
||||||
|
func RegisterTaskListeners() {
|
||||||
|
registerTaskListenersOnce.Do(func() {
|
||||||
|
admin_push.RegisterTaskListeners()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterAPI wires integrations required by the HTTP API process.
|
||||||
|
func RegisterAPI() {
|
||||||
|
RegisterTasks()
|
||||||
|
RegisterPushDomainEvents()
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterWorker wires integrations required by the task worker process.
|
||||||
|
func RegisterWorker() {
|
||||||
|
RegisterTasks()
|
||||||
|
RegisterTaskListeners()
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterScheduler wires integrations required by the task scheduler process.
|
||||||
|
func RegisterScheduler() {
|
||||||
|
RegisterTasks()
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterAll wires integrations for fused mode (API + Worker + Scheduler).
|
||||||
|
func RegisterAll() {
|
||||||
|
RegisterTasks()
|
||||||
|
RegisterPushDomainEvents()
|
||||||
|
RegisterTaskListeners()
|
||||||
|
}
|
||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/bootstrap"
|
||||||
"github.com/Rain-kl/Wavelet/internal/router"
|
"github.com/Rain-kl/Wavelet/internal/router"
|
||||||
"github.com/Rain-kl/Wavelet/internal/task/scheduler"
|
"github.com/Rain-kl/Wavelet/internal/task/scheduler"
|
||||||
"github.com/Rain-kl/Wavelet/internal/task/worker"
|
"github.com/Rain-kl/Wavelet/internal/task/worker"
|
||||||
@@ -20,6 +21,7 @@ var allCmd = &cobra.Command{
|
|||||||
Short: "以融合模式同时启动 API、Worker 和 Scheduler",
|
Short: "以融合模式同时启动 API、Worker 和 Scheduler",
|
||||||
Run: func(_ *cobra.Command, _ []string) {
|
Run: func(_ *cobra.Command, _ []string) {
|
||||||
log.Println("[All] 融合模式启动")
|
log.Println("[All] 融合模式启动")
|
||||||
|
bootstrap.RegisterAll()
|
||||||
|
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
|
|||||||
@@ -16,8 +16,8 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
|
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
|
||||||
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
|
|
||||||
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
|
"github.com/Rain-kl/Wavelet/internal/apps/risk_control"
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/bootstrap"
|
||||||
router_root "github.com/Rain-kl/Wavelet/internal/router/root"
|
router_root "github.com/Rain-kl/Wavelet/internal/router/root"
|
||||||
v1 "github.com/Rain-kl/Wavelet/internal/router/v1"
|
v1 "github.com/Rain-kl/Wavelet/internal/router/v1"
|
||||||
|
|
||||||
@@ -40,8 +40,8 @@ func Serve() {
|
|||||||
// 初始化 ClickHouse 异步日志写入器
|
// 初始化 ClickHouse 异步日志写入器
|
||||||
risk_control.InitLogWriter()
|
risk_control.InitLogWriter()
|
||||||
|
|
||||||
// 装配推送模块与领域事件的集成(组合根显式注册,避免 init 副作用)
|
// 组合根显式装配跨模块集成(避免 init 副作用与 import 顺序依赖)
|
||||||
custom_events.Register()
|
bootstrap.RegisterAPI()
|
||||||
|
|
||||||
// 运行内置事件同步
|
// 运行内置事件同步
|
||||||
if err := admin_push.SyncEvents(context.Background()); err != nil {
|
if err := admin_push.SyncEvents(context.Background()); err != nil {
|
||||||
|
|||||||
@@ -26,8 +26,16 @@ import (
|
|||||||
// handlerRegistry 已注册的任务处理器
|
// handlerRegistry 已注册的任务处理器
|
||||||
var handlerRegistry = make(map[string]TaskHandler)
|
var handlerRegistry = make(map[string]TaskHandler)
|
||||||
|
|
||||||
// OnTaskCompleted is a hook called when a task execution completes.
|
// CompletedHandler is called when a task execution completes.
|
||||||
var OnTaskCompleted func(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error)
|
type CompletedHandler func(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error)
|
||||||
|
|
||||||
|
var taskCompletedHandlers []CompletedHandler
|
||||||
|
|
||||||
|
// OnTaskCompleted registers a handler for task completion events.
|
||||||
|
// Handlers must be registered during application bootstrap before processing tasks.
|
||||||
|
func OnTaskCompleted(handler CompletedHandler) {
|
||||||
|
taskCompletedHandlers = append(taskCompletedHandlers, handler)
|
||||||
|
}
|
||||||
|
|
||||||
// RegisterHandler 注册任务处理器
|
// RegisterHandler 注册任务处理器
|
||||||
// 传入任务类型标识(对应 constants.go 中的 AsynqTask 常量)和 TaskHandler 实现
|
// 传入任务类型标识(对应 constants.go 中的 AsynqTask 常量)和 TaskHandler 实现
|
||||||
@@ -405,9 +413,17 @@ func completeTaskExecution(ctx context.Context, execution *model.TaskExecution,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if OnTaskCompleted != nil {
|
notifyTaskCompleted(ctx, execution, result, execErr)
|
||||||
asyncCtx := context.WithoutCancel(ctx)
|
}
|
||||||
go OnTaskCompleted(asyncCtx, execution, result, execErr)
|
|
||||||
|
func notifyTaskCompleted(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error) {
|
||||||
|
if len(taskCompletedHandlers) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
asyncCtx := context.WithoutCancel(ctx)
|
||||||
|
for _, handler := range taskCompletedHandlers {
|
||||||
|
go handler(asyncCtx, execution, result, execErr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import (
|
|||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/bootstrap"
|
||||||
"github.com/Rain-kl/Wavelet/internal/model"
|
"github.com/Rain-kl/Wavelet/internal/model"
|
||||||
"github.com/Rain-kl/Wavelet/internal/task"
|
"github.com/Rain-kl/Wavelet/internal/task"
|
||||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||||
@@ -32,6 +33,8 @@ func GetAsynqClient() *asynq.Client {
|
|||||||
|
|
||||||
// StartScheduler 启动调度器 (该函数阻塞,直到调度器退出)
|
// StartScheduler 启动调度器 (该函数阻塞,直到调度器退出)
|
||||||
func StartScheduler() error {
|
func StartScheduler() error {
|
||||||
|
bootstrap.RegisterScheduler()
|
||||||
|
|
||||||
var err error
|
var err error
|
||||||
schedulerOnce.Do(func() {
|
schedulerOnce.Do(func() {
|
||||||
quitChan = make(chan struct{})
|
quitChan = make(chan struct{})
|
||||||
|
|||||||
@@ -7,22 +7,18 @@ package worker
|
|||||||
import (
|
import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/Rain-kl/Wavelet/internal/bootstrap"
|
||||||
"github.com/Rain-kl/Wavelet/internal/config"
|
"github.com/Rain-kl/Wavelet/internal/config"
|
||||||
"github.com/Rain-kl/Wavelet/internal/task"
|
"github.com/Rain-kl/Wavelet/internal/task"
|
||||||
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
|
|
||||||
"github.com/hibiken/asynq"
|
"github.com/hibiken/asynq"
|
||||||
)
|
)
|
||||||
|
|
||||||
// workerShutdownTimeout Worker 优雅关闭超时时间
|
// workerShutdownTimeout Worker 优雅关闭超时时间
|
||||||
const workerShutdownTimeout = 3 * time.Minute
|
const workerShutdownTimeout = 3 * time.Minute
|
||||||
|
|
||||||
func init() {
|
|
||||||
// 注册所有任务处理器
|
|
||||||
taskhandlers.Register()
|
|
||||||
}
|
|
||||||
|
|
||||||
// StartWorker 启动任务处理服务器
|
// StartWorker 启动任务处理服务器
|
||||||
func StartWorker() error {
|
func StartWorker() error {
|
||||||
|
bootstrap.RegisterWorker()
|
||||||
asynqServer := asynq.NewServer(
|
asynqServer := asynq.NewServer(
|
||||||
task.RedisOpt,
|
task.RedisOpt,
|
||||||
asynq.Config{
|
asynq.Config{
|
||||||
|
|||||||
Reference in New Issue
Block a user