68 lines
1.5 KiB
Go
68 lines
1.5 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/FernandoJVideira/simple_bank/internal/config"
|
|
db "github.com/FernandoJVideira/simple_bank/internal/db/sqlc"
|
|
"github.com/FernandoJVideira/simple_bank/internal/mailer"
|
|
"github.com/hibiken/asynq"
|
|
"github.com/redis/go-redis/v9"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
const (
|
|
CriticalQueue = "critical"
|
|
DefaultQueue = "default"
|
|
)
|
|
|
|
type TaskProcessor interface {
|
|
Start() error
|
|
Shutdown()
|
|
ProcessTaskSendVerifyEmail(ctx context.Context, task *asynq.Task) error
|
|
}
|
|
|
|
type RedisTaskProcessor struct {
|
|
server *asynq.Server
|
|
store db.Store
|
|
mailer mailer.MailSender
|
|
config config.Config
|
|
}
|
|
|
|
func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store, mailer mailer.MailSender, config config.Config) TaskProcessor {
|
|
logger := NewLogger()
|
|
redis.SetLogger(logger)
|
|
|
|
server := asynq.NewServer(
|
|
redisOpt,
|
|
asynq.Config{
|
|
Queues: map[string]int{
|
|
CriticalQueue: 10,
|
|
DefaultQueue: 5,
|
|
},
|
|
ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) {
|
|
log.Error().Err(err).Str("type", task.Type()).Bytes("payload", task.Payload()).Msg("process task failed")
|
|
}),
|
|
Logger: NewLogger(),
|
|
},
|
|
)
|
|
return &RedisTaskProcessor{
|
|
server: server,
|
|
store: store,
|
|
mailer: mailer,
|
|
config: config,
|
|
}
|
|
}
|
|
|
|
func (r *RedisTaskProcessor) Start() error {
|
|
mux := asynq.NewServeMux()
|
|
|
|
mux.HandleFunc(TaskSendVerifyEmail, r.ProcessTaskSendVerifyEmail)
|
|
|
|
return r.server.Start(mux)
|
|
}
|
|
|
|
func (r *RedisTaskProcessor) Shutdown() {
|
|
r.server.Shutdown()
|
|
}
|