Method Run

cron/cron.go:114–173

Run starts the cron scheduler. It is a blocking call.

(ctx context.Context)

// Run starts the cron scheduler. It is a blocking call.
func (c *Cron) Run(ctx context.Context) error {
	defer c.quitWaiter.Wait()

	c.lock.L.Lock()
	now := c.now()
	for _, descriptor := range c.jobDescriptors {
		descriptor.next = descriptor.Schedule.Next(now)
	}
	heap.Init(&c.jobDescriptors)
	c.lock.L.Unlock()

	var once sync.Once
	for {
		c.lock.L.Lock()
		// Determine the next entry to run.
		for c.jobDescriptors.Len() == 0 || c.jobDescriptors[0].next.IsZero() {
			c.broadcastAtDeadlineOnce(ctx, &once)
			select {
			case <-ctx.Done():
				c.lock.L.Unlock()
				return ctx.Err()
			default:
				c.lock.Wait()
			}
		}
		gap := c.jobDescriptors[0].next.Sub(now)
		c.lock.L.Unlock()

		timer := time.NewTimer(gap)

		select {
		case now = <-timer.C:
			c.lock.L.Lock()
			for {
				if c.jobDescriptors[0].next.After(now) || c.jobDescriptors[0].next.IsZero() {
					break
				}
				descriptor := heap.Pop(&c.jobDescriptors).(*JobDescriptor)

				descriptor.prev = descriptor.next
				descriptor.next = descriptor.Schedule.Next(now)
				heap.Push(&c.jobDescriptors, descriptor)

				var innerCtx context.Context
				innerCtx = context.WithValue(ctx, prevContextKey, descriptor.prev)
				innerCtx = context.WithValue(innerCtx, nextContextKey, descriptor.next)

				c.quitWaiter.Add(1)
				go func() {
					defer c.quitWaiter.Done()
					descriptor.Run(innerCtx)
				}()
			}
			c.lock.L.Unlock()
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		}