feat(scheduler)!: support custom mutex for job queue operations (#133)

examples/queue/file_system.go

@@ -39,7 +39,7 @@ func main() {
  jobQueue := newJobQueue()
  scheduler := quartz.NewStdSchedulerWithOptions(quartz.StdSchedulerOptions{
  OutdatedThreshold: time.Second, // considering file system I/O latency
- }, jobQueue)
+ }, jobQueue, nil)
  scheduler.Start(ctx)
 
  jobQueueSize, err := jobQueue.Size()

quartz/scheduler.go

@@ -72,6 +72,7 @@ type StdScheduler struct {
  mtx sync.Mutex
  wg sync.WaitGroup
  queue JobQueue
+ queueMtx sync.Locker
  interrupt chan struct{}
  cancel context.CancelFunc
  feeder chan ScheduledJob
@@ -125,30 +126,37 @@ type StdSchedulerOptions struct {
 // Verify StdScheduler satisfies the Scheduler interface.
 var _ Scheduler = (*StdScheduler)(nil)
 
-// NewStdScheduler returns a new StdScheduler with the default
-// configuration.
+// NewStdScheduler returns a new StdScheduler with the default configuration.
 func NewStdScheduler() Scheduler {
  return NewStdSchedulerWithOptions(StdSchedulerOptions{
  OutdatedThreshold: 100 * time.Millisecond,
  RetryInterval: 100 * time.Millisecond,
- }, nil)
+ }, nil, nil)
 }
 
-// NewStdSchedulerWithOptions returns a new StdScheduler configured
-// as specified.
-// A custom JobQueue implementation can be provided to manage scheduled
-// jobs. This can be useful when distributed mode is required, so that
-// jobs can be stored in persistent storage.
-// Pass in nil to use the internal in-memory implementation.
+// NewStdSchedulerWithOptions returns a new StdScheduler configured as specified.
+//
+// A custom [JobQueue] implementation may be provided to manage scheduled jobs.
+// This is useful when distributed mode is required, so that jobs can be stored
+// in persistent storage. Pass in nil to use the internal in-memory implementation.
+//
+// A custom [sync.Locker] may also be provided to ensure that scheduler operations
+// on the job queue are atomic when used in distributed mode. Pass in nil to use
+// the default [sync.Mutex].
 func NewStdSchedulerWithOptions(
  opts StdSchedulerOptions,
  jobQueue JobQueue,
+ jobQueueMtx sync.Locker,
 ) *StdScheduler {
  if jobQueue == nil {
  jobQueue = newJobQueue()
  }
+ if jobQueueMtx == nil {
+ jobQueueMtx = &sync.Mutex{}
+ }
  return &StdScheduler{
  queue: jobQueue,
+ queueMtx: jobQueueMtx,
  interrupt: make(chan struct{}, 1),
  feeder: make(chan ScheduledJob),
  dispatch: make(chan ScheduledJob),
@@ -161,8 +169,8 @@ func (sched *StdScheduler) ScheduleJob(
  jobDetail *JobDetail,
  trigger Trigger,
 ) error {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  if jobDetail == nil {
  return illegalArgumentError("jobDetail is nil")
@@ -244,8 +252,8 @@ func (sched *StdScheduler) IsStarted() bool {
 // For a job key to be returned, the job must satisfy all of the matchers specified.
 // Given no matchers, it returns the keys of all scheduled jobs.
 func (sched *StdScheduler) GetJobKeys(matchers ...Matcher[ScheduledJob]) ([]*JobKey, error) {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  scheduledJobs, err := sched.queue.ScheduledJobs(matchers)
  if err != nil {
@@ -260,8 +268,8 @@ func (sched *StdScheduler) GetJobKeys(matchers ...Matcher[ScheduledJob]) ([]*Job
 
 // GetScheduledJob returns the ScheduledJob with the specified key.
 func (sched *StdScheduler) GetScheduledJob(jobKey *JobKey) (ScheduledJob, error) {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  if jobKey == nil {
  return nil, illegalArgumentError("jobKey is nil")
@@ -271,8 +279,8 @@ func (sched *StdScheduler) GetScheduledJob(jobKey *JobKey) (ScheduledJob, error)
 
 // DeleteJob removes the Job with the specified key if present.
 func (sched *StdScheduler) DeleteJob(jobKey *JobKey) error {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  if jobKey == nil {
  return illegalArgumentError("jobKey is nil")
@@ -290,8 +298,8 @@ func (sched *StdScheduler) DeleteJob(jobKey *JobKey) error {
 // PauseJob suspends the job with the specified key from being
 // executed by the scheduler.
 func (sched *StdScheduler) PauseJob(jobKey *JobKey) error {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  if jobKey == nil {
  return illegalArgumentError("jobKey is nil")
@@ -324,8 +332,8 @@ func (sched *StdScheduler) PauseJob(jobKey *JobKey) error {
 
 // ResumeJob restarts the suspended job with the specified key.
 func (sched *StdScheduler) ResumeJob(jobKey *JobKey) error {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  if jobKey == nil {
  return illegalArgumentError("jobKey is nil")
@@ -362,8 +370,8 @@ func (sched *StdScheduler) ResumeJob(jobKey *JobKey) error {
 
 // Clear removes all of the scheduled jobs.
 func (sched *StdScheduler) Clear() error {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  // reset the job queue
  err := sched.queue.Clear()
@@ -545,8 +553,8 @@ func (sched *StdScheduler) validateJob(job ScheduledJob) (bool, func() (int64, e
 }
 
 func (sched *StdScheduler) fetchAndReschedule() (ScheduledJob, bool) {
- sched.mtx.Lock()
- defer sched.mtx.Unlock()
+ sched.queueMtx.Lock()
+ defer sched.queueMtx.Unlock()
 
  // fetch a job for processing
  job, err := sched.queue.Pop()

quartz/scheduler_test.go

@@ -110,7 +110,7 @@ func TestScheduler_BlockingSemantics(t *testing.T) {
 
  opts.OutdatedThreshold = 10 * time.Millisecond
 
- sched := quartz.NewStdSchedulerWithOptions(opts, nil)
+ sched := quartz.NewStdSchedulerWithOptions(opts, nil, nil)
  ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
  defer cancel()
  sched.Start(ctx)
@@ -394,7 +394,7 @@ func TestScheduler_MisfiredJob(t *testing.T) {
  OutdatedThreshold: time.Millisecond,
  RetryInterval: time.Millisecond,
  MisfiredChan: misfiredChan,
- }, nil)
+ }, nil, nil)
 
  jobDetail := quartz.NewJobDetail(funcJob, quartz.NewJobKey("funcJob"))
  err := sched.ScheduleJob(jobDetail, quartz.NewSimpleTrigger(2*time.Millisecond))