fix: add missing job provider

This commit is contained in:
rogeecn
2025-03-17 09:56:47 +08:00
parent 6442bbf58c
commit 206aee6951

View File

@@ -114,22 +114,31 @@ func (q *Job) StopAndCancel(ctx context.Context) error {
} }
func (q *Job) AddPeriodicJobs(job contracts.CronJob) (map[string]rivertype.PeriodicJobHandle, error) { func (q *Job) AddPeriodicJobs(job contracts.CronJob) (map[string]rivertype.PeriodicJobHandle, error) {
client, err := q.Client() var err error
if err != nil {
return nil, err
}
handles := make(map[string]rivertype.PeriodicJobHandle) handles := make(map[string]rivertype.PeriodicJobHandle)
for _, job := range job.Args() { for _, job := range job.Args() {
handles[job.Kind] = client.PeriodicJobs().Add(river.NewPeriodicJob( handles[job.Kind], err = q.AddPeriodicJob(job)
job.PeriodicInterval, if err != nil {
func() (river.JobArgs, *river.InsertOpts) { return handles, err
return job.Arg, lo.ToPtr(job.Arg.InsertOpts()) }
},
&river.PeriodicJobOpts{
RunOnStart: job.RunOnStart,
},
))
} }
return handles, nil return handles, nil
} }
func (q *Job) AddPeriodicJob(job contracts.CronJobArg) (rivertype.PeriodicJobHandle, error) {
client, err := q.Client()
if err != nil {
return 0, err
}
return client.PeriodicJobs().Add(river.NewPeriodicJob(
job.PeriodicInterval,
func() (river.JobArgs, *river.InsertOpts) {
return job.Arg, lo.ToPtr(job.Arg.InsertOpts())
},
&river.PeriodicJobOpts{
RunOnStart: job.RunOnStart,
},
)), nil
}