package tasker import ( "context" "time" "github.com/grassrootseconomics/cic-custodial/pkg/logg" "github.com/grassrootseconomics/cic-custodial/pkg/redis" "github.com/hibiken/asynq" "github.com/zerodha/logf" ) const ( retryRequeueInterval = 2 * time.Second ) type TaskerServerOpts struct { Concurrency int ErrorHandler asynq.ErrorHandler IsFailureHandler func(error) bool Logg logf.Logger LogLevel asynq.LogLevel RedisPool *redis.RedisPool RetryHandler asynq.RetryDelayFunc } type TaskerServer struct { mux *asynq.ServeMux server *asynq.Server } func NewTaskerServer(o TaskerServerOpts) *TaskerServer { server := asynq.NewServer( o.RedisPool, asynq.Config{ Concurrency: o.Concurrency, DelayedTaskCheckInterval: retryRequeueInterval, ErrorHandler: o.ErrorHandler, IsFailure: o.IsFailureHandler, LogLevel: o.LogLevel, Logger: logg.AsynqCompatibleLogger(o.Logg), Queues: map[string]int{ string(HighPriority): 7, string(DefaultPriority): 3, }, RetryDelayFunc: o.RetryHandler, }, ) mux := asynq.NewServeMux() return &TaskerServer{ mux: mux, server: server, } } func (ts *TaskerServer) RegisterMiddlewareStack(middlewareStack []asynq.MiddlewareFunc) { ts.mux.Use(middlewareStack...) } func (ts *TaskerServer) RegisterHandlers(taskName TaskName, taskHandler func(context.Context, *asynq.Task) error) { ts.mux.HandleFunc(string(taskName), taskHandler) } func (ts *TaskerServer) Start() error { if err := ts.server.Start(ts.mux); err != nil { return err } return nil } func (ts *TaskerServer) Stop() { ts.server.Stop() ts.server.Shutdown() }