func (p *manager) schedule() error {
	for {
		now := time.Now().Unix()
		if p.heap.Len() == 0 {
			return nil
		}
		if p.heap.Top().activeAt > now {
			return nil
		}

		summary := heap.Pop(&p.heap).(*ruleSummary)
		ruleConfig, ok := p.cache.Get(summary.id)
		if !ok {
			// drop it if not exist in cache
			delete(p.index, summary.id)
			continue
		}

		rule, ok := p.index[ruleConfig.Id]
		if !ok {
			// impossible
			log.Printf("manager.index[%d] not exists", ruleConfig.Id)
			continue
		}

		// update rule
		if err := rule.update(ruleConfig); err != nil {
			logger.Warningf("ruleEntity update err %s", err)
		}

		p.collectRuleCh <- rule

		logger.Debugf("%s %s %d lastAt %ds before nextAt %ds later",
			rule.CollectType, rule.Name, rule.Id,
			now-rule.lastAt, ruleConfig.Step)

		summary.activeAt = now + int64(ruleConfig.Step)
		rule.lastAt = now
		heap.Push(&p.heap, summary)

		continue
	}
}
// schedule return until there are no jobs
schedule return until there are no jobs
// schedule return until there are no jobs
func (p *manager) schedule() error {
	for {
		now := time.Now().Unix()
		if p.heap.Len() == 0 {
			return nil
		}
		if p.heap.Top().activeAt > now {
			return nil
		}
		summary := heap.Pop(&p.heap).(*ruleSummary)
		ruleConfig, ok := p.cache.Get(summary.id)
		if !ok {
			delete(p.index, summary.id)
			continue
		}
		rule, ok := p.index[ruleConfig.Id]
		if !ok {
			log.Printf("manager.index[%d] not exists", ruleConfig.Id)
			continue
		}
		if err := rule.update(ruleConfig); err != nil {
			logger.Warningf("ruleEntity update err %s", err)
		}
		p.collectRuleCh <- rule
		logger.Debugf("%s %s %d lastAt %ds before nextAt %ds later",
			rule.CollectType, rule.Name, rule.Id,
			now-rule.lastAt, ruleConfig.Step)
		summary.activeAt = now + int64(ruleConfig.Step)
		rule.lastAt = now
		heap.Push(&p.heap, summary)
		continue
	}
}
func ( p * manager ) schedule ( ) error { for { now := time . Now ( ) . Unix ( ) if p . heap . Len ( ) == 0 { return nil } if p . heap . Top ( ) . activeAt > now { return nil } summary := heap . Pop ( & p . heap ) . ( * ruleSummary ) ruleConfig , ok := p . cache . Get ( summary . id ) if ! ok { delete ( p . index , summary . id ) continue } rule , ok := p . index [ ruleConfig . Id ] if ! ok { log . Printf ( " " , ruleConfig . Id ) continue } if err := rule . update ( ruleConfig ) ; err != nil { logger . Warningf ( " " , err ) } p . collectRuleCh <- rule logger . Debugf ( " " , rule . CollectType , rule . Name , rule . Id , now - rule . lastAt , ruleConfig . Step ) summary . activeAt = now + int64 ( ruleConfig . Step ) rule . lastAt = now heap . Push ( & p . heap , summary ) continue } }
schedule return until there are no jobs
schedule return until there are no jobs
// drop it if not exist in cache // impossible // update rule