package main

import (
    "fmt"
    "github.com/robfig/cron/v3"
)

func main() {
    // 創(chuàng)建一個默認(rèn)的cron對象
    c := cron.New()

    //添加執(zhí)行任務(wù)
    c.AddFunc("30 * * * *", func() { fmt.Println("Every hour on the half hour") })
    c.AddFunc("@hourly", func() { fmt.Println("Every hour, starting an hour from now") })
    c.AddFunc("@every 1h30m", func() { fmt.Println("Every hour thirty, starting an hour thirty from now") })

    //開始執(zhí)行任務(wù)
    c.Start()
    select {} //阻塞
}

// Entry 數(shù)據(jù)結(jié)構(gòu)，每一個被調(diào)度實體一個
type Entry struct {
  // 唯一id，用于查詢和刪除
  ID EntryID
  // 本Entry的調(diào)度時間，不是絕對時間，在生成entry時會計算出來
  Schedule Schedule
  // 本entry下次需要執(zhí)行的絕對時間，會一直被更新
  // 被封裝的含義是Job可以多層嵌套，可以實現(xiàn)基于需要執(zhí)行Job的額外處理
  // 比如抓取Job異常、如果Job沒有返回下一個時間點的Job是還是繼續(xù)執(zhí)行還是delay
  Next time.Time
  // 上一次被執(zhí)行時間，主要用來查詢
  Prev time.Time
  // WrappedJob 是真實執(zhí)行的Job實體
  WrappedJob Job
  // Job 主要給用戶查詢
  Job Job
}
// Cron保持任意數(shù)量的任務(wù)的軌道，調(diào)用相關(guān)的func時間表指定。它可以被啟動，停止，可運行的同時進(jìn)行檢查。
type Cron struct {
  entries   []*Entry          // 保存了所有加入到 Cron 的任務(wù)
   // chain 用來定義entry里的warppedJob使用什么邏輯（e.g. skipIfLastRunning）
   // 即一個cron里所有entry只有一個封裝邏輯
  chain     Chain            
  stop      chan struct{}     // 停止整個cron的channel
  add       chan *Entry       // 增加一個entry的channel
  remove    chan EntryID      // 移除一個entry的channel
  snapshot  chan chan []Entry // 獲取entry整體快照的channel
  running   bool              // 代表是否已經(jīng)在執(zhí)行，是cron為使用者提供的動態(tài)修改entry的接口準(zhǔn)備的
  logger    Logger            // 封裝golang的log包
  runningMu sync.Mutex        // 用來修改運行中的cron數(shù)據(jù)，比如增加entry，移除entry
  location  *time.Location    // 地理位置
  parser    ScheduleParser    // 對時間格式的解析，為interface, 可以定制自己的時間規(guī)則。
  nextID    EntryID           // entry的全局ID，新增一個entry就加1
  jobWaiter sync.WaitGroup    // run job時會進(jìn)行add(1)， job 結(jié)束會done()，stop整個cron，以此保證所有job都能退出
}

func New(opts ...Option) *Cron {
	c := &Cron{
		entries:   nil,
		chain:     NewChain(),
		add:       make(chan *Entry),
		stop:      make(chan struct{}),
		snapshot:  make(chan chan []Entry),
		remove:    make(chan EntryID),
		running:   false,
		runningMu: sync.Mutex{},
		logger:    DefaultLogger,
		location:  time.Local,
		parser:    standardParser,
	}
	for _, opt := range opts {
		opt(c)
	}
	return c
}

func (c *Cron) AddFunc(spec string, cmd func()) (EntryID, error) {
	return c.AddJob(spec, FuncJob(cmd)) //包裝成job類型然后調(diào)用AddJob()方法
}

func (c *Cron) AddJob(spec string, cmd Job) (EntryID, error) {
	schedule, err := c.parser.Parse(spec) //將cron表達(dá)式解析成schedule類型
	if err != nil {
		return 0, err
	}
	return c.Schedule(schedule, cmd), nil //調(diào)用Schedule()
}

func (c *Cron) Schedule(schedule Schedule, cmd Job) EntryID {
	c.runningMu.Lock() //為了保證線程安全，加鎖
	defer c.runningMu.Unlock()
	c.nextID++ //下一EntryID
	entry := &Entry{
		ID:         c.nextID,
		Schedule:   schedule,
		WrappedJob: c.chain.Then(cmd),
		Job:        cmd,
	}
    // Cron是否處于運行狀態(tài)
	if !c.running {
		c.entries = append(c.entries, entry) // 追加到entries列表中
	} else {
		c.add <- entry // 發(fā)送到Cron的add chan
	}
	return entry.ID
}

func (c *Cron) Start() {
	c.runningMu.Lock() // 獲取鎖
	defer c.runningMu.Unlock()
	if c.running {
		return
	}
	c.running = true // 將 c.running 置為 true 表示 cron 已經(jīng)在運行中了
	go c.run() //開啟一個 goroutine 執(zhí)行 c.run()
}

// run the scheduler.. this is private just due to the need to synchronize
// access to the 'running' state variable.
func (c *Cron) run() {
	c.logger.Info("start")
 
	// Figure out the next activation times for each entry.
	now := c.now()
	for _, entry := range c.entries {
		entry.Next = entry.Schedule.Next(now)
		c.logger.Info("schedule", "now", now, "entry", entry.ID, "next", entry.Next)
	}
 
	for {
		// Determine the next entry to run.
		// 將定時任務(wù)執(zhí)行時間進(jìn)行排序,最近最早執(zhí)行的放在前面
		sort.Sort(byTime(c.entries))
 
		var timer *time.Timer
		if len(c.entries) == 0 || c.entries[0].Next.IsZero() {
			// If there are no entries yet, just sleep - it still handles new entries
			// and stop requests.
			timer = time.NewTimer(100000 * time.Hour)
		} else {
			// 生成一個定時器,距離最近的任務(wù)時間到時 觸發(fā)定時器的channel,發(fā)送通知
			timer = time.NewTimer(c.entries[0].Next.Sub(now))
		}
 
		for {
			select {
			// 定時時間到了,執(zhí)行定時任務(wù),并設(shè)置下次執(zhí)行的時刻
			case now = <-timer.C:
				now = now.In(c.location)
				c.logger.Info("wake", "now", now)
 
				// Run every entry whose next time was less than now
				//對每個定時任務(wù)嘗試執(zhí)行
				for _, e := range c.entries {
					if e.Next.After(now) || e.Next.IsZero() {
						break
					}
					c.startJob(e.WrappedJob)
					e.Prev = e.Next
					e.Next = e.Schedule.Next(now)
					c.logger.Info("run", "now", now, "entry", e.ID, "next", e.Next)
				}
			//新增的定時任務(wù)添加到 任務(wù)列表中
			case newEntry := <-c.add:
				timer.Stop()
				now = c.now()
				newEntry.Next = newEntry.Schedule.Next(now)
				c.entries = append(c.entries, newEntry)
				c.logger.Info("added", "now", now, "entry", newEntry.ID, "next", newEntry.Next)
			//獲取 當(dāng)前所有定時任務(wù)(快照)
			case replyChan := <-c.snapshot:
				replyChan <- c.entrySnapshot()
				continue
			//停止定時任務(wù),timer停止即可完成此功能
			case <-c.stop:
				timer.Stop()
				c.logger.Info("stop")
				return
			//刪除某個定時任務(wù)
			case id := <-c.remove:
				timer.Stop()
				now = c.now()
				c.removeEntry(id)
				c.logger.Info("removed", "entry", id)
			}
 
			break
		}
	}
}

func (c *Cron) Stop() context.Context {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		c.stop <- struct{}{} // 會發(fā)出一個 stop 信號
		c.running = false
	}
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
         // 等待所有已經(jīng)在執(zhí)行的任務(wù)執(zhí)行完畢
		c.jobWaiter.Wait()
         // 會發(fā)出一個 cancelCtx.Done() 信號
		cancel()
	}()
	return ctx
}

func (c *Cron) Remove(id EntryID) {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		c.remove <- id // 會發(fā)出一個 remove 信號
	} else {
		c.removeEntry(id)
	}
}

func (c *Cron) removeEntry(id EntryID) {
	var entries []*Entry
	for _, e := range c.entries {
		if e.ID != id {
			entries = append(entries, e)
		}
	}
	c.entries = entries
}