refactor(drivers): decouple inproc cron from inproc worker and satisfy linter

This commit is contained in:
ryan
2026-08-28 15:17:11 +08:00
parent 0ff117da47
commit 06508b6f13
6 changed files with 60 additions and 20 deletions
@@ -9,6 +9,7 @@ import (
"sync"
"Wavelet/core"
"Wavelet/core/contracts"
)
// Plugin implements core.Plugin and core.Driver for in-process cron job scheduling.
@@ -62,7 +63,8 @@ func (p *Plugin) Start(_ context.Context) error {
defer p.mu.Unlock()
if p.scheduler == nil {
p.scheduler = newInprocScheduler(p.coreCtx.Schedules(), p.coreCtx.Tasks())
taskSvc, _ := core.Inject[contracts.TaskService](p.coreCtx)
p.scheduler = newInprocScheduler(p.coreCtx.Schedules(), p.coreCtx.Tasks(), taskSvc)
}
return p.scheduler.Start()
@@ -14,17 +14,11 @@ import (
"Wavelet/core"
"Wavelet/plugins/drivers/driver_inproc_cron"
"Wavelet/plugins/drivers/driver_inproc_worker"
)
func TestInprocCronPlugin(t *testing.T) {
ctx := core.NewContext(context.Background())
workerPlugin := driver_inproc_worker.New()
require.NoError(t, workerPlugin.Apply(ctx))
require.NoError(t, workerPlugin.Start(context.Background()))
defer func() { _ = workerPlugin.Stop(context.Background()) }()
cronPlugin := driver_inproc_cron.New()
assert.Equal(t, "driver_inproc_cron", cronPlugin.Name())
assert.Equal(t, core.DriverTypeScheduler, cronPlugin.Type())
@@ -6,12 +6,17 @@ package driver_inproc_cron
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"time"
"github.com/robfig/cron/v3"
"Wavelet/core/contracts"
"Wavelet/core/extpoints"
"Wavelet/pkg/logger"
"Wavelet/plugins/drivers/driver_inproc_worker"
"github.com/robfig/cron/v3"
"Wavelet/pkg/util"
)
type inprocScheduler struct {
@@ -19,14 +24,16 @@ type inprocScheduler struct {
cronRunner *cron.Cron
scheduleReg extpoints.ScheduleExtension
taskReg extpoints.TaskExtension
taskSvc contracts.TaskService
running bool
}
func newInprocScheduler(scheduleReg extpoints.ScheduleExtension, taskReg extpoints.TaskExtension) *inprocScheduler {
func newInprocScheduler(scheduleReg extpoints.ScheduleExtension, taskReg extpoints.TaskExtension, taskSvc contracts.TaskService) *inprocScheduler {
return &inprocScheduler{
cronRunner: cron.New(cron.WithSeconds()),
scheduleReg: scheduleReg,
taskReg: taskReg,
taskSvc: taskSvc,
}
}
@@ -62,14 +69,15 @@ func (s *inprocScheduler) Stop() {
s.running = false
}
const standardCronFields = 5
func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
spec := def.Spec
taskType := def.TaskType
// Support 5-field cron expression by prepending "0 " for 6-field seconds parser
fields := len(cronFields(spec))
cronSpec := spec
if fields == 5 {
if fields == standardCronFields {
cronSpec = "0 " + spec
}
@@ -86,9 +94,25 @@ func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
}
_, err := s.cronRunner.AddFunc(cronSpec, func() {
_, dispatchErr := driver_inproc_worker.DispatchTask(context.Background(), taskType, payloadBytes, "inproc_cron")
if dispatchErr != nil {
logger.ErrorF(context.Background(), "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr)
if s.taskSvc != nil {
if _, dispatchErr := s.taskSvc.Dispatch(context.Background(), taskType, payloadBytes, "inproc_cron"); dispatchErr != nil {
logger.ErrorF(context.Background(), "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr)
}
return
}
if s.taskReg != nil {
if td, ok := s.taskReg.Get(taskType); ok {
util.Go(func() {
timeout := td.Timeout
if timeout <= 0 {
timeout = 5 * time.Minute
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
_ = invokeHandler(ctx, td.Handler, payloadBytes)
})
}
}
})
if err != nil {
@@ -96,6 +120,25 @@ func (s *inprocScheduler) registerJob(def extpoints.ScheduleDefinition) {
}
}
func invokeHandler(ctx context.Context, handler any, payload []byte) error {
if handler == nil {
return errors.New("nil task handler")
}
switch fn := handler.(type) {
case func(context.Context, []byte) error:
return fn(ctx, payload)
case func(context.Context) error:
return fn(ctx)
case func([]byte) error:
return fn(payload)
case func() error:
return fn()
default:
return fmt.Errorf("unsupported handler type: %T", handler)
}
}
func cronFields(s string) []string {
var fields []string
var current []rune