Merge pull request #20 from Delavalom/ft/add-async-worker-logger

Ft/add async worker logger

worker/logger.go

@@ -0,0 +1,33 @@
+package worker
+
+import (
+	"fmt"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+type Logger struct{}
+
+func NewLogger() *Logger {
+	return &Logger{}
+}
+
+func (logger *Logger) Print(level zerolog.Level, args ...interface{}) {
+	log.WithLevel(level).Msg(fmt.Sprint(args...))
+}
+func (logger *Logger) Debug(args ...interface{}) {
+	logger.Print(zerolog.DebugLevel, args...)
+}
+func (logger *Logger) Info(args ...interface{}) {
+	logger.Print(zerolog.InfoLevel, args...)
+}
+func (logger *Logger) Warn(args ...interface{}) {
+	logger.Print(zerolog.WarnLevel, args...)
+}
+func (logger *Logger) Error(args ...interface{}) {
+	logger.Print(zerolog.ErrorLevel, args...)
+}
+func (logger *Logger) Fatal(args ...interface{}) {
+	logger.Print(zerolog.FatalLevel, args...)
+}

worker/processor.go

@@ -5,6 +5,7 @@ import (
 
 	db "github.com/Delavalom/RBD/db/sqlc"
 	"github.com/hibiken/asynq"
+	"github.com/rs/zerolog/log"
 )
 
 const (
@@ -22,12 +23,19 @@ type RedisTaskProcessor struct {
 }
 
 func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store) TaskProcessor {
-	server := asynq.NewServer(redisOpt, asynq.Config{
-		Queues: map[string]int{
-			QueueCritical: 10,
-			QueueDefault:  5,
-		},
-	})
+	server := asynq.NewServer(
+		redisOpt,
+		asynq.Config{
+			Queues: map[string]int{
+				QueueCritical: 10,
+				QueueDefault:  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,