12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970 |
- package service
- import (
- "context"
- "sync"
- "go-common/app/job/main/workflow/conf"
- "go-common/app/job/main/workflow/dao"
- "go-common/app/job/main/workflow/model"
- "go-common/library/sync/pipeline/fanout"
- )
- type Service struct {
- c *conf.Config
- dao *dao.Dao
- wg *sync.WaitGroup
- closeCh chan struct{}
- businessAttr []*model.BusinessAttr
-
- cache *fanout.Fanout
- }
- func New(c *conf.Config) (s *Service) {
- s = &Service{
- c: c,
- dao: dao.New(c),
- wg: &sync.WaitGroup{},
- closeCh: make(chan struct{}),
- cache: fanout.New("cache", fanout.Worker(1), fanout.Buffer(1024)),
- }
- var err error
- if s.businessAttr, err = s.dao.BusinessAttr(context.Background()); err != nil {
- panic(err)
- }
-
-
- go s.queueproc(context.Background(), _feedbackDealType)
- go s.taskExpireproc(context.Background(), _feedbackDealType)
- go s.repairQueueproc(context.Background(), _feedbackDealType)
-
- go s.notifyproc(context.Background())
-
- go s.singleExpireproc()
-
- go s.overallExpireproc()
-
- go s.releaseExpireproc()
-
- go s.refreshWeightproc()
-
- go s.enterPoolproc()
- return
- }
- func (s *Service) Ping(c context.Context) error {
- return s.dao.Ping(c)
- }
- func (s *Service) Close() (err error) {
- err = s.dao.Close()
- close(s.closeCh)
- s.wg.Wait()
- return
- }
|