mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-06 23:56:37 +08:00
feat(core): bind request services and implement registered tasks
Wire plugin services through Bind/InjectFrom and AppContext so HTTP and workers resolve dependencies after Apply. Register TaskHandler objects with persisted results, and implement send_email_code, mail:send, cleanup_inactive_users, and dispatch_bot_msg.
This commit is contained in:
@@ -24,10 +24,8 @@ func setDBService(s contracts.DBService) {
|
||||
}
|
||||
|
||||
func getDB(ctx context.Context) *gorm.DB {
|
||||
if c, ok := ctx.(*core.Context); ok && c != nil {
|
||||
if s, err := core.Inject[contracts.DBService](c); err == nil && s != nil {
|
||||
return s.DB(ctx)
|
||||
}
|
||||
if s, err := core.InjectFrom[contracts.DBService](ctx); err == nil && s != nil {
|
||||
return s.DB(ctx)
|
||||
}
|
||||
dbMu.RLock()
|
||||
s := dbSvc
|
||||
|
||||
@@ -4,11 +4,14 @@
|
||||
package driver_inproc_worker
|
||||
|
||||
import (
|
||||
"Wavelet/core"
|
||||
"Wavelet/core/contracts"
|
||||
"Wavelet/core/extpoints"
|
||||
"Wavelet/pkg/idgen"
|
||||
"Wavelet/pkg/logger"
|
||||
"Wavelet/pkg/util"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
@@ -40,6 +43,7 @@ type InprocQueue struct {
|
||||
// baseCtx is the app-lifetime context captured at Start; task handlers
|
||||
// derive their timeouts from it so shutdown cancellation propagates.
|
||||
baseCtx context.Context
|
||||
appCtx *core.Context
|
||||
}
|
||||
|
||||
// NewInprocQueue creates a new InprocQueue with a given concurrency and queue capacity.
|
||||
@@ -182,10 +186,14 @@ func (q *InprocQueue) executeTask(ctx context.Context, msg TaskMessage) {
|
||||
|
||||
taskCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
if q.appCtx != nil {
|
||||
taskCtx = core.WithAppContext(taskCtx, q.appCtx)
|
||||
ctx = core.WithAppContext(ctx, q.appCtx)
|
||||
}
|
||||
|
||||
q.markRunning(ctx, msg)
|
||||
start := time.Now()
|
||||
err := invokeHandler(taskCtx, td.Handler, msg.Payload)
|
||||
result, err := invokeHandler(taskCtx, td.Handler, msg.Payload)
|
||||
duration := time.Since(start)
|
||||
|
||||
if err != nil {
|
||||
@@ -210,25 +218,29 @@ func (q *InprocQueue) executeTask(ctx context.Context, msg TaskMessage) {
|
||||
}
|
||||
return
|
||||
}
|
||||
q.succeedExecution(ctx, msg, duration)
|
||||
q.succeedExecution(ctx, msg, duration, result)
|
||||
}
|
||||
|
||||
func invokeHandler(ctx context.Context, handler any, payload []byte) error {
|
||||
func invokeHandler(ctx context.Context, handler any, payload []byte) (*contracts.TaskResultDTO, error) {
|
||||
if handler == nil {
|
||||
return errors.New("nil task handler")
|
||||
return nil, errors.New("nil task handler")
|
||||
}
|
||||
|
||||
switch fn := handler.(type) {
|
||||
case func(context.Context, []byte) error:
|
||||
case contracts.TaskHandler:
|
||||
return fn.Execute(ctx, payload)
|
||||
case func(context.Context, []byte) (*contracts.TaskResultDTO, error):
|
||||
return fn(ctx, payload)
|
||||
case func(context.Context, []byte) error:
|
||||
return nil, fn(ctx, payload)
|
||||
case func(context.Context) error:
|
||||
return fn(ctx)
|
||||
return nil, fn(ctx)
|
||||
case func([]byte) error:
|
||||
return fn(payload)
|
||||
return nil, fn(payload)
|
||||
case func() error:
|
||||
return fn()
|
||||
return nil, fn()
|
||||
default:
|
||||
return fmt.Errorf("unsupported handler type: %T", handler)
|
||||
return nil, fmt.Errorf("unsupported handler type: %T", handler)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -280,16 +292,27 @@ func (q *InprocQueue) markRunning(ctx context.Context, msg TaskMessage) {
|
||||
q.appendExecutionLog(ctx, msg.ID, fmt.Sprintf("[系统] 开始执行异步任务 [类型: %s]", msg.TaskType))
|
||||
}
|
||||
|
||||
func (q *InprocQueue) succeedExecution(ctx context.Context, msg TaskMessage, duration time.Duration) {
|
||||
func (q *InprocQueue) succeedExecution(ctx context.Context, msg TaskMessage, duration time.Duration, result *contracts.TaskResultDTO) {
|
||||
db := getDB(ctx)
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
resultText := "ok"
|
||||
if result != nil {
|
||||
resultText = result.Message
|
||||
if result.Detail != nil {
|
||||
if s, ok := result.Detail.(string); ok && s != "" {
|
||||
resultText = result.Message + "\n" + s
|
||||
} else if b, err := json.Marshal(result.Detail); err == nil && len(b) > 0 && string(b) != "null" {
|
||||
resultText = result.Message + "\n" + string(b)
|
||||
}
|
||||
}
|
||||
}
|
||||
updates := map[string]any{
|
||||
taskExecutionColStatus: taskExecutionStatusSucceeded,
|
||||
"error_message": "",
|
||||
"result": "ok",
|
||||
"result": resultText,
|
||||
"finished_at": now,
|
||||
"duration": duration.Milliseconds(),
|
||||
}
|
||||
|
||||
@@ -119,13 +119,7 @@ func (p *Plugin) ConfigEnabled(view core.ConfigView) bool {
|
||||
func (p *Plugin) Apply(ctx *core.Context) error {
|
||||
p.coreCtx = ctx
|
||||
|
||||
if db, err := core.Inject[contracts.DBService](ctx); err == nil && db != nil {
|
||||
setDBService(db)
|
||||
} else {
|
||||
core.When[contracts.DBService](ctx, func(db contracts.DBService) {
|
||||
setDBService(db)
|
||||
})
|
||||
}
|
||||
core.Bind[contracts.DBService](ctx, setDBService)
|
||||
|
||||
taskSvc := newInprocTaskService(ctx.Tasks())
|
||||
core.Provide[contracts.TaskService](ctx, taskSvc)
|
||||
@@ -151,6 +145,9 @@ func (p *Plugin) Start(ctx context.Context) error {
|
||||
if p.queue == nil {
|
||||
p.queue = NewInprocQueue(p.concurrency, p.queueCapacity, p.coreCtx.Tasks())
|
||||
}
|
||||
if p.coreCtx != nil {
|
||||
p.queue.appCtx = p.coreCtx.Root()
|
||||
}
|
||||
|
||||
globalMu.Lock()
|
||||
globalQueue = p.queue
|
||||
|
||||
@@ -80,9 +80,9 @@ func TestInprocWorkerDispatchByTypeTracksExecution(t *testing.T) {
|
||||
require.NoError(t, p.Apply(ctx))
|
||||
|
||||
var executedCount atomic.Int32
|
||||
ctx.Tasks().Register("system:cleanup", func(_ context.Context, _ []byte) error {
|
||||
ctx.Tasks().Register("system:cleanup", func(_ context.Context, _ []byte) (*contracts.TaskResultDTO, error) {
|
||||
executedCount.Add(1)
|
||||
return nil
|
||||
return &contracts.TaskResultDTO{Message: "cleaned 3 files"}, nil
|
||||
},
|
||||
extpoints.WithTaskType("system_cleanup"),
|
||||
extpoints.WithTaskName("系统垃圾清理"),
|
||||
@@ -111,6 +111,6 @@ func TestInprocWorkerDispatchByTypeTracksExecution(t *testing.T) {
|
||||
if listErr != nil || total == 0 || len(execs) == 0 {
|
||||
return false
|
||||
}
|
||||
return execs[0].TaskID == taskID && execs[0].Status == "succeeded" && execs[0].TaskName == "系统垃圾清理"
|
||||
return execs[0].TaskID == taskID && execs[0].Status == "succeeded" && execs[0].TaskName == "系统垃圾清理" && execs[0].Result == "cleaned 3 files"
|
||||
}, 2*time.Second, 20*time.Millisecond, "inproc worker should persist a succeeded execution record")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user