redis_backend.go

// Copyright (c) 2019 Sick Yoon
// This file is part of gocelery which is released under MIT license.
// See file LICENSE for full license details.

package gocelery

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// RedisCeleryBackend is celery backend for redis
type RedisCeleryBackend struct {
	redis.UniversalClient
}

// NewRedisBackend creates new RedisCeleryBackend with given redis pool.
// RedisCeleryBackend can be initialized manually as well.
func NewRedisBackend(conn redis.UniversalClient) *RedisCeleryBackend {
	return &RedisCeleryBackend{
		UniversalClient: conn,
	}
}

// NewRedisCeleryBackend creates new RedisCeleryBackend
//
// Deprecated: NewRedisCeleryBackend exists for historical compatibility
// and should not be used. Pool should be initialized outside of gocelery package.
func NewRedisCeleryBackend(uri string) *RedisCeleryBackend {
	connectOpions, err := redis.ParseURL(uri)
	if err != nil {
		return nil
	}

	client := redis.NewClient(connectOpions)

	return &RedisCeleryBackend{
		UniversalClient: client,
	}
}

// GetResult queries redis backend to get asynchronous result
func (cb *RedisCeleryBackend) GetResult(ctx context.Context, taskID string) (*ResultMessage, error) {
	resp := cb.Get(ctx, fmt.Sprintf("celery-task-meta-%s", taskID))
	if resp.Err() != nil {
		return nil, resp.Err()
	}

	val, err := resp.Bytes()
	if err != nil {
		return nil, err
	}
	if val == nil {
		return nil, fmt.Errorf("result not available")
	}
	var resultMessage ResultMessage
	err = json.Unmarshal(val, &resultMessage)
	if err != nil {
		return nil, err
	}
	return &resultMessage, nil
}

// SetResult pushes result back into redis backend
func (cb *RedisCeleryBackend) SetResult(ctx context.Context, taskID string, result *ResultMessage) error {
	resBytes, err := json.Marshal(result)
	if err != nil {
		return err
	}

	resp := cb.SetEx(ctx, fmt.Sprintf("celery-task-meta-%s", taskID), resBytes, time.Second*86400)
	if resp.Err() != nil {
		return resp.Err()
	}
	return nil
}

// DeleteResult deletes the result from the redis backend
func (cb *RedisCeleryBackend) DeleteResult(ctx context.Context, taskID string) error {
	resp := cb.Del(ctx, fmt.Sprintf("celery-task-meta-%s", taskID))
	if resp.Err() != nil {
		return resp.Err()
	}
	return nil
}