45 lines
978 B
Go
45 lines
978 B
Go
|
package job
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"log"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
domusecase "haixun-backend/internal/model/job/domain/usecase"
|
||
|
|
)
|
||
|
|
|
||
|
|
type Reaper struct {
|
||
|
|
jobs domusecase.UseCase
|
||
|
|
}
|
||
|
|
|
||
|
|
func NewReaper(jobs domusecase.UseCase) *Reaper {
|
||
|
|
return &Reaper{jobs: jobs}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (r *Reaper) Start(ctx context.Context, interval time.Duration) {
|
||
|
|
if interval <= 0 {
|
||
|
|
interval = 30 * time.Second
|
||
|
|
}
|
||
|
|
log.Printf("job reaper started: interval=%s", interval)
|
||
|
|
ticker := time.NewTicker(interval)
|
||
|
|
defer ticker.Stop()
|
||
|
|
|
||
|
|
for {
|
||
|
|
select {
|
||
|
|
case <-ctx.Done():
|
||
|
|
log.Printf("job reaper stopped")
|
||
|
|
return
|
||
|
|
case <-ticker.C:
|
||
|
|
result, err := r.jobs.RunMaintenance(ctx)
|
||
|
|
if err != nil {
|
||
|
|
log.Printf("job reaper error: %v", err)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if result.EnqueuedPending > 0 || result.ReapedCancelGrace > 0 || result.ReapedExpiredLocks > 0 {
|
||
|
|
log.Printf("job reaper: enqueued=%d cancel_grace=%d expired=%d",
|
||
|
|
result.EnqueuedPending, result.ReapedCancelGrace, result.ReapedExpiredLocks)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|