package main

import (
	"fmt"
	"time"

	"github.com/jasonlvhit/gocron"
)

func task() {
	fmt.Println("I am running task.")
}

func taskWithParams(a int, b string) {
	fmt.Println(a, b)
}

func main() {
	// Do jobs without params
	gocron.Every(1).Second().Do(task)
	gocron.Every(2).Seconds().Do(task)
	gocron.Every(1).Minute().Do(task)
	gocron.Every(2).Minutes().Do(task)
	gocron.Every(1).Hour().Do(task)
	gocron.Every(2).Hours().Do(task)
	gocron.Every(1).Day().Do(task)
	gocron.Every(2).Days().Do(task)
	gocron.Every(1).Week().Do(task)
	gocron.Every(2).Weeks().Do(task)

	// Do jobs with params
	gocron.Every(1).Second().Do(taskWithParams, 1, "hello")

	// Do jobs on specific weekday
	gocron.Every(1).Monday().Do(task)
	gocron.Every(1).Thursday().Do(task)

	// Do a job at a specific time - 'hour:min:sec' - seconds optional
	gocron.Every(1).Day().At("10:30").Do(task)
	gocron.Every(1).Monday().At("18:30").Do(task)
	gocron.Every(1).Tuesday().At("18:30:59").Do(task)

	// Begin job immediately upon start
	gocron.Every(1).Hour().From(gocron.NextTick()).Do(task)

	// Begin job at a specific date/time
	t := time.Date(2019, time.November, 10, 15, 0, 0, 0, time.Local)
	gocron.Every(1).Hour().From(&t).Do(task)

	// NextRun gets the next running time
	_, time := gocron.NextRun()
	fmt.Println(time)

	// Remove a specific job
	gocron.Remove(task)

	// Clear all scheduled jobs
	gocron.Clear()

	// Start all the pending jobs
	<- gocron.Start()

	// also, you can create a new scheduler
	// to run two schedulers concurrently
	s := gocron.NewScheduler()
	s.Every(3).Seconds().Do(task)
	<- s.Start()
}

// gocron/scheduler.go
// Scheduler struct, the only data member is the list of jobs.
// - implements the sort.Interface{} for sorting jobs, by the time nextRun
type Scheduler struct {
	jobs [MAXJOBNUM]*Job // Array store jobs
	size int             // Size of jobs which jobs holding.
	loc  *time.Location  // Location to use when scheduling jobs with specified times
}

var (
	defaultScheduler = NewScheduler()
)

s := gocron.NewScheduler()
s.Every(3).Seconds().Do(task)
<- s.Start()

// Job struct keeping information about job
type Job struct {
	interval uint64                   // pause interval * unit between runs
	jobFunc  string                   // the job jobFunc to run, func[jobFunc]
	//......
	funcs    map[string]interface{}   // Map for the function task store
	fparams  map[string][]interface{} // Map for function and  params of function
	//......
}

// gocron/job.go
// func (j *Job) Do(jobFun interface{}, params ...interface{}) error
fname := getFunctionName(jobFun)
j.funcs[fname] = jobFun
j.fparams[fname] = params
j.jobFunc = fname

// gocron/job.go
// func (j *Job) run() ([]reflect.Value, error)
result, err := callJobFuncWithParams(j.funcs[j.jobFunc], j.fparams[j.jobFunc])
if err != nil {
	return nil, err
}

// gocron/gocron.go
func callJobFuncWithParams(jobFunc interface{}, params []interface{}) ([]reflect.Value, error) {
	f := reflect.ValueOf(jobFunc)
	if len(params) != f.Type().NumIn() {
		return nil, ErrParamsNotAdapted
	}
	in := make([]reflect.Value, len(params))
	for k, param := range params {
		in[k] = reflect.ValueOf(param)
	}
	return f.Call(in), nil
}

// Start all the pending jobs
// Add seconds ticker
func (s *Scheduler) Start() chan bool {
	stopped := make(chan bool, 1)
	// ticker每秒產(chǎn)生一個信號
	ticker := time.NewTicker(1 * time.Second)

	go func() {
		for {
			// select選擇器阻塞
			// case接收到信號則執(zhí)行
			// 同時接收到多個信號則隨機選擇一個執(zhí)行
			select {
			// ticker每秒產(chǎn)生一次信號
			// RunPending輪詢jobs列表，尋找到了時間可執(zhí)行的任務
			case <-ticker.C:
				s.RunPending()
			// stopped接收到停止信號，退出調(diào)度器協(xié)程
			case <-stopped:
				ticker.Stop()
				return
			}
		}
	}()

	return stopped
}

// RunPending runs all the jobs that are scheduled to run.
func (s *Scheduler) RunPending() {
	// 輪詢jobs列表，找到到時間可執(zhí)行的任務，創(chuàng)建可執(zhí)行任務列表
	runnableJobs, n := s.getRunnableJobs()

	if n != 0 {
		for i := 0; i < n; i++ {
			// 啟動協(xié)程運行
			go runnableJobs[i].run()
			// 刷新job執(zhí)行信息，等待下一輪調(diào)度
			runnableJobs[i].lastRun = time.Now()
			runnableJobs[i].scheduleNextRun()
		}
	}
}

// cron expressions supported
s.Cron("*/1 * * * *").Do(task) // every minute

// you can start running the scheduler in two different ways:
// starts the scheduler asynchronously
s.StartAsync()
// starts the scheduler and blocks current execution path
s.StartBlocking()

// gocron/scheduler.go
// SetMaxConcurrentJobs limits how many jobs can be running at the same time.
// This is useful when running resource intensive jobs and a precise start time is not critical.
func (s *Scheduler) SetMaxConcurrentJobs(n int, mode limitMode) {
	// 通過對n的配置修改并發(fā)任務數(shù)的大小
	s.executor.maxRunningJobs = semaphore.NewWeighted(int64(n))
	// limitMode即當可執(zhí)行任務達到最大并發(fā)量時，應該如何處理的邏輯
	// RescheduleMode：跳過本次執(zhí)行，等待下一次調(diào)度
	// WaitMode：持續(xù)等待，知道可執(zhí)行隊列空出。但，由于等待的任務數(shù)積累，可能導致不可預知的后果，某些任務可能一直等不到執(zhí)行
	s.executor.limitMode = mode
}

// gocron/executor.go
// 通過信號量的方式從最大數(shù)量中取一位
// 若通過，下一步可以執(zhí)行函數(shù)
if e.maxRunningJobs != nil {
	if !e.maxRunningJobs.TryAcquire(1) {

		switch e.limitMode {
		case RescheduleMode:
			return
		case WaitMode:
			select {
			case <-stopCtx.Done():
				return
			case <-f.ctx.Done():
				return
			default:
			}

			if err := e.maxRunningJobs.Acquire(f.ctx, 1); err != nil {
				break

			}
		}
	}

	defer e.maxRunningJobs.Release(1)
}

// gocron/job.go
// SingletonMode prevents a new job from starting if the prior job has not yet
// completed it's run
// Note: If a job is added to a running scheduler and this method is then used
// you may see the job run overrun itself as job is scheduled immediately
// by default upon being added to the scheduler. It is recommended to use the
// SingletonMode() func on the scheduler chain when scheduling the job.
func (j *Job) SingletonMode() {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.runConfig.mode = singletonMode
	j.jobFunction.limiter = &singleflight.Group{}
}

// gocron/executor.go
switch f.runConfig.mode {
case defaultMode:
	runJob()
case singletonMode:
	// limiter是singlefilght對象，Do方法內(nèi)僅會執(zhí)行一次，保證一次只運行一個任務
	_, _, _ = f.limiter.Do("main", func() (interface{}, error) {
		select {
		case <-stopCtx.Done():
			return nil, nil
		case <-f.ctx.Done():
			return nil, nil
		default:
		}
		runJob()
		return nil, nil
	})
}

func NewScheduler(loc *time.Location) *Scheduler {
	//	這時已經(jīng)將executor同步初始化完畢
	// scheduler和executor是一對一的關(guān)系
	executor := newExecutor()

	return &Scheduler{
		jobs:       make([]*Job, 0),
		location:   loc,
		running:    false,
		time:       &trueTime{},
		executor:   &executor,
		tagsUnique: false,
		timer:      afterFunc,
	}
}

if s.updateJob || s.jobCreated {
	job = s.getCurrentJob()
}

s.setJobs(append(s.Jobs(), job))

if job.error != nil {
	// delete the job from the scheduler as this job
	// cannot be executed
	s.RemoveByReference(job)
	return nil, job.error
}
// 還有很多判斷條件，這里不一一列舉

// we should not schedule if not running since we can't foresee how long it will take for the scheduler to start
if s.IsRunning() {
	s.runContinuous(job)
}

// StartAsync starts all jobs without blocking the current thread
func (s *Scheduler) StartAsync() {
	if !s.IsRunning() {
		s.start()
	}
}

// StartBlocking starts all jobs and blocks the current thread.
// This blocking method can be stopped with Stop() from a separate goroutine.
func (s *Scheduler) StartBlocking() {
	s.StartAsync()
	s.startBlockingStopChanMutex.Lock()
	s.startBlockingStopChan = make(chan struct{}, 1)
	s.startBlockingStopChanMutex.Unlock()
	<-s.startBlockingStopChan
}

// start starts the scheduler, scheduling and running jobs
func (s *Scheduler) start() {
	// 啟動監(jiān)聽協(xié)程，select選擇器配合channel阻塞
	// 直到Job準備執(zhí)行發(fā)送通知
	go s.executor.start()
	// 將scheduler置位為running
	s.setRunning(true)
	// 遍歷所有任務，以遞歸的方式監(jiān)控起來
	s.runJobs(s.Jobs())
}

func (s *Scheduler) runJobs(jobs []*Job) {
	for _, job := range jobs {
		// 這個函數(shù)是一個遞歸調(diào)用
		// 這里對所有Job都以遞歸的方式監(jiān)聽著
		s.runContinuous(job)
	}
}

// 這是runContinuous的部分代碼
job.setTimer(s.timer(nextRun, func() {
	if !next.dateTime.IsZero() {
		for {
			n := s.now().UnixNano() - next.dateTime.UnixNano()
			// 某個任務滿足執(zhí)行條件了，退出循環(huán)
			if n >= 0 {
				break
			}
			s.time.Sleep(time.Duration(n))
		}
	}
	// 遞歸執(zhí)行本方法
	// runContinuous會判斷當前Job是否可執(zhí)行
	// 若不則退出，若可以則將Job設(shè)置為立即執(zhí)行，并刷新執(zhí)行時間
	// 若Job“立即執(zhí)行”的標志已經(jīng)置位，直接調(diào)用run發(fā)送通知給監(jiān)聽協(xié)程
	s.runContinuous(job)
}))

runJob := func() {
	f.incrementRunState()
	callJobFunc(f.eventListeners.onBeforeJobExecution)
	callJobFuncWithParams(f.function, f.parameters)
	callJobFunc(f.eventListeners.onAfterJobExecution)
	f.decrementRunState()
}