Documentation
¶
Overview ¶
Example (Pipeline) ¶
Example_pipeline demonstrates driving a pipe.Scheduler with a periodic cron trigger.
package main
import (
"context"
"time"
"github.com/xoctopus/concx/pkg/orch/cron"
"github.com/xoctopus/concx/pkg/orch/pipe"
"github.com/xoctopus/concx/pkg/schedx"
)
func main() {
ctx := context.Background()
p := pipe.FromJob[string, string, string](
"fetch",
pipe.TransformFunc[string, string](func(_ context.Context, v string) (string, error) {
return v, nil
}),
).EndJob(
"persist",
pipe.TransformFunc[string, string](func(_ context.Context, v string) (string, error) {
return v, nil
}),
).Build(
pipe.WithMaxPending(4),
pipe.WithParallel(2),
)
_ = p.Run(ctx)
defer func() { _ = p.Close() }()
c := cron.NewWithSchedule(
ctx,
cron.MustSpec("@every 10s"),
schedx.JobFunc[time.Time](func(ctx context.Context, t time.Time) error {
ret, err := p.Push(ctx, t.Format(time.RFC3339))
if err != nil {
return err
}
_, err = ret.Result(ctx)
return err
}),
cron.WithOverlapSkip(),
)
defer func() { _ = c.Close() }()
}
Output:
Index ¶
- type Cron
- type Option
- func WithCallback(cb schedx.HandlerCallback[time.Time]) Option
- func WithLocation(loc *time.Location) Option
- func WithMaxPending(maxPending int) Option
- func WithName(name string) Option
- func WithOverlapQueue(maxPending int) Option
- func WithOverlapSkip() Option
- func WithParallel(parallel int) Option
- func WithParser(p robfig.ScheduleParser) Option
- func WithSeconds() Option
- func WithShutdownTimeout(timeout time.Duration) Option
- func WithoutDetached() Option
- func WithoutPendingLimitation() Option
- type OverlapPolicy
- type Schedule
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Cron ¶
type Cron interface {
// Done closes when the cron orchestrator is shut down.
Done() <-chan struct{}
// Close stops the cron orchestrator and waits for active jobs to exit gracefully up to shutdownTimeout.
Close() error
}
Cron is an orchestrator recipe for a single recurring scheduled job.
func New ¶
New creates and starts a single-task Cron orchestrator parsing spec. spec supports standard cron expression, Quartz 6-field with WithSeconds(), and "@every <duration>".
Example ¶
ExampleNew demonstrates registering a recurring job.
package main
import (
"context"
"fmt"
"time"
"github.com/xoctopus/concx/pkg/orch/cron"
"github.com/xoctopus/concx/pkg/schedx"
)
func main() {
ctx := context.Background()
c, _ := cron.New(ctx, "@every 50ms", schedx.JobFunc[time.Time](func(_ context.Context, _ time.Time) error {
fmt.Println("periodic tick")
return nil
}), cron.WithName("periodic"), cron.WithShutdownTimeout(2*time.Second))
defer func() { _ = c.Close() }()
time.Sleep(60 * time.Millisecond)
}
Output: periodic tick
type Option ¶
type Option func(*option)
Option configures a single-task Cron orchestrator.
func WithCallback ¶
func WithCallback(cb schedx.HandlerCallback[time.Time]) Option
WithCallback sets a callback invoked after each job execution finishes (with tick time and error/panic).
func WithLocation ¶
WithLocation sets the timezone for parsing expressions and scheduling. Defaults to time.Local.
func WithMaxPending ¶
WithMaxPending sets the maximum queued tick backlog for the entry. Defaults to 1.
func WithOverlapQueue ¶
WithOverlapQueue sets the overlap policy to queue new ticks up to maxPending.
func WithOverlapSkip ¶
func WithOverlapSkip() Option
WithOverlapSkip sets the overlap policy to skip new ticks when busy. This is the default policy.
func WithParallel ¶
WithParallel sets how many instances of the job may run concurrently. Defaults to 1.
func WithParser ¶
func WithParser(p robfig.ScheduleParser) Option
WithParser sets a custom cron ScheduleParser.
func WithSeconds ¶
func WithSeconds() Option
WithSeconds enables the 6-field parser where seconds is required (Quartz style).
func WithShutdownTimeout ¶
WithShutdownTimeout sets the maximum duration to wait for running jobs to finish during Close. Defaults to 5 seconds.
func WithoutDetached ¶
func WithoutDetached() Option
WithoutDetached causes the job execution context to be canceled immediately when Cron is canceled.
func WithoutPendingLimitation ¶
func WithoutPendingLimitation() Option
WithoutPendingLimitation disables the queue backlog limit for the entry.
type OverlapPolicy ¶
type OverlapPolicy int
OverlapPolicy controls behavior when a new tick occurs while the job is still executing.
const ( // OverlapSkip drops the new tick if all parallel slots for the job are busy. OverlapSkip OverlapPolicy = iota // OverlapQueue pushes the new tick into the job's scheduler queue up to maxPending. OverlapQueue )
Source Files