bitmagnet/internal/queue/consumer/consumer.go

package consumer

import (
	"context"
	"encoding/json"
	"github.com/hibiken/asynq"
)

type Consumer interface {
	Pattern() string
	ProcessTask(ctx context.Context, t *asynq.Task) error
}

func New[T interface{}](pattern string, handler Handler[T]) Consumer {
	return &consumer[T]{
		pattern:     pattern,
		unmarshaler: NewJsonUnmarshaler[T](),
		handler:     handler,
	}
}

type Unmarshaler[T interface{}] interface {
	Unmarshal(t *asynq.Task) (T, error)
}

func NewJsonUnmarshaler[T interface{}]() Unmarshaler[T] {
	return &jsonUnmarshaler[T]{}
}

type jsonUnmarshaler[T interface{}] struct{}

func (u jsonUnmarshaler[T]) Unmarshal(t *asynq.Task) (T, error) {
	var payload T
	err := json.Unmarshal(t.Payload(), &payload)
	return payload, err
}

type Handler[T interface{}] interface {
	Handle(ctx context.Context, payload T) error
}

type HandlerFunc[T interface{}] func(ctx context.Context, payload T) error

func (f HandlerFunc[T]) Handle(ctx context.Context, payload T) error {
	return f(ctx, payload)
}

type consumer[T interface{}] struct {
	pattern     string
	unmarshaler Unmarshaler[T]
	handler     Handler[T]
}

func (c consumer[T]) Pattern() string {
	return c.pattern
}

func (c consumer[T]) ProcessTask(ctx context.Context, t *asynq.Task) error {
	payload, err := c.unmarshaler.Unmarshal(t)
	if err != nil {
		return err
	}
	return c.handler.Handle(ctx, payload)
}